diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index de8f1e14591..80009709aeb 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -60,6 +60,11 @@ const MAX_MISSED_PONGS: u8 = 3; /// Auth timeout. const AUTH_TIMEOUT: Duration = Duration::from_secs(5); +/// How long an emptied huddle room stays open for a rejoin before it +/// auto-ends. Covers a laptop sleep/wake or a client-side reconnect without +/// ending the huddle for everyone who was about to come back. +const ROOM_EMPTY_GRACE: Duration = Duration::from_secs(20); + /// WebSocket upgrade handler for `/huddle/:channel_id/audio`. pub async fn ws_audio_handler( State(state): State>, @@ -290,6 +295,19 @@ async fn handle_active_audio_connection( } }; + // The relay, never the client, derives publisher authority. A managed-agent + // socket must present a valid NIP-OA owner attestation and hold bot-role + // membership in this exact huddle channel; every other socket remains an + // ordinary participant, including reconnecting humans. + let peer_role = derive_audio_peer_role( + &state, + tenant.community(), + channel_id, + &pubkey, + auth_tag_json.as_deref(), + ) + .await; + // Huddle cross-pod routing (mesh) OR single-pod guardrail. // // When the mesh is live (`state.mesh()` is `Some`), a huddle can span pods: @@ -467,8 +485,11 @@ async fn handle_active_audio_connection( owner_runtime_id, fenced, tenant.community(), - pubkey_hex.clone(), - requested_version, + crate::audio::join::RemotePeerRegistration { + pubkey: pubkey_hex.clone(), + protocol_version: requested_version, + role: peer_role.into(), + }, ) .await { @@ -510,22 +531,27 @@ async fn handle_active_audio_connection( } let admission = if let Some(session) = remote_session.as_ref() { - room.add_peer_at_index(pubkey_hex.clone(), requested_version, session.peer_index()) - .map(|(id, _mirror_epoch, audio, ctrl, revision)| { - // Report the owner-assigned epoch, not the local mirror's: - // the mirror never fans out via `broadcast_frame`, so its epoch - // is inert. The client's self-entry must match the owner roster. - ( - id, - session.peer_index(), - session.epoch(), - audio, - ctrl, - revision, - ) - }) + room.add_peer_at_index_with_role( + pubkey_hex.clone(), + requested_version, + session.peer_index(), + peer_role, + ) + .map(|(id, _mirror_epoch, audio, ctrl, revision)| { + // Report the owner-assigned epoch, not the local mirror's: + // the mirror never fans out via `broadcast_frame`, so its epoch + // is inert. The client's self-entry must match the owner roster. + ( + id, + session.peer_index(), + session.epoch(), + audio, + ctrl, + revision, + ) + }) } else { - room.add_peer(pubkey_hex.clone(), requested_version) + room.add_peer_with_role(pubkey_hex.clone(), requested_version, peer_role) }; let (peer_id, peer_index, peer_epoch, audio_rx, peer_ctrl_rx, admission_revision) = match admission { @@ -545,6 +571,21 @@ async fn handle_active_audio_connection( } return; } + Err(crate::audio::room::AdmissionError::DuplicateIdentity) => { + info!(channel_id = %channel_id, pubkey = %pubkey_hex, "audio identity already has a live publisher"); + let _ = ws_send.send(WsMessage::Text(serde_json::json!({"type":"error","code":"duplicate_identity","message":"this identity is already connected to the huddle"}).to_string().into())).await; + if let (Some(session), Some(stream)) = + (remote_session.as_ref(), remote_stream.as_mut()) + { + crate::audio::join::send_clean_close( + stream, + session.fenced(), + session.pubkey(), + ) + .await; + } + return; + } Err(crate::audio::room::AdmissionError::Ended) => { debug!(channel_id = %channel_id, "room ended before admission"); let _ = ws_send.send(WsMessage::Text(serde_json::json!({"type":"error","code":"room_ended","message":"huddle has ended"}).to_string().into())).await; @@ -645,7 +686,7 @@ async fn handle_active_audio_connection( .peers .iter() .map(|peer| { - serde_json::json!({"pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch}) + serde_json::json!({"pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch, "role": peer.role}) }) .collect(), session.roster().revision, @@ -657,7 +698,7 @@ async fn handle_active_audio_connection( .peers .into_iter() .map(|peer| { - serde_json::json!({"pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch}) + serde_json::json!({"pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch, "role": crate::audio::join::AudioPeerRoleWire::from(peer.role)}) }) .collect(), snapshot.revision, @@ -671,6 +712,7 @@ async fn handle_active_audio_connection( "pubkey": pubkey_hex, "peer_index": peer_index, "epoch": peer_epoch, + "role": crate::audio::join::AudioPeerRoleWire::from(peer_role), "peers": peers_snapshot, }) .to_string(); @@ -790,6 +832,10 @@ async fn handle_active_audio_connection( // the local generation floor so the fresh generation is accepted. The cause // distinction is carried on the remote control streams; locally the action // is the same WS teardown. Silent on ordinary client leave. + // Clones survive the teardown watcher's move so the leave path below can + // tell a drain-driven teardown from an ordinary leave. + let owner_lost_after_leave = owner_lost.clone(); + let owner_draining_after_leave = owner_draining.clone(); let owner_teardown_task = if owner_lost.is_some() || owner_draining.is_some() { let fence = Arc::clone( &state @@ -862,14 +908,15 @@ async fn handle_active_audio_connection( let _ = owner_teardown_task.await; } - // Atomic owner remove + end check: remove_peer_and_check_ended holds the - // AdmissionGuard lock across index recycling AND the is_empty + ended=true - // check. Ingress mirrors never archive authoritative huddle state; they + // Atomic owner remove + idle observation: remove_peer_and_check_idle holds + // the AdmissionGuard lock across index recycling AND the is_empty check, + // capturing the admission generation so a later end_if_idle cannot race a + // rejoin. Ingress mirrors never archive authoritative huddle state; they // remove locally and let the owner decide room lifetime. let removal = if remote_session.is_some() { - room.remove_peer(peer_id).map(|delta| (delta, false)) + room.remove_peer(peer_id).map(|delta| (delta, None)) } else { - room.remove_peer_and_check_ended(peer_id) + room.remove_peer_and_check_idle(peer_id) }; let removal_revision = if remote_session.is_none() { removal.as_ref().map(|(delta, _)| delta.revision) @@ -878,7 +925,7 @@ async fn handle_active_audio_connection( // ordering. Omit it rather than publishing a plausible-but-wrong value. None }; - let should_auto_end = removal.as_ref().map(|(_, ended)| *ended).unwrap_or(false); + let idle = removal.as_ref().and_then(|(_, idle)| *idle); if remote_session.is_none() { if let Some((delta, _)) = removal { @@ -887,6 +934,7 @@ async fn handle_active_audio_connection( "type": "left", "revision": delta.revision, "pubkey": left.pubkey, + "role": crate::audio::join::AudioPeerRoleWire::from(left.role), "peer_index": left.peer_index, "epoch": left.epoch, }) @@ -916,45 +964,51 @@ async fn handle_active_audio_connection( ) .await; - let room_emptied; - if should_auto_end { - info!(channel_id = %channel_id, "audio room empty — auto-ending huddle"); - - match state - .db - .archive_channel(tenant.community(), channel_id) - .await - { - Err(e) => { - warn!(channel_id = %channel_id, "auto-archive failed, huddle stays alive: {e}"); - room.clear_ended(); - room_emptied = false; - } - Ok(()) => { - room_emptied = state - .audio_rooms - .cleanup_if_empty(tenant.community(), channel_id); - - emit_participant_event( - &state, - &tenant, - channel_id, - parent_id_for_event, - ParticipantLifecycle { - kind: Kind::Custom(48103), - participant_pubkey: &pubkey_hex, - roster_revision: None, - admission_id: None, - }, - ) - .await; - } + // A relay drain (SIGTERM / owner-drain) tore every local client down at + // once. That empties the room without anyone choosing to leave, so it must + // never end the huddle: release the lease so rejoiners re-acquire through + // Redis, and leave the channel alive. Ordinary last-leaver departures get a + // grace window before the room ends so a reconnecting client keeps its + // huddle. + let draining = owner_draining_after_leave.is_some_and(|token| token.is_cancelled()) + || relay_is_draining(&state); + let room_emptied = match idle { + Some(idle) if !draining => { + info!( + channel_id = %channel_id, + grace_secs = ROOM_EMPTY_GRACE.as_secs(), + "audio room empty — holding huddle open for rejoin" + ); + tokio::spawn(end_room_after_grace( + Arc::clone(&state), + tenant.clone(), + channel_id, + parent_id_for_event, + pubkey_hex.clone(), + Arc::clone(&room), + idle, + owner_lost_after_leave, + owner_generation, + )); + // The grace task now owns room cleanup and lease release. + false } - } else { - room_emptied = state + Some(idle) => { + info!(channel_id = %channel_id, "audio room emptied by relay drain — huddle stays alive"); + room.release_idle_hold(idle); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id) + } + // Another peer was present under the same lock that removed this one + // (or this is an ingress mirror, which never observes idle). Cleanup is + // safe from any departure: a pending grace window pins the room in the + // manager, so a slow pre-last leaver cannot detach a room whose end is + // still in flight. + None => state .audio_rooms - .cleanup_if_empty(tenant.community(), channel_id); - } + .cleanup_if_empty(tenant.community(), channel_id), + }; // Owner path: release this room's lease when the room empties, so a new // owner can acquire and the renewer stops cleanly (silent, not owner-loss). @@ -963,9 +1017,7 @@ async fn handle_active_audio_connection( // is a no-op for the stale generation and leaves the live renewer running. // Only the last leaver empties the room, so exactly one release fires. if room_emptied { - if let (Some(mesh), Some(generation)) = (state.mesh(), owner_generation) { - mesh.owners.release(channel_id, generation); - } + release_owner_lease(&state, channel_id, owner_generation); } info!( @@ -975,6 +1027,134 @@ async fn handle_active_audio_connection( ); } +/// Whether this runtime is shutting down or draining its huddle ownership. +/// `shutting_down` flips on SIGTERM before the mesh watcher propagates it to +/// `owners.drain_all()`, and is the only signal in single-pod mode. +fn relay_is_draining(state: &AppState) -> bool { + state.shutting_down.load(Ordering::Relaxed) + || state.mesh().is_some_and(|mesh| mesh.owners.is_draining()) +} + +fn release_owner_lease(state: &AppState, channel_id: Uuid, owner_generation: Option) { + if let (Some(mesh), Some(generation)) = (state.mesh(), owner_generation) { + mesh.owners.release(channel_id, generation); + } +} + +/// Outcome of waiting out the empty-room grace window. +#[derive(Debug, PartialEq, Eq)] +enum IdleOutcome { + /// The grace window elapsed with no rejoin; the room is now `ended` and the + /// caller must archive + emit 48103. + Ended, + /// A peer was admitted during the window (whether or not it has since + /// left). That peer's own departure owns the next lifecycle decision. + Rejoined, + /// The lease was lost or the relay began draining before the window + /// elapsed. The room must not end; another owner may hold it now. + Aborted, +} + +/// Wait `grace`, then end the room if it is still idle. `owner_lost` aborts the +/// wait; `draining` is re-polled when the timer fires so a drain that began +/// mid-window (single-pod mode has no token) cannot end the huddle. +async fn await_room_idle( + room: &crate::audio::room::Room, + idle: crate::audio::room::IdleGeneration, + grace: Duration, + owner_lost: Option, + draining: impl Fn() -> bool, +) -> IdleOutcome { + let lost = async { + match owner_lost { + Some(token) => token.cancelled().await, + None => std::future::pending().await, + } + }; + tokio::select! { + // Owner loss wins a tie with the timer: another pod may own the room. + biased; + _ = lost => IdleOutcome::Aborted, + _ = tokio::time::sleep(grace) => { + if draining() { + IdleOutcome::Aborted + } else if room.end_if_idle(idle) { + IdleOutcome::Ended + } else { + IdleOutcome::Rejoined + } + } + } +} + +#[allow(clippy::too_many_arguments)] +async fn end_room_after_grace( + state: Arc, + tenant: TenantContext, + channel_id: Uuid, + parent_id_for_event: Uuid, + last_leaver_pubkey: String, + room: Arc, + idle: crate::audio::room::IdleGeneration, + owner_lost: Option, + owner_generation: Option, +) { + let outcome = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, owner_lost, || { + relay_is_draining(&state) + }) + .await; + let room_emptied = match outcome { + IdleOutcome::Rejoined => return, + IdleOutcome::Aborted => { + info!(channel_id = %channel_id, "audio room grace aborted — huddle stays alive"); + room.release_idle_hold(idle); + state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id) + } + IdleOutcome::Ended => { + info!(channel_id = %channel_id, "audio room stayed empty — auto-ending huddle"); + match state + .db + .archive_channel(tenant.community(), channel_id) + .await + { + Err(e) => { + warn!(channel_id = %channel_id, "auto-archive failed, huddle stays alive: {e}"); + room.clear_ended(); + room.release_idle_hold(idle); + return; + } + Ok(()) => { + // The hold outlived `end_if_idle` so no joiner could land + // on a replacement room while the archive was in flight. + room.release_idle_hold(idle); + let emptied = state + .audio_rooms + .cleanup_if_empty(tenant.community(), channel_id); + emit_participant_event( + &state, + &tenant, + channel_id, + parent_id_for_event, + ParticipantLifecycle { + kind: Kind::Custom(48103), + participant_pubkey: &last_leaver_pubkey, + roster_revision: None, + admission_id: None, + }, + ) + .await; + emptied + } + } + } + }; + if room_emptied { + release_owner_lease(&state, channel_id, owner_generation); + } +} + /// React to a non-owner huddle teardown signal read off the owner's control /// stream: cancel the connection (which drives the client's WS to close so it /// rejoins) and forget the local generation floor for this session. @@ -1013,6 +1193,10 @@ fn remote_rejection_ws_error(reason: &crate::audio::join::RegisterRejection) -> RegisterRejection::RoomEnded => serde_json::json!({ "type": "error", "code": "room_ended", "message": "huddle has ended" }), + RegisterRejection::DuplicateIdentity => serde_json::json!({ + "type": "error", "code": "duplicate_identity", + "message": "this identity is already connected to the huddle" + }), RegisterRejection::VersionMismatch { pinned, requested } => serde_json::json!({ "type": "error", "code": "upgrade_required", "message": format!( @@ -1257,6 +1441,40 @@ async fn heartbeat_loop( } } +async fn derive_audio_peer_role( + state: &AppState, + community: buzz_core::CommunityId, + channel_id: Uuid, + pubkey: &nostr::PublicKey, + auth_tag_json: Option<&str>, +) -> crate::audio::room::AudioPeerRole { + let has_valid_owner_attestation = + crate::api::relay_members::extract_nip_oa_owner(pubkey.as_bytes(), auth_tag_json).is_some(); + let membership_role = match state + .db + .get_member_role(community, channel_id, pubkey.as_bytes()) + .await + { + Ok(role) => role, + Err(error) => { + warn!(%channel_id, error = %error, "audio publisher role lookup failed closed"); + None + } + }; + classify_audio_peer_role(has_valid_owner_attestation, membership_role.as_deref()) +} + +fn classify_audio_peer_role( + has_valid_owner_attestation: bool, + membership_role: Option<&str>, +) -> crate::audio::room::AudioPeerRole { + if has_valid_owner_attestation && membership_role == Some("bot") { + crate::audio::room::AudioPeerRole::AgentTtsPublisher + } else { + crate::audio::room::AudioPeerRole::Participant + } +} + async fn ensure_membership( state: &AppState, tenant: &TenantContext, @@ -1483,6 +1701,30 @@ mod tests { use super::*; + #[test] + fn publisher_role_requires_both_nip_oa_and_bot_membership() { + use crate::audio::room::AudioPeerRole; + + assert_eq!( + classify_audio_peer_role(true, Some("bot")), + AudioPeerRole::AgentTtsPublisher + ); + assert_eq!( + classify_audio_peer_role(false, Some("bot")), + AudioPeerRole::Participant, + "a bot role alone cannot claim publisher authority" + ); + assert_eq!( + classify_audio_peer_role(true, Some("member")), + AudioPeerRole::Participant, + "a human with an auth-tag-shaped request remains an ordinary participant" + ); + assert_eq!( + classify_audio_peer_role(true, None), + AudioPeerRole::Participant + ); + } + #[test] fn audio_connection_permits_share_the_global_websocket_budget() { let semaphore = Arc::new(Semaphore::new(1)); @@ -1691,4 +1933,85 @@ mod tests { "oversized messages must be rejected by the WebSocket parser before the handler sees them" ); } + + fn idle_room() -> (crate::audio::room::Room, crate::audio::room::IdleGeneration) { + let room = crate::audio::room::Room::new( + buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()), + Uuid::new_v4(), + ); + let (peer, ..) = room.add_peer("alice".into(), 3).expect("admit"); + let (_, idle) = room.remove_peer_and_check_idle(peer).expect("peer existed"); + (room, idle.expect("last leaver observes idle")) + } + + /// The grace window holds the room open; once it elapses with no rejoin the + /// room ends and refuses further admission. + #[tokio::test(start_paused = true)] + async fn empty_room_ends_only_after_grace_elapses() { + let (room, idle) = idle_room(); + let wait = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, None, || false); + tokio::pin!(wait); + + tokio::time::advance(ROOM_EMPTY_GRACE - Duration::from_millis(1)).await; + assert!( + futures_util::poll!(&mut wait).is_pending(), + "room must stay open for the whole grace window" + ); + let (bob, ..) = room + .add_peer("bob".into(), 3) + .expect("rejoin is admitted during grace"); + room.remove_peer(bob); + + tokio::time::advance(Duration::from_millis(1)).await; + assert_eq!( + wait.await, + IdleOutcome::Rejoined, + "a rejoin during grace fences out the stale observation" + ); + assert!(room.add_peer("carol".into(), 3).is_ok(), "room never ended"); + } + + #[tokio::test(start_paused = true)] + async fn empty_room_with_no_rejoin_ends_after_grace() { + let (room, idle) = idle_room(); + let outcome = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, None, || false).await; + assert_eq!(outcome, IdleOutcome::Ended); + assert!(matches!( + room.add_peer("bob".into(), 3), + Err(crate::audio::room::AdmissionError::Ended) + )); + } + + /// Owner-loss during the window aborts without ending: the room now belongs + /// to whichever pod re-acquires the lease. + #[tokio::test(start_paused = true)] + async fn owner_loss_during_grace_aborts_without_ending() { + let (room, idle) = idle_room(); + let lost = CancellationToken::new(); + let wait = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, Some(lost.clone()), || false); + tokio::pin!(wait); + + tokio::time::advance(Duration::from_secs(1)).await; + lost.cancel(); + assert_eq!(wait.await, IdleOutcome::Aborted); + assert!(room.add_peer("bob".into(), 3).is_ok(), "room was not ended"); + } + + /// A drain that begins mid-window (single-pod mode has no drain token, only + /// `shutting_down`) must not end the huddle when the timer fires. + #[tokio::test(start_paused = true)] + async fn drain_during_grace_aborts_without_ending() { + let (room, idle) = idle_room(); + let draining = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let flag = Arc::clone(&draining); + let wait = await_room_idle(&room, idle, ROOM_EMPTY_GRACE, None, move || { + flag.load(Ordering::Relaxed) + }); + tokio::pin!(wait); + + tokio::time::advance(Duration::from_secs(5)).await; + draining.store(true, Ordering::Relaxed); + assert_eq!(wait.await, IdleOutcome::Aborted); + assert!(room.add_peer("bob".into(), 3).is_ok(), "room was not ended"); + } } diff --git a/crates/buzz-relay/src/audio/join.rs b/crates/buzz-relay/src/audio/join.rs index 96cc66b4e07..4d8cc7d6c48 100644 --- a/crates/buzz-relay/src/audio/join.rs +++ b/crates/buzz-relay/src/audio/join.rs @@ -52,7 +52,8 @@ use uuid::Uuid; use super::mesh::spawn_remote_peer_sink; use super::room::{ - AdmissionError, AudioRoomManager, Room, RosterDelta as RoomRosterDelta, RosterPeer, + AdmissionError, AudioPeerRole, AudioRoomManager, Room, RosterDelta as RoomRosterDelta, + RosterPeer, }; use crate::tunnel::directory::{ReleaseResult, RenewResult, SessionDirectory, SessionLease}; @@ -823,6 +824,8 @@ pub enum HuddleControlMsg { /// Huddle audio protocol version the client negotiated; the owner's /// room is pinned to one version and rejects mismatches. protocol_version: u8, + /// Relay-derived media role; ingress cannot upgrade client authority. + role: AudioPeerRoleWire, }, /// Owner → non-owner: the client is registered; here is its assigned index. PeerRegistered { @@ -878,11 +881,41 @@ pub enum HuddleControlMsg { }, } +/// Relay-derived media role carried over the trusted pod-to-pod control stream. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub enum AudioPeerRoleWire { + /// Ordinary audio participant. + #[default] + Participant, + /// NIP-OA-authenticated managed agent with bot membership. + AgentTtsPublisher, +} + +impl From for AudioPeerRoleWire { + fn from(role: AudioPeerRole) -> Self { + match role { + AudioPeerRole::Participant => Self::Participant, + AudioPeerRole::AgentTtsPublisher => Self::AgentTtsPublisher, + } + } +} + +impl From for AudioPeerRole { + fn from(role: AudioPeerRoleWire) -> Self { + match role { + AudioPeerRoleWire::Participant => Self::Participant, + AudioPeerRoleWire::AgentTtsPublisher => Self::AgentTtsPublisher, + } + } +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] /// One participant in the authoritative owner roster. pub struct RosterEntry { /// Nostr pubkey hex. pub pubkey: String, + /// Relay-authoritative media role. + pub role: AudioPeerRoleWire, /// Owner-assigned media routing index. pub peer_index: u8, /// Occupancy epoch for `peer_index`, bumped each time the index is reused @@ -903,6 +936,7 @@ impl From for RosterEntry { fn from(peer: RosterPeer) -> Self { Self { pubkey: peer.pubkey, + role: peer.role.into(), peer_index: peer.peer_index, epoch: peer.epoch, } @@ -918,6 +952,8 @@ pub enum RegisterRejection { RoomFull, /// Owner's room has ended (auto-ended or archived). RoomEnded, + /// The identity already has a live publisher in the owner's room. + DuplicateIdentity, /// Owner's room is pinned to a different protocol version. VersionMismatch { /// Version the owner's room is pinned to. @@ -1233,6 +1269,7 @@ impl HuddleControlAcceptor { community_id, pubkey, protocol_version, + role, } => { // Latch the community on first receipt; reject any later // frame that names a different one (tenant-boundary guard). @@ -1271,7 +1308,7 @@ impl HuddleControlAcceptor { fenced, from, &pubkey, - protocol_version, + (protocol_version, role), &mut registered, ), Err(e) => match FenceRejection::from_mesh_error(&e) { @@ -1368,10 +1405,11 @@ impl HuddleControlAcceptor { fenced: FencedHeader, from: RuntimeId, pubkey: &str, - protocol_version: u8, + registration: (u8, AudioPeerRoleWire), registered: &mut std::collections::HashMap, ) -> HuddleControlMsg { - match room.add_peer(pubkey.to_string(), protocol_version) { + let (protocol_version, role) = registration; + match room.add_peer_with_role(pubkey.to_string(), protocol_version, role.into()) { Ok((peer_id, peer_index, epoch, audio_rx, _peer_ctrl_rx, roster_revision)) => { registered.insert(pubkey.to_string(), peer_id); // The owner's Room fans out to this remote peer's `audio_tx`; @@ -1384,7 +1422,8 @@ impl HuddleControlAcceptor { "pubkey": pubkey, "peer_index": peer_index, "epoch": epoch, - "peers": [{"pubkey": pubkey, "peer_index": peer_index, "epoch": epoch}], + "role": role, + "peers": [{"pubkey": pubkey, "peer_index": peer_index, "epoch": epoch, "role": role}], }) .to_string(); room.broadcast_control(joined); @@ -1424,6 +1463,7 @@ fn peer_left_control(delta: RoomRosterDelta, session_id: Uuid) -> Option "type": "left", "revision": delta.revision, "pubkey": left.pubkey, + "role": AudioPeerRoleWire::from(left.role), "peer_index": left.peer_index, "epoch": left.epoch, }) @@ -1462,6 +1502,7 @@ fn admission_to_rejection(err: AdmissionError) -> RegisterRejection { match err { AdmissionError::Full => RegisterRejection::RoomFull, AdmissionError::Ended => RegisterRejection::RoomEnded, + AdmissionError::DuplicateIdentity => RegisterRejection::DuplicateIdentity, AdmissionError::VersionMismatch { pinned, requested } => { RegisterRejection::VersionMismatch { pinned, requested } } @@ -1574,6 +1615,7 @@ pub async fn read_owner_control( "type": "roster", "revision": revision, "peers": peers.into_iter().map(|p| serde_json::json!({ "pubkey": p.pubkey, "peer_index": p.peer_index, "epoch": p.epoch, + "role": p.role, })).collect::>() }) .to_string(); @@ -1595,12 +1637,17 @@ pub async fn read_owner_control( serde_json::json!({ "type": "joined", "revision": revision, "pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch, - "peers": [{"pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch}], + "role": peer.role, + "peers": [{ + "pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch, + "role": peer.role, + }], }) } else if let Some(peer) = left { serde_json::json!({ "type": "left", "revision": revision, "pubkey": peer.pubkey, "peer_index": peer.peer_index, "epoch": peer.epoch, + "role": peer.role, }) } else { continue; @@ -1686,6 +1733,16 @@ impl From for DialError { } } +/// Identity and admission policy carried from the external socket to the room owner. +pub struct RemotePeerRegistration { + /// Authenticated Nostr pubkey. + pub pubkey: String, + /// Negotiated Huddle audio protocol version. + pub protocol_version: u8, + /// Relay-derived media role from the authenticated ingress. + pub role: AudioPeerRoleWire, +} + /// Open a `HuddleControl` stream to the owner and register the local client. /// /// On success the owner has admitted the client as a remote peer and returned @@ -1698,8 +1755,7 @@ pub async fn dial_remote_owner( owner: RuntimeId, fenced: FencedHeader, community_id: CommunityId, - pubkey: String, - protocol_version: u8, + registration: RemotePeerRegistration, ) -> Result<(RemoteHuddleSession, MeshStream), DialError> { let hello = StreamHello { sender: local_runtime_id, @@ -1716,8 +1772,9 @@ pub async fn dial_remote_owner( fenced, payload: encode_control(&HuddleControlMsg::RegisterPeer { community_id: *community_id.as_uuid(), - pubkey: pubkey.clone(), - protocol_version, + pubkey: registration.pubkey.clone(), + protocol_version: registration.protocol_version, + role: registration.role, })?, }) .await?; @@ -1733,11 +1790,11 @@ pub async fn dial_remote_owner( RemoteHuddleSession { peer_index, epoch, - protocol_version, + protocol_version: registration.protocol_version, roster, fenced, owner, - pubkey, + pubkey: registration.pubkey, transport, seq: 0, }, @@ -2107,6 +2164,7 @@ mod tests { community_id: *community().as_uuid(), pubkey: "abc123".into(), protocol_version: 2, + role: AudioPeerRoleWire::Participant, }, HuddleControlMsg::PeerRegistered { pubkey: "abc123".into(), @@ -2116,6 +2174,7 @@ mod tests { revision: 1, peers: vec![RosterEntry { pubkey: "abc123".into(), + role: AudioPeerRoleWire::Participant, peer_index: 42, epoch: 0, }], @@ -2126,6 +2185,7 @@ mod tests { joined: None, left: Some(RosterEntry { pubkey: "abc123".into(), + role: AudioPeerRoleWire::Participant, peer_index: 42, epoch: 0, }), @@ -2193,6 +2253,64 @@ mod tests { (owner, client) } + #[tokio::test] + async fn ordered_roster_delta_preserves_agent_publisher_role() { + let session_id = Uuid::new_v4(); + let fenced = fenced_owned_by(rt(1), session_id); + let (mut owner, mut client) = stream_pair(); + let (ctrl_tx, mut ctrl_rx) = tokio::sync::mpsc::channel(4); + let reader = + tokio::spawn(async move { read_owner_control(&mut client, fenced, 1, &ctrl_tx).await }); + let publisher = RosterEntry { + pubkey: "agent".into(), + role: AudioPeerRoleWire::AgentTtsPublisher, + peer_index: 7, + epoch: 2, + }; + + owner + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control(&HuddleControlMsg::RosterDelta { + revision: 2, + joined: Some(publisher.clone()), + left: None, + }) + .unwrap(), + }) + .await + .unwrap(); + let joined = ctrl_rx.recv().await.expect("joined delta forwarded"); + let axum::extract::ws::Message::Text(joined) = joined else { + panic!("expected joined JSON"); + }; + let joined: serde_json::Value = serde_json::from_str(&joined).unwrap(); + assert_eq!(joined["role"], "AgentTtsPublisher"); + assert_eq!(joined["peers"][0]["role"], "AgentTtsPublisher"); + + owner + .send_frame(MeshStreamFrame::Data { + fenced, + payload: encode_control(&HuddleControlMsg::RosterDelta { + revision: 3, + joined: None, + left: Some(publisher), + }) + .unwrap(), + }) + .await + .unwrap(); + let left = ctrl_rx.recv().await.expect("left delta forwarded"); + let axum::extract::ws::Message::Text(left) = left else { + panic!("expected left JSON"); + }; + let left: serde_json::Value = serde_json::from_str(&left).unwrap(); + assert_eq!(left["role"], "AgentTtsPublisher"); + + drop(owner); + assert_eq!(reader.await.unwrap(), HuddleTeardownCause::StreamClosed); + } + #[tokio::test] async fn roster_revision_gap_requests_resync_before_forwarding_new_state() { let session_id = Uuid::new_v4(); @@ -2209,6 +2327,7 @@ mod tests { revision: 3, joined: Some(RosterEntry { pubkey: "bob".into(), + role: AudioPeerRoleWire::Participant, peer_index: 7, epoch: 0, }), @@ -2239,6 +2358,7 @@ mod tests { revision: 3, peers: vec![RosterEntry { pubkey: "bob".into(), + role: AudioPeerRoleWire::Participant, peer_index: 7, epoch: 0, }], @@ -2327,6 +2447,7 @@ mod tests { community_id: *community().as_uuid(), pubkey: "client-a".into(), protocol_version: 2, + role: AudioPeerRoleWire::Participant, }) .unwrap(), }) @@ -2380,6 +2501,7 @@ mod tests { community_id: *community().as_uuid(), pubkey: "remote".into(), protocol_version: 2, + role: AudioPeerRoleWire::Participant, }) .unwrap(), }) @@ -2443,6 +2565,7 @@ mod tests { community_id: *community().as_uuid(), pubkey: "client-a".into(), protocol_version: 2, + role: AudioPeerRoleWire::Participant, }) .unwrap(), }) @@ -2979,6 +3102,10 @@ mod tests { admission_to_rejection(AdmissionError::Ended), RegisterRejection::RoomEnded ); + assert_eq!( + admission_to_rejection(AdmissionError::DuplicateIdentity), + RegisterRejection::DuplicateIdentity + ); assert_eq!( admission_to_rejection(AdmissionError::VersionMismatch { pinned: 2, diff --git a/crates/buzz-relay/src/audio/room.rs b/crates/buzz-relay/src/audio/room.rs index d2849f3e0bd..ac327ea8da2 100644 --- a/crates/buzz-relay/src/audio/room.rs +++ b/crates/buzz-relay/src/audio/room.rs @@ -35,6 +35,8 @@ pub struct AudioPeer { /// same index. Prefixed onto protocol-v3 relayed frames alongside /// `peer_index`. pub epoch: u8, + /// Relay-authoritative media role for this socket. + pub role: AudioPeerRole, /// Pinned wire version used to shape outbound relay prefixes without /// taking the admission mutex on the per-frame audio hot path. pub protocol_version: u8, @@ -60,11 +62,23 @@ const CTRL_CHANNEL_CAPACITY: usize = 32; /// is reasonable. Routing identities rotate through a larger 255-value pool. const MAX_PEERS_PER_ROOM: usize = 25; +/// Relay-authoritative media role for one socket. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum AudioPeerRole { + /// Ordinary human audio participant. + #[default] + Participant, + /// Managed-agent speech publisher, proven by NIP-OA plus bot membership. + AgentTtsPublisher, +} + /// One authoritative owner-roster entry. #[derive(Clone, Debug, PartialEq, Eq)] pub struct RosterPeer { /// Nostr pubkey hex. pub pubkey: String, + /// Relay-authoritative media role for this socket. + pub role: AudioPeerRole, /// Owner-assigned media routing index. pub peer_index: u8, /// Per-index reuse generation for `peer_index` (see [`AudioPeer::epoch`]). @@ -117,6 +131,14 @@ pub type IndexedPeerAdmission = ( u64, ); +/// Snapshot of a room's admission history taken the moment it was observed +/// empty. Passed back to [`Room::end_if_idle`] after the grace window: if any +/// peer admitted in between, the snapshot is stale and the end is refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct IdleGeneration { + admissions: u64, +} + /// Reason a peer was refused entry to a room. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum AdmissionError { @@ -125,6 +147,9 @@ pub enum AdmissionError { /// The room has hit its participant cap or the requested routing identity /// is already active. Full, + /// Publisher sockets elect a single live writer for one managed-agent + /// identity. Ordinary participant sockets preserve the existing behavior. + DuplicateIdentity, /// The room is pinned to a different protocol version than the requested one. /// The caller should reply to the WS client with an `upgrade_required` error /// and the room's actual `pinned` version, then close the socket. @@ -179,6 +204,18 @@ struct AdmissionGuard { /// this behavior. pinned_version: Option, roster_revision: u64, + /// Count of successful admissions. A room-empty observation captures this + /// value; [`Room::end_if_idle`] only ends the room if no admission has + /// happened since, so a rejoin during the empty-room grace window fences + /// out the stale auto-end without any explicit timer cancellation. + admissions: u64, + /// The last peer left and an [`IdleGeneration`] was handed out: a grace + /// window may be pending against *this* `Room`. While set, the manager + /// must not drop the room — a rejoiner has to land on the same `Room` so + /// its admission fences the pending [`Room::end_if_idle`]. Stays set + /// through `end_if_idle` while the archive is in flight; cleared by + /// admission or by [`Room::release_idle_hold`]. + idle_hold: bool, } impl AdmissionGuard { @@ -190,6 +227,8 @@ impl AdmissionGuard { ended: false, pinned_version: None, roster_revision: 0, + admissions: 0, + idle_hold: false, } } @@ -288,6 +327,16 @@ impl Room { &self, pubkey: String, requested_version: u8, + ) -> Result { + self.add_peer_with_role(pubkey, requested_version, AudioPeerRole::Participant) + } + + /// Add a peer with a relay-derived media role. + pub fn add_peer_with_role( + &self, + pubkey: String, + requested_version: u8, + role: AudioPeerRole, ) -> Result { let mut g = self.guard.lock().map_err( |_| AdmissionError::Ended, /* poisoned ≈ shutting down */ @@ -298,6 +347,14 @@ impl Room { if self.peers.len() >= MAX_PEERS_PER_ROOM { return Err(AdmissionError::Full); } + if role == AudioPeerRole::AgentTtsPublisher + && self.peers.iter().any(|peer| { + peer.role == AudioPeerRole::AgentTtsPublisher + && peer.pubkey.eq_ignore_ascii_case(&pubkey) + }) + { + return Err(AdmissionError::DuplicateIdentity); + } if let Some(pinned) = g.pinned_version { if pinned != requested_version { return Err(AdmissionError::VersionMismatch { @@ -323,14 +380,18 @@ impl Room { peer_index, epoch, protocol_version: requested_version, + role, }, ); + g.admissions += 1; + g.idle_hold = false; g.roster_revision = g.roster_revision.wrapping_add(1); let revision = g.roster_revision; let delta = RosterDelta { revision, joined: Some(RosterPeer { pubkey, + role, peer_index, epoch, }), @@ -349,6 +410,22 @@ impl Room { pubkey: String, requested_version: u8, peer_index: u8, + ) -> Result { + self.add_peer_at_index_with_role( + pubkey, + requested_version, + peer_index, + AudioPeerRole::Participant, + ) + } + + /// Add an owner-assigned ingress peer with its relay-derived media role. + pub fn add_peer_at_index_with_role( + &self, + pubkey: String, + requested_version: u8, + peer_index: u8, + role: AudioPeerRole, ) -> Result { let mut g = self.guard.lock().map_err(|_| AdmissionError::Ended)?; if g.ended { @@ -357,6 +434,14 @@ impl Room { if self.peers.len() >= MAX_PEERS_PER_ROOM || g.active_indices.contains(&peer_index) { return Err(AdmissionError::Full); } + if role == AudioPeerRole::AgentTtsPublisher + && self.peers.iter().any(|peer| { + peer.role == AudioPeerRole::AgentTtsPublisher + && peer.pubkey.eq_ignore_ascii_case(&pubkey) + }) + { + return Err(AdmissionError::DuplicateIdentity); + } if let Some(pinned) = g.pinned_version { if pinned != requested_version { return Err(AdmissionError::VersionMismatch { @@ -384,14 +469,18 @@ impl Room { peer_index, epoch, protocol_version: requested_version, + role, }, ); + g.admissions += 1; + g.idle_hold = false; g.roster_revision = g.roster_revision.wrapping_add(1); let revision = g.roster_revision; let delta = RosterDelta { revision, joined: Some(RosterPeer { pubkey, + role, peer_index, epoch, }), @@ -416,6 +505,7 @@ impl Room { joined: None, left: Some(RosterPeer { pubkey: peer.pubkey, + role: peer.role, peer_index: peer.peer_index, epoch: peer.epoch, }), @@ -425,12 +515,16 @@ impl Room { Some(delta) } - /// Remove a peer AND atomically check if the room should end. - /// If the room is now empty, sets `ended = true` under the same lock - /// acquisition that removes the peer — no window for a concurrent - /// `add_peer` to sneak in between removal and the ended flag. - /// Returns `(roster_delta, should_auto_end)`. - pub fn remove_peer_and_check_ended(&self, peer_id: Uuid) -> Option<(RosterDelta, bool)> { + /// Remove a peer AND atomically observe whether it left the room empty. + /// Returns `(roster_delta, idle)` where `idle` is `Some` only for the + /// departure that emptied a not-yet-ended room. The [`IdleGeneration`] + /// captures the admission count under the same lock acquisition that + /// removed the peer, so the caller can wait out a grace window and then + /// call [`Self::end_if_idle`] without racing a concurrent `add_peer`. + pub fn remove_peer_and_check_idle( + &self, + peer_id: Uuid, + ) -> Option<(RosterDelta, Option)> { let mut g = self.guard.lock().ok()?; let (_, peer) = self.peers.remove(&peer_id)?; let peer_index = peer.peer_index; @@ -441,22 +535,65 @@ impl Room { joined: None, left: Some(RosterPeer { pubkey: peer.pubkey, + role: peer.role, peer_index, epoch: peer.epoch, }), }; - // Only the first task to see empty + !ended wins the auto-end. - // This prevents duplicate archive/48103 when two peers disconnect - // simultaneously and both see is_empty() == true. - let should_end = if !g.ended && self.peers.is_empty() { - g.ended = true; - true - } else { - false - }; + let idle = (!g.ended && self.peers.is_empty()).then_some(IdleGeneration { + admissions: g.admissions, + }); + g.idle_hold |= idle.is_some(); let _ = self.roster_tx.send(delta.clone()); drop(g); - Some((delta, should_end)) + Some((delta, idle)) + } + + /// End the room if it is still idle: empty, not already ended, and no + /// admission has happened since `idle` was observed. Sets `ended = true` + /// under the admission lock so no `add_peer` can sneak in after the check. + /// Returns `true` exactly once per idle generation — two leavers who both + /// observed the same empty room cannot both end it, and a rejoin (even one + /// that has since left again) fences out the stale observation. + /// + /// The idle hold is retained: the caller still has to archive the channel, + /// and until that resolves the room must stay registered so a concurrent + /// joiner meets `ended` on this `Room` instead of a fresh one whose + /// pre-join DB check can race the archive. Call + /// [`Self::release_idle_hold`] once the archive has succeeded, or + /// [`Self::clear_ended`] + `release_idle_hold` if it failed. + pub fn end_if_idle(&self, idle: IdleGeneration) -> bool { + let Ok(mut g) = self.guard.lock() else { + return false; + }; + if g.ended || g.admissions != idle.admissions || !self.peers.is_empty() { + return false; + } + g.ended = true; + true + } + + /// Release the hold taken by the idle observation `idle`, so the manager + /// may evict the room — either because the grace window was abandoned + /// (drain, owner loss) or because the end it guarded has fully resolved. + /// Fenced like [`Self::end_if_idle`]: a later admission owns its own hold, + /// which a stale release must not lift. + pub fn release_idle_hold(&self, idle: IdleGeneration) { + if let Ok(mut g) = self.guard.lock() { + if g.admissions == idle.admissions { + g.idle_hold = false; + } + } + } + + /// True when the room may be dropped from the manager: no peers, and no + /// grace window or in-flight end pending that a rejoiner would need to + /// land on. Both are read under the admission lock so a concurrent + /// `add_peer` cannot slip a peer in between the two checks. + fn is_evictable(&self) -> bool { + self.guard + .lock() + .is_ok_and(|g| !g.idle_hold && self.peers.is_empty()) } /// Fan-out a binary frame to all peers except the sender. Protocol v3 @@ -545,6 +682,7 @@ impl Room { .iter() .map(|e| RosterPeer { pubkey: e.pubkey.clone(), + role: e.role, peer_index: e.peer_index, epoch: e.epoch, }) @@ -621,10 +759,16 @@ impl AudioRoomManager { Some(room) } - /// Remove the room if it has no peers. Returns `true` if the room was removed. + /// Remove the room if it has no peers and no pending grace window. + /// Returns `true` if the room was removed. + /// + /// A room whose last peer just left stays registered until its grace + /// window resolves ([`Room::end_if_idle`] / [`Room::release_idle_hold`]): + /// evicting it early would hand a rejoiner a fresh `Room` that the pending + /// end cannot see, and the stale end would then archive the live huddle. pub fn cleanup_if_empty(&self, community_id: CommunityId, channel_id: Uuid) -> bool { self.rooms - .remove_if(&(community_id, channel_id), |_, room| room.is_empty()) + .remove_if(&(community_id, channel_id), |_, room| room.is_evictable()) .is_some() } } @@ -661,6 +805,63 @@ mod tests { ); } + #[test] + fn duplicate_identity_is_rejected_until_the_live_socket_leaves() { + let room = fresh_room(); + let (first, first_index, ..) = room + .add_peer_with_role("agent".into(), 2, AudioPeerRole::AgentTtsPublisher) + .expect("first renderer wins election"); + + assert!(matches!( + room.add_peer_with_role("AGENT".into(), 2, AudioPeerRole::AgentTtsPublisher), + Err(AdmissionError::DuplicateIdentity) + )); + assert_eq!(room.peer_pubkeys(), vec![("agent".into(), first_index)]); + + room.remove_peer(first).expect("winning renderer leaves"); + room.add_peer_with_role("agent".into(), 2, AudioPeerRole::AgentTtsPublisher) + .expect("a replacement renderer can claim the identity"); + } + + #[test] + fn ordinary_socket_cannot_claim_or_block_an_agent_publisher_seat() { + let room = fresh_room(); + room.add_peer("agent".into(), 2) + .expect("ordinary socket admits as a participant"); + room.add_peer_with_role("agent".into(), 2, AudioPeerRole::AgentTtsPublisher) + .expect("participant presence does not impersonate or block the publisher"); + assert!(matches!( + room.add_peer_with_role("agent".into(), 2, AudioPeerRole::AgentTtsPublisher,), + Err(AdmissionError::DuplicateIdentity) + )); + } + + #[test] + fn ordinary_participant_reconnect_is_not_blocked_by_identity_election() { + let room = fresh_room(); + room.add_peer("human".into(), 2) + .expect("first human socket admits"); + room.add_peer("HUMAN".into(), 2) + .expect("overlapping human reconnect remains supported"); + } + + #[test] + fn duplicate_identity_is_rejected_for_owner_assigned_ingress() { + let room = fresh_room(); + room.add_peer_with_role("agent".into(), 2, AudioPeerRole::AgentTtsPublisher) + .expect("owner-local renderer admits"); + + assert!(matches!( + room.add_peer_at_index_with_role( + "agent".into(), + 2, + 7, + AudioPeerRole::AgentTtsPublisher + ), + Err(AdmissionError::DuplicateIdentity) + )); + } + #[test] fn active_owner_assigned_index_cannot_be_readmitted() { let room = fresh_room(); @@ -712,6 +913,7 @@ mod tests { snapshot.peers, vec![RosterPeer { pubkey: "bob".into(), + role: AudioPeerRole::Participant, peer_index: bob_index, epoch: 0, }] @@ -787,11 +989,13 @@ mod tests { let (peer_id, _, _, _, _, _) = room1 .add_peer("alice".to_string(), 2) .expect("first peer admits"); - // Last peer leaves and ends the room atomically. - let (_, ended) = room1 - .remove_peer_and_check_ended(peer_id) + // Last peer leaves; the idle observation ends the room after grace. + let (_, idle) = room1 + .remove_peer_and_check_idle(peer_id) .expect("peer existed"); - assert!(ended, "single-peer room should end on its last departure"); + let idle = idle.expect("single-peer room should be idle on its last departure"); + assert!(room1.end_if_idle(idle), "idle room ends"); + room1.release_idle_hold(idle); assert!(manager.cleanup_if_empty(community_id, channel_id)); // Next joiner with a different version on the same channel id gets a @@ -993,4 +1197,195 @@ mod tests { // And the room state must be unchanged. assert_eq!(room.peers.len(), MAX_PEERS_PER_ROOM); } + + /// Only the departure that empties the room observes idleness; the + /// observation then ends the room exactly once. + #[test] + fn idle_observed_only_by_last_leaver_and_ends_once() { + let room = fresh_room(); + let (alice, ..) = room.add_peer("alice".into(), 2).unwrap(); + let (bob, ..) = room.add_peer("bob".into(), 2).unwrap(); + + let (_, idle) = room.remove_peer_and_check_idle(alice).unwrap(); + assert!(idle.is_none(), "room still has bob"); + let (_, idle) = room.remove_peer_and_check_idle(bob).unwrap(); + let idle = idle.expect("bob emptied the room"); + + assert!(room.end_if_idle(idle)); + assert!( + !room.end_if_idle(idle), + "second end on the same generation is a no-op" + ); + assert!(matches!( + room.add_peer("carol".into(), 2), + Err(AdmissionError::Ended) + )); + } + + /// A rejoin during the grace window fences out the stale idle observation, + /// even if the rejoiner has already left again — its own departure owns the + /// next lifecycle decision. + #[test] + fn rejoin_during_grace_fences_stale_idle() { + let room = fresh_room(); + let (alice, ..) = room.add_peer("alice".into(), 2).unwrap(); + let (_, idle) = room.remove_peer_and_check_idle(alice).unwrap(); + let stale = idle.unwrap(); + + let (alice_again, ..) = room + .add_peer("alice".into(), 2) + .expect("room is not ended during grace"); + assert!(!room.end_if_idle(stale), "occupied room never ends"); + + let (_, idle) = room.remove_peer_and_check_idle(alice_again).unwrap(); + let fresh = idle.expect("alice emptied the room again"); + assert!( + !room.end_if_idle(stale), + "stale generation cannot end a room a later admission touched" + ); + assert!(room.end_if_idle(fresh), "the fresh observation ends it"); + } + + /// Mari's concurrent-leaver ordering: Alice removes first (sees Bob, gets + /// no idle), Bob removes last (gets the idle generation), then Alice's + /// delayed `cleanup_if_empty` runs. The registry must keep the room pinned + /// so a rejoiner lands on the same `Room` and fences Bob's pending end; + /// otherwise the stale end would archive underneath a live replacement. + #[test] + fn pending_grace_pins_room_in_manager_across_concurrent_leavers() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let (bob, ..) = room.add_peer("bob".into(), 3).unwrap(); + + let (_, alice_idle) = room.remove_peer_and_check_idle(alice).unwrap(); + assert!(alice_idle.is_none(), "Bob is still present"); + let (_, bob_idle) = room.remove_peer_and_check_idle(bob).unwrap(); + let bob_idle = bob_idle.expect("Bob emptied the room"); + + // Alice's cleanup arrives late, after Bob's idle observation. + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "an empty room with a pending grace window must not be evicted" + ); + let rejoin_room = manager.get_or_create(community_id, channel_id); + assert!( + Arc::ptr_eq(&room, &rejoin_room), + "rejoin must land on the room the grace task holds" + ); + let (carol, ..) = rejoin_room.add_peer("carol".into(), 3).unwrap(); + assert!( + !room.end_if_idle(bob_idle), + "Carol's admission fences Bob's stale end" + ); + + // Carol leaves; her observation owns the lifecycle and ends the room. + let (_, carol_idle) = room.remove_peer_and_check_idle(carol).unwrap(); + let carol_idle = carol_idle.unwrap(); + assert!(room.end_if_idle(carol_idle)); + room.release_idle_hold(carol_idle); + assert!(manager.cleanup_if_empty(community_id, channel_id)); + assert!(manager.get(community_id, channel_id).is_none()); + } + + /// Mari's archive-in-flight ordering: the grace window expires and + /// `end_if_idle` succeeds, but the archive write has not resolved yet. A + /// delayed `cleanup_if_empty` must not detach the ended room, and a joiner + /// arriving in that gap must meet `ended` on the same `Room` rather than + /// admit into a fresh one whose pre-join DB check raced the archive. + #[test] + fn ended_room_stays_pinned_until_archive_resolves() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let idle = room.remove_peer_and_check_idle(alice).unwrap().1.unwrap(); + assert!(room.end_if_idle(idle)); + + // Archive is in flight: cleanup cannot detach, and a joiner lands on + // the ended room and is refused. + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "an ended room with its archive in flight must stay registered" + ); + let joiner_room = manager.get_or_create(community_id, channel_id); + assert!(Arc::ptr_eq(&room, &joiner_room)); + assert!(matches!( + joiner_room.add_peer("bob".into(), 3), + Err(AdmissionError::Ended) + )); + + // Archive succeeded: release, then the manager may evict. + room.release_idle_hold(idle); + assert!(manager.cleanup_if_empty(community_id, channel_id)); + assert!(manager.get(community_id, channel_id).is_none()); + } + + /// Archive failure rolls the end back: the room reopens, the hold is + /// released, and the next joiner gets a live room again. + #[test] + fn failed_archive_reopens_room_and_releases_hold() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let idle = room.remove_peer_and_check_idle(alice).unwrap().1.unwrap(); + assert!(room.end_if_idle(idle)); + + room.clear_ended(); + room.release_idle_hold(idle); + assert!( + manager.cleanup_if_empty(community_id, channel_id), + "a reopened, empty, unheld room is evictable" + ); + let next = manager.get_or_create(community_id, channel_id); + assert!(next.add_peer("bob".into(), 3).is_ok(), "fresh room admits"); + } + + /// Drain variant: the last leaver's idle observation is abandoned (the + /// huddle must outlive the pod), so it releases its hold and the room is + /// evicted. A pre-last teardown's cleanup racing ahead of that release + /// still cannot detach the room, and a stale release after a rejoin cannot + /// lift the rejoiner's hold. + #[test] + fn released_hold_allows_eviction_but_stale_release_does_not() { + let manager = AudioRoomManager::new(); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let channel_id = Uuid::new_v4(); + let room = manager.get_or_create(community_id, channel_id); + let (alice, ..) = room.add_peer("alice".into(), 3).unwrap(); + let (bob, ..) = room.add_peer("bob".into(), 3).unwrap(); + + assert!(room.remove_peer_and_check_idle(alice).unwrap().1.is_none()); + let drained = room.remove_peer_and_check_idle(bob).unwrap().1.unwrap(); + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "pre-last cleanup must wait for the drain-owned release" + ); + + // Rejoin before the release lands: the rejoiner owns a fresh hold. + let (carol, ..) = room.add_peer("carol".into(), 3).unwrap(); + let fresh = room.remove_peer_and_check_idle(carol).unwrap().1.unwrap(); + room.release_idle_hold(drained); + assert!( + !manager.cleanup_if_empty(community_id, channel_id), + "a stale release must not lift a later observation's hold" + ); + + room.release_idle_hold(fresh); + assert!( + manager.cleanup_if_empty(community_id, channel_id), + "released hold on an empty room permits eviction" + ); + let next = manager.get_or_create(community_id, channel_id); + assert!(!Arc::ptr_eq(&room, &next), "evicted room is replaced"); + assert!( + next.add_peer("dave".into(), 2).is_ok(), + "fresh room, no pin" + ); + } } diff --git a/desktop/src-tauri/src/huddle/agent_tts_publisher.rs b/desktop/src-tauri/src/huddle/agent_tts_publisher.rs index a1d42692666..387c17fe14b 100644 --- a/desktop/src-tauri/src/huddle/agent_tts_publisher.rs +++ b/desktop/src-tauri/src/huddle/agent_tts_publisher.rs @@ -5,14 +5,21 @@ use std::sync::Arc; use super::{relay_api, tts}; use crate::app_state::AppState; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum EnsureOutcome { + Ready, + NotLocal, + LostElection, +} + pub(super) async fn ensure( app: &tauri::AppHandle, state: &AppState, pipeline: &tts::TtsPipeline, speaker_pubkey: &str, -) -> Result { +) -> Result { if pipeline.has_audio_publisher(speaker_pubkey) { - return Ok(true); + return Ok(EnsureOutcome::Ready); } let app_for_load = app.clone(); @@ -28,7 +35,7 @@ pub(super) async fn ensure( .await .map_err(|error| format!("managed-agent identity task failed: {error}"))??; let Some(record) = record else { - return Ok(false); + return Ok(EnsureOutcome::NotLocal); }; let keys = nostr::Keys::parse(record.private_key_nsec.trim()) @@ -61,7 +68,7 @@ pub(super) async fn ensure( if !has_bot_membership { return Err("agent is not an active bot member of the Huddle".to_string()); } - let publisher = relay_api::connect_tts_audio_publisher( + let publisher = match relay_api::connect_tts_audio_publisher( &ephemeral_channel_id, parent_channel_id.as_deref(), state, @@ -69,7 +76,12 @@ pub(super) async fn ensure( record.auth_tag.as_deref(), local_tts_publishers, ) - .await?; + .await + { + Ok(publisher) => publisher, + Err(error) if error.is_lost_election() => return Ok(EnsureOutcome::LostElection), + Err(error) => return Err(error.to_string()), + }; pipeline.register_audio_publisher(speaker_pubkey, publisher); - Ok(true) + Ok(EnsureOutcome::Ready) } diff --git a/desktop/src-tauri/src/huddle/agent_tts_routing.rs b/desktop/src-tauri/src/huddle/agent_tts_routing.rs index 87a56c0dbbc..addb853110b 100644 --- a/desktop/src-tauri/src/huddle/agent_tts_routing.rs +++ b/desktop/src-tauri/src/huddle/agent_tts_routing.rs @@ -29,6 +29,15 @@ pub(super) fn classify_agent_tts_runtime( /// normal long-form huddle replies to play in full. pub(super) const MAX_TTS_TEXT_LEN: usize = 8_096; +pub(super) fn remote_agent_publisher_is_live<'a>( + speaker_pubkey: &str, + peer_pubkeys: impl IntoIterator, +) -> bool { + peer_pubkeys + .into_iter() + .any(|pubkey| pubkey.eq_ignore_ascii_case(speaker_pubkey)) +} + pub(super) fn normalize_agent_tts_text(text: String) -> String { if text.chars().count() > MAX_TTS_TEXT_LEN { let mut truncated: String = text.chars().take(MAX_TTS_TEXT_LEN).collect(); diff --git a/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs b/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs index c9ebabe6b62..737cf5aa203 100644 --- a/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs +++ b/desktop/src-tauri/src/huddle/agent_tts_routing_tests.rs @@ -1,6 +1,6 @@ use super::{ classify_agent_tts_runtime, enqueue_agent_tts_text, normalize_agent_tts_text, - AgentTtsRuntimeGate, MAX_TTS_TEXT_LEN, + remote_agent_publisher_is_live, AgentTtsRuntimeGate, MAX_TTS_TEXT_LEN, }; use crate::huddle::HuddlePhase; @@ -45,6 +45,17 @@ fn disabled_is_the_only_intentional_runtime_no_op() { ); } +#[test] +fn live_remote_agent_publisher_suppresses_only_that_agents_local_fallback() { + let peers = ["human", "AGENT-A", "agent-b"]; + assert!(remote_agent_publisher_is_live("agent-a", peers)); + assert!(!remote_agent_publisher_is_live("agent-c", peers)); + assert!(!remote_agent_publisher_is_live( + "agent-a", + std::iter::empty::<&str>(), + )); +} + #[test] fn assistant_text_truncation_is_unicode_safe_before_voice_routing() { assert_eq!(MAX_TTS_TEXT_LEN, 8_096); diff --git a/desktop/src-tauri/src/huddle/audio_output.rs b/desktop/src-tauri/src/huddle/audio_output.rs index 383a7e8210a..6335100dd41 100644 --- a/desktop/src-tauri/src/huddle/audio_output.rs +++ b/desktop/src-tauri/src/huddle/audio_output.rs @@ -35,7 +35,7 @@ fn list_audio_output_devices_blocking() -> Result, String } /// Set the preferred audio output device by name. Empty string = system default. -/// Takes effect on the next huddle start/join (does not change a live stream). +/// An active huddle moves subsequent playout to the selected route immediately. #[tauri::command] pub fn set_audio_output_device(name: String, state: State<'_, AppState>) -> Result<(), String> { let mut guard = state @@ -43,7 +43,12 @@ pub fn set_audio_output_device(name: String, state: State<'_, AppState>) -> Resu .output_device .lock() .map_err(|e| e.to_string())?; - *guard = if name.is_empty() { None } else { Some(name) }; + let selected = if name.is_empty() { None } else { Some(name) }; + *guard = selected.clone(); + state + .huddle_audio + .output_device_changes + .send_replace(selected); Ok(()) } diff --git a/desktop/src-tauri/src/huddle/latency_bench.rs b/desktop/src-tauri/src/huddle/latency_bench.rs index f928ddbce0f..2f7787df65c 100644 --- a/desktop/src-tauri/src/huddle/latency_bench.rs +++ b/desktop/src-tauri/src/huddle/latency_bench.rs @@ -143,8 +143,8 @@ fn baseline_stt_fake_llm_tts_first_audio() { Arc::clone(&tts_cancel), super::human_floor::HumanFloor::new(), "eve", - None, // default output device - None, // no Tauri app handle + tokio::sync::watch::channel(None).1, // default output device + None, // no Tauri app handle ) .expect("tts pipeline"); eprintln!( @@ -158,7 +158,7 @@ fn baseline_stt_fake_llm_tts_first_audio() { None, None, super::human_floor::HumanFloor::new(), - None, + tokio::sync::watch::channel(None).1, ) .expect("stt pipeline"); // Recognizer loads inside the worker thread; give it time, then verify diff --git a/desktop/src-tauri/src/huddle/mod.rs b/desktop/src-tauri/src/huddle/mod.rs index e219b2f75fa..c0bb11839cc 100644 --- a/desktop/src-tauri/src/huddle/mod.rs +++ b/desktop/src-tauri/src/huddle/mod.rs @@ -94,7 +94,7 @@ use crate::{app_state::AppState, events, relay::submit_event}; use agent_tts_routing::{ classify_agent_tts_runtime, enqueue_agent_tts_text, normalize_agent_tts_text, - AgentTtsRuntimeGate, + remote_agent_publisher_is_live, AgentTtsRuntimeGate, }; pub use pipeline::check_pipeline_hotstart; use pipeline::{ @@ -499,8 +499,8 @@ fn teardown_huddle(state: &AppState) -> Result<(), String> { let cancel = hs.audio_ws_cancel.take(); // Cancel the relay token BEFORE dropping the sender. If we drop // pcm_tx first, the send task sees None from recv() and can exit - // the pipeline before is_cancelled() is true — causing a spurious - // huddle-audio-disconnected event on intentional teardown. + // the pipeline before is_cancelled() is true — and would start a + // spurious audio reconnect on intentional teardown. if let Some(ref c) = cancel { c.cancel(); } @@ -900,16 +900,52 @@ pub async fn speak_agent_message( ); return Err("Agent text to speech is enabled but its audio pipeline is unavailable".into()); }; - match agent_tts_publisher::ensure(&app, &state, &pipeline, &speaker_pubkey).await { - Ok(true) => eprintln!( - "buzz-desktop: tts broadcast status=ready route_id={route_id}" - ), - Ok(false) => eprintln!( - "buzz-desktop: tts broadcast status=unavailable reason=agent_identity_not_local route_id={route_id}" - ), - Err(error) => eprintln!( - "buzz-desktop: tts broadcast status=unavailable reason=publisher_setup_failed route_id={route_id} error={error}" - ), + let owns_agent_publisher = match agent_tts_publisher::ensure( + &app, + &state, + &pipeline, + &speaker_pubkey, + ) + .await + { + Ok(agent_tts_publisher::EnsureOutcome::Ready) => { + eprintln!("buzz-desktop: tts broadcast status=ready route_id={route_id}"); + true + } + Ok(agent_tts_publisher::EnsureOutcome::NotLocal) => { + eprintln!( + "buzz-desktop: tts broadcast status=unavailable reason=agent_identity_not_local route_id={route_id}" + ); + false + } + Ok(agent_tts_publisher::EnsureOutcome::LostElection) => { + eprintln!( + "buzz-desktop: tts stage=queue status=dropped reason=publisher_election_lost route_id={route_id}" + ); + return Ok(()); + } + Err(error) => { + eprintln!( + "buzz-desktop: tts broadcast status=unavailable reason=publisher_setup_failed route_id={route_id} error={error}" + ); + false + } + }; + if !owns_agent_publisher { + let remote_publisher_is_live = { + let hs = state.huddle()?; + let peers = hs + .audio_peer_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()); + remote_agent_publisher_is_live(&speaker_pubkey, peers.values().map(String::as_str)) + }; + if remote_publisher_is_live { + eprintln!( + "buzz-desktop: tts stage=queue status=dropped reason=remote_agent_publisher_live route_id={route_id}" + ); + return Ok(()); + } } let sender = pipeline.text_sender(); let speaker_generation = sender.speaker_generation(&speaker_pubkey); diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index 47d4aeb43d1..4254bbc729c 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -250,7 +250,7 @@ pub(crate) async fn post_connect_setup( } return Ok(PostConnectOutcome::Stale); } - let (cancel, pcm_tx) = audio_result?; + let (cancel, pcm_tx) = audio_result.map_err(|error| error.to_string())?; hs.audio_ws_cancel = Some(cancel); hs.audio_relay_pcm_tx = Some(pcm_tx); } @@ -349,12 +349,7 @@ pub(crate) async fn maybe_start_stt_pipeline( ptt, manual_mic_unmuted, hs.human_floor.clone(), - state - .huddle_audio - .output_device - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone(), + state.huddle_audio.output_device_changes.subscribe(), old, ) }; @@ -447,12 +442,7 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result Result bool { + peer["role"].as_str() == Some("AgentTtsPublisher") +} + +fn update_publisher_roster( + publishers: &mut std::collections::HashMap, + peer_index: u8, + pubkey: &str, + is_publisher: bool, +) { + if is_publisher { + publishers.insert(peer_index, pubkey.to_string()); + } else { + publishers.remove(&peer_index); + } +} + #[allow(clippy::too_many_arguments)] pub(crate) async fn run_playout_recv_loop( mut ws_rx: futures_util::stream::SplitStream, ws_tx_for_pongs: Arc>>, - sink_handle: rodio::MixerDeviceSink, + mut sink_handle: rodio::MixerDeviceSink, cancel: CancellationToken, app_handle: Option, - initial_peers: Vec<(u8, String, u8)>, + initial_peers: Vec<(u8, String, u8, bool)>, tts_active: Arc, tts_cancel: Arc, local_tts_publishers: super::tts::LocalTtsPublishers, + audio_peer_pubkeys: Arc>>, remote_stt_pipeline: Arc>>>, agent_pubkeys: Arc>>, human_floor: HumanFloor, + mut output_device_changes: tokio::sync::watch::Receiver>, ) { use rodio::buffer::SamplesBuffer; use std::num::NonZero; @@ -298,10 +317,21 @@ pub(crate) async fn run_playout_recv_loop( // departed occupant that arrives after its index is reassigned carries the // old epoch and is fenced rather than mis-attributed to the new occupant. let mut index_to_epoch: std::collections::HashMap = std::collections::HashMap::new(); - for (idx, pubkey, epoch) in initial_peers { + let mut publisher_indices = std::collections::HashSet::new(); + for (idx, pubkey, epoch, is_agent_tts_publisher) in initial_peers { index_to_pubkey.insert(idx, pubkey); index_to_epoch.insert(idx, epoch); + if is_agent_tts_publisher { + publisher_indices.insert(idx); + } } + *audio_peer_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()) = index_to_pubkey + .iter() + .filter(|(idx, _)| publisher_indices.contains(idx)) + .map(|(idx, pubkey)| (*idx, pubkey.clone())) + .collect(); let mut active_indices: std::collections::HashSet = std::collections::HashSet::new(); let mut speaker_levels: std::collections::HashMap = std::collections::HashMap::new(); let mut remote_release_deadlines: std::collections::HashMap = @@ -328,6 +358,25 @@ pub(crate) async fn run_playout_recv_loop( tokio::select! { biased; _ = cancel.cancelled() => break, + changed = output_device_changes.changed() => { + if changed.is_err() { + break; + } + let selected = output_device_changes.borrow_and_update().clone(); + match super::audio_output::open_output_sink_by_name(selected.as_deref()) { + Ok(next_sink) => { + sink_handle = next_sink; + let mixer = sink_handle.mixer().clone(); + for slot in peers.values_mut() { + slot.player = rodio::Player::connect_new(&mixer); + slot.recovering_playout = false; + } + } + Err(error) => { + eprintln!("buzz-desktop: live output device switch failed: {error}"); + } + } + } _ = playout_tick.tick() => { // Drain one 10 ms frame from each *active* peer's NetEq into // its Player. NetEq always emits a frame (Expand/silence when @@ -571,6 +620,21 @@ pub(crate) async fn run_playout_recv_loop( } index_to_pubkey.insert(key, pk.to_string()); index_to_epoch.insert(key, epoch); + let is_publisher = + roster_peer_is_agent_tts_publisher(p); + if is_publisher { + publisher_indices.insert(key); + } else { + publisher_indices.remove(&key); + } + update_publisher_roster( + &mut audio_peer_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()), + key, + pk, + is_publisher, + ); } } } @@ -580,6 +644,8 @@ pub(crate) async fn run_playout_recv_loop( let mut replacement = std::collections::HashMap::new(); let mut replacement_epochs = std::collections::HashMap::new(); + let mut replacement_publishers = + std::collections::HashSet::new(); for p in peer_list { if let (Some(pk), Some(idx)) = ( p["pubkey"].as_str(), @@ -590,6 +656,9 @@ pub(crate) async fn run_playout_recv_loop( p["epoch"].as_u64().unwrap_or(0) as u8; replacement.insert(key, pk.to_string()); replacement_epochs.insert(key, epoch); + if roster_peer_is_agent_tts_publisher(p) { + replacement_publishers.insert(key); + } } } let identity_unchanged = |idx: &u8| { @@ -621,6 +690,15 @@ pub(crate) async fn run_playout_recv_loop( speaker_levels.retain(|idx, _| identity_unchanged(idx)); index_to_pubkey = replacement; index_to_epoch = replacement_epochs; + publisher_indices = replacement_publishers; + *audio_peer_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()) = + index_to_pubkey + .iter() + .filter(|(idx, _)| publisher_indices.contains(idx)) + .map(|(idx, pubkey)| (*idx, pubkey.clone())) + .collect(); } } Some("left") => { @@ -628,6 +706,11 @@ pub(crate) async fn run_playout_recv_loop( let key = idx as u8; index_to_pubkey.remove(&key); index_to_epoch.remove(&key); + publisher_indices.remove(&key); + audio_peer_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()) + .remove(&key); frame_counts.remove(&key); remote_release_deadlines.remove(&key); remote_floor_owners.remove(&key); @@ -656,6 +739,10 @@ pub(crate) async fn run_playout_recv_loop( } human_floor.clear_remote(); + audio_peer_pubkeys + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clear(); if let Some(ref app) = app_handle { use tauri::Emitter; let _ = app.emit( @@ -669,6 +756,37 @@ pub(crate) async fn run_playout_recv_loop( mod tests { use super::*; + #[test] + fn publisher_roster_uses_only_authoritative_role_and_tracks_transitions() { + let participant = serde_json::json!({ + "pubkey": "agent", + "peer_index": 4, + "role": "Participant", + }); + let publisher = serde_json::json!({ + "pubkey": "agent", + "peer_index": 4, + "role": "AgentTtsPublisher", + }); + let legacy_peer = serde_json::json!({"pubkey": "agent", "peer_index": 4}); + + assert!(!roster_peer_is_agent_tts_publisher(&participant)); + assert!(roster_peer_is_agent_tts_publisher(&publisher)); + assert!( + !roster_peer_is_agent_tts_publisher(&legacy_peer), + "pubkey presence without an authoritative role must not suppress local TTS" + ); + + let mut publishers = std::collections::HashMap::new(); + update_publisher_roster(&mut publishers, 4, "agent", true); + assert_eq!(publishers.get(&4).map(String::as_str), Some("agent")); + update_publisher_roster(&mut publishers, 4, "human", false); + assert!( + publishers.is_empty(), + "participant replacement clears the seat" + ); + } + #[test] fn continuous_silence_does_not_extend_remote_floor_deadline() { let peer = 7; diff --git a/desktop/src-tauri/src/huddle/reconnect.rs b/desktop/src-tauri/src/huddle/reconnect.rs index 6996776f1c9..7639038e677 100644 --- a/desktop/src-tauri/src/huddle/reconnect.rs +++ b/desktop/src-tauri/src/huddle/reconnect.rs @@ -1,52 +1,204 @@ -//! Audio-only huddle reconnection after an unexpected relay disconnect. +//! Rust-owned audio reconnect after an unexpected relay disconnect. +//! +//! Only the audio relay WebSocket is rebuilt. Huddle membership, mic capture, +//! STT/TTS, and agent voice stay live because `phase` never leaves `Active`; +//! the renderer sees progress through `HuddleState::audio_link` instead. +//! +//! The loop is fenced by huddle identity (`is_current_huddle`) after every +//! await: an intentional leave, or a replacement huddle started during the +//! backoff, always wins and any socket opened for the old huddle is cancelled +//! rather than installed. -use std::sync::atomic::Ordering; +use std::future::Future; +use std::pin::Pin; +use std::time::Duration; -use tauri::State; +use tokio::time::Instant; +use tokio_util::sync::CancellationToken; use crate::app_state::AppState; -use super::{relay_api, HuddlePhase}; +use super::relay_api::{self, AudioRelayConnectError}; +use super::state::{AudioLink, HuddleState}; -/// Re-establish only the audio relay WebSocket after an unexpected owner/pod -/// disconnect. Huddle membership, mic capture, STT/TTS, and frontend state stay -/// live, so a successful reconnect is a short audio blip rather than a leave. +/// Redials continue until at least this long after the disconnect. Long +/// enough to ride out a relay deploy (pod drain + Service endpoint +/// convergence) and a laptop Wi-Fi roam without dropping the huddle. +pub(crate) const RECONNECT_WINDOW: Duration = Duration::from_secs(60); +/// First backoff step after a failed dial; doubles per failure up to the ceiling. +const BACKOFF_BASE: Duration = Duration::from_millis(250); +const BACKOFF_MAX: Duration = Duration::from_secs(5); +/// A `huddle_relay_draining` refusal means a replacement pod is coming up, so +/// redial at the floor instead of growing the backoff. +const DRAINING_REDIAL_DELAY: Duration = BACKOFF_BASE; + +/// Identity of the huddle whose audio socket is being rebuilt. +#[derive(Debug, Clone)] +pub(crate) struct ReconnectTarget { + pub ephemeral_channel_id: String, + pub parent_channel_id: Option, + huddle_generation: u64, +} + +pub(crate) type AudioConnection = (CancellationToken, tokio::sync::mpsc::Sender>); + +/// Entry point for the audio pipeline task when its socket exits unexpectedly. +/// Awaited in place by that task, so the loop ends with it; nothing is spawned. /// -/// The session generation and channel id are re-checked after the network dial: -/// an intentional leave/end racing this command wins and the newly-opened audio -/// pipeline is cancelled instead of resurrecting a terminal huddle. -#[tauri::command] -pub async fn reconnect_huddle_audio(state: State<'_, AppState>) -> Result<(), String> { - let (ephemeral_channel_id, parent_channel_id, session_generation) = { - let hs = state.huddle()?; - if matches!(hs.phase, HuddlePhase::Idle | HuddlePhase::Leaving) { - return Err("huddle is no longer active".into()); - } - ( - hs.ephemeral_channel_id - .clone() - .ok_or("active huddle has no channel id")?, - hs.parent_channel_id.clone(), - hs.session_generation.load(Ordering::Acquire), - ) - }; +/// Boxed because the call graph is recursive (pipeline task → reconnect → +/// `connect_audio_relay` → pipeline task) and the compiler needs a type-erased +/// edge to prove the future is `Send`. +pub(crate) fn after_unexpected_disconnect( + app: tauri::AppHandle, +) -> Pin + Send>> { + Box::pin(async move { + use tauri::Manager; + let state = app.state::(); + let state: &AppState = &state; + run(state, |target: ReconnectTarget| async move { + relay_api::connect_audio_relay( + &target.ephemeral_channel_id, + target.parent_channel_id.as_deref(), + state, + ) + .await + }) + .await; + }) +} - let (cancel, pcm_tx) = - relay_api::connect_audio_relay(&ephemeral_channel_id, parent_channel_id.as_deref(), &state) - .await?; - - let mut hs = state.huddle()?; - let still_current = !matches!(hs.phase, HuddlePhase::Idle | HuddlePhase::Leaving) - && hs.ephemeral_channel_id.as_deref() == Some(ephemeral_channel_id.as_str()) - && hs.session_generation.load(Ordering::Acquire) == session_generation; - if !still_current { - cancel.cancel(); - return Err("huddle ended while audio was reconnecting".into()); +/// Mark the link as reconnecting and capture the target, or `None` when the +/// huddle is not live or a reconnect is already owned by another pipeline +/// exit. This is the single guard against duplicate disconnect signals. +fn claim_reconnect(hs: &mut HuddleState) -> Option { + if hs.audio_link != AudioLink::Live { + return None; + } + let ephemeral_channel_id = hs.ephemeral_channel_id.clone()?; + if !hs.is_current_huddle(&ephemeral_channel_id, hs.huddle_generation) { + return None; } + hs.audio_link = AudioLink::Reconnecting { + attempt: 0, + draining: false, + }; + // The dead pipeline's sender would only buffer frames nobody encodes. + hs.audio_relay_pcm_tx = None; + Some(ReconnectTarget { + ephemeral_channel_id, + parent_channel_id: hs.parent_channel_id.clone(), + huddle_generation: hs.huddle_generation, + }) +} - if let Some(old_cancel) = hs.audio_ws_cancel.replace(cancel) { - old_cancel.cancel(); +/// Drive redials with `dial` until one succeeds, the huddle ends or is +/// replaced, or the retry window is exhausted (`AudioLink::Lost`). +pub(crate) async fn run(state: &AppState, mut dial: F) +where + F: FnMut(ReconnectTarget) -> Fut, + Fut: Future>, +{ + let Some(target) = state + .huddle() + .ok() + .and_then(|mut hs| claim_reconnect(&mut hs)) + else { + return; + }; + state.emit_huddle_state_changed(); + let deadline = Instant::now() + RECONNECT_WINDOW; + let mut failures: u32 = 0; + + loop { + let result = dial(target.clone()).await; + + // Lock scope is a block, not `drop()`: the guard must provably end + // before the sleep for the future to stay `Send`. + let next_delay = { + let Ok(mut hs) = state.huddle() else { return }; + if !hs.is_current_huddle(&target.ephemeral_channel_id, target.huddle_generation) { + // Leave or replacement won while we were dialing; never resurrect. + if let Ok((cancel, _)) = result { + cancel.cancel(); + } + return; + } + // The replacement pipeline is spawned before its handles reach us + // and cancels its own token on an unexpected exit. Its disconnect + // callback cannot claim a reconnect while this loop owns the + // link, so a token that is already cancelled here is a dial that + // failed after auth — retry it, never install it as Live. + let result = result.and_then(|conn| { + if conn.0.is_cancelled() { + Err("replacement pipeline exited before it was installed".into()) + } else { + Ok(conn) + } + }); + match result { + Ok((cancel, pcm_tx)) => { + if let Some(old_cancel) = hs.audio_ws_cancel.replace(cancel) { + old_cancel.cancel(); + } + hs.audio_relay_pcm_tx = Some(pcm_tx); + hs.audio_link = AudioLink::Live; + None + } + Err(error) => { + failures += 1; + let draining = error.code() == Some("huddle_relay_draining"); + eprintln!("buzz-desktop: huddle audio redial {failures} failed: {error}"); + if Instant::now() >= deadline { + hs.audio_link = AudioLink::Lost; + None + } else { + hs.audio_link = AudioLink::Reconnecting { + attempt: failures, + draining, + }; + Some(if draining { + DRAINING_REDIAL_DELAY + } else { + backoff_delay(failures, jitter_unit()) + }) + } + } + } + }; + state.emit_huddle_state_changed(); + let Some(delay) = next_delay else { return }; + tokio::time::sleep(delay).await; + if !is_current(state, &target) { + return; + } } - hs.audio_relay_pcm_tx = Some(pcm_tx); - Ok(()) } + +fn is_current(state: &AppState, target: &ReconnectTarget) -> bool { + state + .huddle() + .map(|hs| hs.is_current_huddle(&target.ephemeral_channel_id, target.huddle_generation)) + .unwrap_or(false) +} + +/// Exponential delay for the `failures`-th consecutive failure, capped at +/// `BACKOFF_MAX`, then scaled into `[0.5, 1.0]` by `jitter` so clients that +/// lost the same pod do not redial in lockstep. +fn backoff_delay(failures: u32, jitter: f64) -> Duration { + let exponent = failures.saturating_sub(1).min(16); + let full = BACKOFF_BASE + .saturating_mul(1_u32 << exponent) + .min(BACKOFF_MAX); + full.mul_f64(0.5 + 0.5 * jitter.clamp(0.0, 1.0)) +} + +fn jitter_unit() -> f64 { + let mut bytes = [0u8; 4]; + // Entropy failure degrades to no jitter; the window is still honoured. + let _ = getrandom::getrandom(&mut bytes); + f64::from(u32::from_le_bytes(bytes)) / f64::from(u32::MAX) +} + +#[cfg(test)] +#[path = "reconnect_tests.rs"] +mod tests; diff --git a/desktop/src-tauri/src/huddle/reconnect_tests.rs b/desktop/src-tauri/src/huddle/reconnect_tests.rs new file mode 100644 index 00000000000..dae96ce0b50 --- /dev/null +++ b/desktop/src-tauri/src/huddle/reconnect_tests.rs @@ -0,0 +1,394 @@ +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use tokio::time::Instant; +use tokio_util::sync::CancellationToken; + +use super::{backoff_delay, run, AudioConnection, ReconnectTarget, RECONNECT_WINDOW}; +use crate::app_state::{build_app_state, AppState}; +use crate::huddle::relay_api::AudioRelayConnectError; +use crate::huddle::state::{AudioLink, HuddlePhase}; + +const HUDDLE: &str = "huddle-a"; + +fn live_huddle_state() -> Arc { + let state = build_app_state(); + { + let mut hs = state.huddle().expect("huddle lock"); + hs.phase = HuddlePhase::Active; + hs.ephemeral_channel_id = Some(HUDDLE.into()); + hs.parent_channel_id = Some("parent".into()); + hs.begin_huddle_lifetime(); + hs.audio_ws_cancel = Some(CancellationToken::new()); + } + Arc::new(state) +} + +fn connection() -> AudioConnection { + (CancellationToken::new(), tokio::sync::mpsc::channel(1).0) +} + +fn refused(code: &str) -> AudioRelayConnectError { + AudioRelayConnectError::from_relay_payload(&serde_json::json!({ + "code": code, + "message": "refused", + })) +} + +fn audio_link(state: &AppState) -> AudioLink { + state.huddle().expect("huddle lock").audio_link.clone() +} + +/// Dial that fails until `succeed_after` of virtual time has passed. +fn flaky_dial( + dials: Arc, + started: Instant, + succeed_after: Option, +) -> impl FnMut(ReconnectTarget) -> std::future::Ready> +{ + move |_| { + dials.fetch_add(1, Ordering::SeqCst); + let ok = succeed_after.is_some_and(|after| started.elapsed() >= after); + std::future::ready(if ok { + Ok(connection()) + } else { + Err("connection refused".into()) + }) + } +} + +#[tokio::test(start_paused = true)] +async fn recovers_after_a_thirty_five_second_outage_without_leaving() { + let state = live_huddle_state(); + let dials = Arc::new(AtomicU32::new(0)); + let started = Instant::now(); + + run( + &state, + flaky_dial(Arc::clone(&dials), started, Some(Duration::from_secs(35))), + ) + .await; + + assert!(started.elapsed() >= Duration::from_secs(35)); + assert!(dials.load(Ordering::SeqCst) > 1, "must have retried"); + let hs = state.huddle().expect("huddle lock"); + assert_eq!(hs.audio_link, AudioLink::Live); + assert_eq!(hs.phase, HuddlePhase::Active, "phase never leaves Active"); + assert!(hs.audio_relay_pcm_tx.is_some(), "new sender installed"); + assert!(hs + .audio_ws_cancel + .as_ref() + .is_some_and(|c| !c.is_cancelled())); +} + +#[tokio::test(start_paused = true)] +async fn draining_refusal_redials_promptly_instead_of_backing_off() { + let state = live_huddle_state(); + let dial_times = Arc::new(std::sync::Mutex::new(Vec::::new())); + let times = Arc::clone(&dial_times); + + run(&state, move |_| { + let mut t = times.lock().unwrap(); + t.push(Instant::now()); + let n = t.len(); + std::future::ready(match n { + // Two ordinary refusals grow the backoff; the drain hint resets it. + 1 | 2 => Err("connection refused".into()), + 3 => Err(refused("huddle_relay_draining")), + _ => Ok(connection()), + }) + }) + .await; + + let t = dial_times.lock().unwrap(); + assert_eq!(t.len(), 4); + let ordinary_gap = t[2] - t[1]; + let draining_gap = t[3] - t[2]; + assert!( + draining_gap <= Duration::from_millis(250), + "{draining_gap:?}" + ); + assert!( + draining_gap < ordinary_gap, + "{draining_gap:?} vs {ordinary_gap:?}" + ); + assert_eq!(audio_link(&state), AudioLink::Live); +} + +#[tokio::test(start_paused = true)] +async fn draining_is_visible_in_state_while_waiting() { + let state = live_huddle_state(); + let seen = Arc::new(std::sync::Mutex::new(Vec::::new())); + let observer = Arc::clone(&seen); + let observed_state = Arc::clone(&state); + + run(&state, move |_| { + observer.lock().unwrap().push(audio_link(&observed_state)); + let n = observer.lock().unwrap().len(); + std::future::ready(match n { + 1 => Err(refused("huddle_relay_draining")), + _ => Ok(connection()), + }) + }) + .await; + + let seen = seen.lock().unwrap(); + assert_eq!( + seen.as_slice(), + [ + AudioLink::Reconnecting { + attempt: 0, + draining: false + }, + AudioLink::Reconnecting { + attempt: 1, + draining: true + }, + ] + ); +} + +#[tokio::test(start_paused = true)] +async fn leave_during_backoff_stops_the_loop_and_never_installs_a_socket() { + let state = live_huddle_state(); + let dials = Arc::new(AtomicU32::new(0)); + let loop_state = Arc::clone(&state); + let loop_dials = Arc::clone(&dials); + + let loop_task = tokio::spawn(async move { + run(&loop_state, move |_| { + loop_dials.fetch_add(1, Ordering::SeqCst); + std::future::ready(Err::("connection refused".into())) + }) + .await; + }); + tokio::task::yield_now().await; + assert!(matches!(audio_link(&state), AudioLink::Reconnecting { .. })); + + // Intentional leave while the loop sleeps between dials. + state + .huddle() + .expect("huddle lock") + .reset_preserving_generation(); + + tokio::time::timeout(RECONNECT_WINDOW * 2, loop_task) + .await + .expect("loop must exit on leave, not run out the window") + .expect("loop task"); + let dials_at_exit = dials.load(Ordering::SeqCst); + assert!(dials_at_exit <= 2, "{dials_at_exit} dials after leave"); + let hs = state.huddle().expect("huddle lock"); + assert_eq!(hs.phase, HuddlePhase::Idle); + assert_eq!(hs.audio_link, AudioLink::Live, "idle state is not Lost"); + assert!(hs.audio_ws_cancel.is_none()); +} + +#[tokio::test(start_paused = true)] +async fn replacement_huddle_during_backoff_is_left_untouched() { + let state = live_huddle_state(); + let loop_state = Arc::clone(&state); + let returned = Arc::new(std::sync::Mutex::new(Vec::::new())); + let returned_for_dial = Arc::clone(&returned); + + let loop_task = tokio::spawn(async move { + run(&loop_state, move |_| { + let mut r = returned_for_dial.lock().unwrap(); + std::future::ready(if r.is_empty() { + // First dial fails so the loop sleeps; the huddle is swapped + // underneath it, and the next dial "succeeds" for the old one. + r.push(CancellationToken::new()); + Err("connection refused".into()) + } else { + let (cancel, tx) = connection(); + r.push(cancel.clone()); + Ok((cancel, tx)) + }) + }) + .await; + }); + tokio::task::yield_now().await; + + let replacement_cancel = CancellationToken::new(); + { + let mut hs = state.huddle().expect("huddle lock"); + hs.reset_preserving_generation(); + hs.phase = HuddlePhase::Active; + hs.ephemeral_channel_id = Some("huddle-b".into()); + hs.begin_huddle_lifetime(); + hs.audio_ws_cancel = Some(replacement_cancel.clone()); + } + + tokio::time::timeout(RECONNECT_WINDOW * 2, loop_task) + .await + .expect("loop must exit when the huddle is replaced") + .expect("loop task"); + let hs = state.huddle().expect("huddle lock"); + assert_eq!(hs.ephemeral_channel_id.as_deref(), Some("huddle-b")); + assert!( + !replacement_cancel.is_cancelled(), + "new huddle's socket untouched" + ); + assert_eq!(hs.audio_link, AudioLink::Live); + let returned = returned.lock().unwrap(); + // Any socket opened for the stale huddle is cancelled, not installed. + for stale in returned.iter().skip(1) { + assert!(stale.is_cancelled()); + } +} + +#[tokio::test(start_paused = true)] +async fn leave_while_a_dial_is_in_flight_cancels_the_fresh_socket() { + let state = live_huddle_state(); + let dial_state = Arc::clone(&state); + let fresh = CancellationToken::new(); + let fresh_for_dial = fresh.clone(); + + run(&state, move |_| { + // The user leaves while the dial is on the wire; the dial still wins + // a socket for the huddle that no longer exists. + dial_state + .huddle() + .expect("huddle lock") + .reset_preserving_generation(); + std::future::ready(Ok(( + fresh_for_dial.clone(), + tokio::sync::mpsc::channel(1).0, + ))) + }) + .await; + + assert!(fresh.is_cancelled(), "stale socket must be cancelled"); + let hs = state.huddle().expect("huddle lock"); + assert_eq!(hs.phase, HuddlePhase::Idle); + assert!( + hs.audio_ws_cancel.is_none(), + "nothing installed on an idle huddle" + ); + assert!(hs.audio_relay_pcm_tx.is_none()); +} + +#[tokio::test(start_paused = true)] +async fn replacement_pipeline_dying_before_install_is_redialed_not_installed() { + let state = live_huddle_state(); + let installed = Arc::new(std::sync::Mutex::new(Vec::::new())); + let handed_out = Arc::clone(&installed); + + run(&state, move |_| { + let (cancel, pcm_tx) = connection(); + let mut h = handed_out.lock().unwrap(); + if h.is_empty() { + // Auth and join succeed, then the spawned pipeline exits before + // the loop installs it: the pipeline cancels its own token and + // its disconnect callback is refused by `claim_reconnect`. + cancel.cancel(); + } + h.push(cancel.clone()); + std::future::ready(Ok((cancel, pcm_tx))) + }) + .await; + + let installed = installed.lock().unwrap(); + assert_eq!(installed.len(), 2, "dead replacement must be redialed"); + let hs = state.huddle().expect("huddle lock"); + assert_eq!(hs.audio_link, AudioLink::Live); + let live = hs.audio_ws_cancel.as_ref().expect("socket installed"); + assert!(!live.is_cancelled(), "Live must never hold a dead pipeline"); + assert!( + hs.audio_relay_pcm_tx.is_some(), + "sender belongs to the live pipeline" + ); +} + +#[tokio::test(start_paused = true)] +async fn duplicate_disconnect_signals_do_not_start_a_second_loop() { + let state = live_huddle_state(); + let dials = Arc::new(AtomicU32::new(0)); + let started = Instant::now(); + + let first_state = Arc::clone(&state); + let first_dials = Arc::clone(&dials); + let first = tokio::spawn(async move { + run( + &first_state, + flaky_dial(first_dials, started, Some(Duration::from_secs(3))), + ) + .await; + }); + tokio::task::yield_now().await; + + let second_dials = Arc::new(AtomicU32::new(0)); + run(&state, flaky_dial(Arc::clone(&second_dials), started, None)).await; + assert_eq!( + second_dials.load(Ordering::SeqCst), + 0, + "second loop must not dial" + ); + + first.await.expect("first loop"); + assert_eq!(audio_link(&state), AudioLink::Live); +} + +#[tokio::test(start_paused = true)] +async fn exhausting_the_window_marks_audio_lost_and_ends_the_loop() { + let state = live_huddle_state(); + let dials = Arc::new(AtomicU32::new(0)); + let started = Instant::now(); + + tokio::time::timeout( + RECONNECT_WINDOW * 2, + run(&state, flaky_dial(Arc::clone(&dials), started, None)), + ) + .await + .expect("loop must terminate (no task leak)"); + + assert!( + started.elapsed() >= RECONNECT_WINDOW, + "{:?}", + started.elapsed() + ); + let hs = state.huddle().expect("huddle lock"); + assert_eq!(hs.audio_link, AudioLink::Lost); + assert_eq!(hs.phase, HuddlePhase::Active, "still joined; user decides"); + assert!(hs.audio_relay_pcm_tx.is_none()); + let n = dials.load(Ordering::SeqCst); + assert!((12..=300).contains(&n), "{n} dials in the window"); +} + +#[tokio::test(start_paused = true)] +async fn not_live_huddle_is_ignored() { + let state = Arc::new(build_app_state()); + let dials = Arc::new(AtomicU32::new(0)); + run(&state, flaky_dial(Arc::clone(&dials), Instant::now(), None)).await; + assert_eq!(dials.load(Ordering::SeqCst), 0); + assert_eq!(audio_link(&state), AudioLink::Live); +} + +#[test] +fn backoff_grows_caps_and_jitters_within_half_to_full() { + assert_eq!(backoff_delay(1, 1.0), Duration::from_millis(250)); + assert_eq!(backoff_delay(2, 1.0), Duration::from_millis(500)); + assert_eq!(backoff_delay(6, 1.0), Duration::from_secs(5)); + assert_eq!( + backoff_delay(40, 1.0), + Duration::from_secs(5), + "no overflow" + ); + assert_eq!(backoff_delay(1, 0.0), Duration::from_millis(125)); + assert_eq!(backoff_delay(1, 7.0), Duration::from_millis(250), "clamped"); +} + +#[test] +fn audio_link_serializes_as_tagged_status() { + let live = serde_json::to_value(AudioLink::Live).unwrap(); + assert_eq!(live, serde_json::json!({ "status": "live" })); + let reconnecting = serde_json::to_value(AudioLink::Reconnecting { + attempt: 2, + draining: true, + }) + .unwrap(); + assert_eq!( + reconnecting, + serde_json::json!({ "status": "reconnecting", "attempt": 2, "draining": true }) + ); +} diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index 190397aa054..51b83b5dd52 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -69,13 +69,84 @@ fn build_audio_auth_event( .map_err(|e| format!("sign: {e}")) } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct AudioRelayConnectError { + code: Option, + message: String, +} + +impl AudioRelayConnectError { + pub(crate) fn code(&self) -> Option<&str> { + self.code.as_deref() + } + + /// The relay admitted another socket as this agent's voice publisher. + /// Callers drop the utterance instead of surfacing an error. + pub(crate) fn is_lost_election(&self) -> bool { + self.code() == Some("duplicate_identity") + } + + pub(crate) fn from_relay_payload(value: &serde_json::Value) -> Self { + Self { + code: value["code"].as_str().map(str::to_string), + message: value["message"] + .as_str() + .unwrap_or("unknown relay error") + .to_string(), + } + } +} + +impl From for AudioRelayConnectError { + fn from(message: String) -> Self { + Self { + code: None, + message, + } + } +} + +impl From<&str> for AudioRelayConnectError { + fn from(message: &str) -> Self { + message.to_string().into() + } +} + +impl std::fmt::Display for AudioRelayConnectError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self.code.as_deref() { + Some(code) => write!( + formatter, + "audio relay auth error [{code}]: {}", + self.message + ), + None => formatter.write_str(&self.message), + } + } +} + +fn format_audio_relay_error(value: &serde_json::Value) -> AudioRelayConnectError { + AudioRelayConnectError::from_relay_payload(value) +} + +fn parse_audio_roster_peer(peer: &serde_json::Value) -> Option<(u8, String, u8, bool)> { + Some(( + u8::try_from(peer["peer_index"].as_u64()?).ok()?, + 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. + u8::try_from(peer["epoch"].as_u64().unwrap_or(0)).ok()?, + peer["role"].as_str() == Some("AgentTtsPublisher"), + )) +} + async fn connect_authenticated_audio_socket( channel_id: &str, parent_channel_id: Option<&str>, relay_url: &str, keys: &nostr::Keys, auth_tag_json: Option<&str>, -) -> Result<(WsSink, WsReceiver, u8, Vec<(u8, String, u8)>), String> { +) -> Result<(WsSink, WsReceiver, u8, Vec<(u8, String, u8, bool)>), AudioRelayConnectError> { use nostr::JsonUtil; let ws_url = format!("{relay_url}/huddle/{channel_id}/audio"); @@ -134,19 +205,7 @@ async fn connect_authenticated_audio_socket( 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() + peers.iter().filter_map(parse_audio_roster_peer).collect() }) .unwrap_or_default(); let peer_index = value["peer_index"] @@ -156,7 +215,7 @@ async fn connect_authenticated_audio_socket( break Ok((peer_index, peers)); } Some("error") => { - break Err(format!("audio relay auth error: {}", value["message"])); + break Err(format_audio_relay_error(&value)); } _ => continue, } @@ -182,7 +241,7 @@ pub(crate) async fn connect_audio_relay( channel_id: &str, parent_channel_id: Option<&str>, state: &AppState, -) -> Result<(CancellationToken, tokio::sync::mpsc::Sender>), String> { +) -> Result<(CancellationToken, tokio::sync::mpsc::Sender>), AudioRelayConnectError> { let relay_url = crate::relay::relay_ws_url_with_override(state); let keys = state.keys.lock().map_err(|e| e.to_string())?.clone(); @@ -191,6 +250,7 @@ pub(crate) async fn connect_audio_relay( tts_cancel, tts_active, local_tts_publishers, + audio_peer_pubkeys, remote_stt_pipeline, agent_pubkeys, human_floor, @@ -200,6 +260,7 @@ pub(crate) async fn connect_audio_relay( Arc::clone(&hs.tts_cancel), Arc::clone(&hs.tts_active), Arc::clone(&hs.local_tts_publishers), + Arc::clone(&hs.audio_peer_pubkeys), Arc::clone(&hs.remote_stt_pipeline), Arc::clone(&hs.agent_pubkeys), hs.human_floor.clone(), @@ -215,12 +276,8 @@ pub(crate) async fn connect_audio_relay( let cancel = CancellationToken::new(); let cancel_clone = cancel.clone(); let (pcm_tx, pcm_rx) = tokio::sync::mpsc::channel::>(50); - let output_device_name = state - .huddle_audio - .output_device - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone(); + let output_device_changes = state.huddle_audio.output_device_changes.subscribe(); + let output_device_name = output_device_changes.borrow().clone(); tokio::spawn(async move { if let Err(e) = audio_relay_pipeline(AudioRelayPipelineArgs { @@ -233,23 +290,27 @@ pub(crate) async fn connect_audio_relay( tts_cancel, tts_active, local_tts_publishers, + audio_peer_pubkeys, remote_stt_pipeline, agent_pubkeys, human_floor, output_device_name, + output_device_changes, }) .await { eprintln!("buzz-desktop: audio relay pipeline exited: {e}"); } - // Only emit the disconnect event for UNEXPECTED exits. - // Skip if already cancelled (teardown_huddle in progress). + // Only UNEXPECTED exits reconnect. An already-cancelled token means + // teardown_huddle is in progress and the huddle is going away. + // Cancelling before the reconnect call is load-bearing: if this + // pipeline dies before a running reconnect loop has installed its + // handles, the loop reads the cancelled token as a failed dial. if !cancel_clone.is_cancelled() { cancel_clone.cancel(); - if let Some(ref app) = app_handle { - use tauri::Emitter; - let _ = app.emit("huddle-audio-disconnected", ()); + if let Some(app) = app_handle { + super::reconnect::after_unexpected_disconnect(app).await; } } }); @@ -265,6 +326,75 @@ type WsReceiver = futures_util::stream::SplitStream; const TTS_BROADCAST_QUEUE_DEPTH: usize = 8; const TTS_BROADCAST_MAX_FRAMES: usize = 1_500; // 30 seconds at 20 ms/frame. +const AUDIO_SEND_QUEUE_DEPTH: usize = 4; + +#[derive(Default)] +struct AudioSendQueueState { + frames: std::collections::VecDeque>, + closed: bool, +} + +#[derive(Default)] +struct AudioSendQueue { + state: std::sync::Mutex, + ready: tokio::sync::Notify, +} + +impl AudioSendQueue { + fn push_latest(&self, frame: Vec) { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + if state.closed { + return; + } + if state.frames.len() == AUDIO_SEND_QUEUE_DEPTH { + state.frames.pop_front(); + } + state.frames.push_back(frame); + drop(state); + self.ready.notify_one(); + } + + fn close(&self) { + self.state + .lock() + .unwrap_or_else(|error| error.into_inner()) + .closed = true; + self.ready.notify_waiters(); + } + + async fn pop(&self) -> Option> { + loop { + let notified = self.ready.notified(); + { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + if let Some(frame) = state.frames.pop_front() { + return Some(frame); + } + if state.closed { + return None; + } + } + notified.await; + } + } +} + +async fn wire_send_loop( + queue: std::sync::Arc, + sink: std::sync::Arc>, +) -> Result<(), String> +where + S: futures_util::Sink + Unpin, + S::Error: std::fmt::Display, +{ + while let Some(frame) = queue.pop().await { + let mut sink = sink.lock().await; + sink.send(WsMsg::Binary(frame.into())) + .await + .map_err(|error| format!("audio send: {error}"))?; + } + Ok(()) +} struct QueuedTtsFrame { epoch: u64, @@ -313,14 +443,14 @@ fn queue_tts_broadcast_packet( /// Open a send-only v2 Huddle audio peer authenticated as a locally managed /// agent. The relay therefore assigns the synthesized stream to that agent's /// existing pubkey; no backend or wire-protocol extension is required. -pub(crate) async fn connect_tts_audio_publisher( +pub(super) async fn connect_tts_audio_publisher( channel_id: &str, parent_channel_id: Option<&str>, state: &AppState, keys: &nostr::Keys, auth_tag_json: Option<&str>, local_tts_publishers: super::tts::LocalTtsPublishers, -) -> Result { +) -> Result { let relay_url = crate::relay::relay_ws_url_with_override(state); let (ws_tx, ws_rx, peer_index, _) = connect_authenticated_audio_socket( channel_id, @@ -454,14 +584,16 @@ struct AudioRelayPipelineArgs { pcm_rx: tokio::sync::mpsc::Receiver>, cancel: CancellationToken, app_handle: Option, - initial_peers: Vec<(u8, String, u8)>, + initial_peers: Vec<(u8, String, u8, bool)>, tts_cancel: Arc, tts_active: Arc, local_tts_publishers: super::tts::LocalTtsPublishers, + audio_peer_pubkeys: Arc>>, remote_stt_pipeline: Arc>>>, agent_pubkeys: Arc>>, human_floor: super::human_floor::HumanFloor, output_device_name: Option, + output_device_changes: tokio::sync::watch::Receiver>, } async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String> { @@ -475,16 +607,18 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String tts_cancel, tts_active, local_tts_publishers, + audio_peer_pubkeys, remote_stt_pipeline, agent_pubkeys, human_floor, output_device_name, + output_device_changes, } = 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)) + .set_bitrate(opus::Bitrate::Bits(32_000)) .map_err(|e| format!("opus bitrate: {e}"))?; encoder .set_dtx(true) @@ -496,8 +630,13 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String let ws_tx = StdArc::new(tokio::sync::Mutex::new(ws_tx)); let ws_tx_send = StdArc::clone(&ws_tx); let cancel_send = cancel.clone(); + let send_queue = StdArc::new(AudioSendQueue::default()); + let wire_queue = StdArc::clone(&send_queue); - let send_task = tokio::spawn(async move { + let mut wire_send_task = tokio::spawn(wire_send_loop(wire_queue, ws_tx_send)); + + let encode_queue = StdArc::clone(&send_queue); + let mut encode_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; @@ -528,7 +667,6 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]])) .collect(); - 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 @@ -565,20 +703,17 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String 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. - } + encode_queue.push_latest(frame); seq = seq.wrapping_add(1); ts_48k = ts_48k.wrapping_add(super::jitter::FRAME_TIMESTAMP_DELTA); } } } - let mut tx = ws_tx_send.lock().await; - let _ = tx.send(WsMsg::Close(None)).await; + encode_queue.close(); }); - let recv_task = tokio::spawn(super::playout::run_playout_recv_loop( + let mut recv_task = tokio::spawn(super::playout::run_playout_recv_loop( ws_rx, ws_tx, sink_handle, @@ -588,19 +723,30 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String tts_active, tts_cancel, local_tts_publishers, + audio_peer_pubkeys, remote_stt_pipeline, agent_pubkeys, human_floor, + output_device_changes, )); - // Wait for either task to finish, then abort the survivor. - use futures_util::future::Either; - match futures_util::future::select(std::pin::pin!(send_task), std::pin::pin!(recv_task)).await { - Either::Left((_, recv_handle)) => recv_handle.abort(), - Either::Right((_, send_handle)) => send_handle.abort(), - } + // Any pipeline task ending tears down the other two. The encoder never + // waits on socket flow control; the bounded queue keeps only fresh audio. + let pipeline_result = tokio::select! { + result = &mut encode_task => result + .map_err(|error| format!("audio encode task: {error}")), + result = &mut wire_send_task => result + .map_err(|error| format!("audio send task: {error}"))?, + result = &mut recv_task => result + .map_err(|error| format!("audio receive task: {error}")), + }; - Ok(()) + send_queue.close(); + encode_task.abort(); + wire_send_task.abort(); + recv_task.abort(); + + pipeline_result } /// Fetch channel members with roles from the relay. Returns (pubkey, role) tuples. @@ -671,43 +817,5 @@ pub(crate) async fn count_human_members( } #[cfg(test)] -mod tests { - use super::*; - - #[test] - fn tts_upsampling_doubles_rate_with_linear_midpoints() { - assert_eq!( - upsample_tts_24k_to_48k(&[0.0, 1.0, -1.0]), - vec![0.0, 0.5, 1.0, 0.0, -1.0, -1.0] - ); - } - - #[test] - fn tts_queue_rejects_cancelled_versions_and_pads_twenty_ms_frames() { - let mut queue = std::collections::VecDeque::new(); - queue_tts_broadcast_packet( - &mut queue, - super::super::tts::TtsBroadcastPacket { - epoch: 1, - speaker_generation: 7, - samples_24k: vec![0.25; 480], - }, - 1, - 7, - ); - assert_eq!(queue.len(), 1); - assert_eq!(queue[0].samples_48k.len(), 960); - - queue_tts_broadcast_packet( - &mut queue, - super::super::tts::TtsBroadcastPacket { - epoch: 1, - speaker_generation: 7, - samples_24k: vec![0.5; 480], - }, - 2, - 7, - ); - assert_eq!(queue.len(), 1, "cancelled epoch must not enqueue"); - } -} +#[path = "relay_api_tests.rs"] +mod tests; diff --git a/desktop/src-tauri/src/huddle/relay_api_tests.rs b/desktop/src-tauri/src/huddle/relay_api_tests.rs new file mode 100644 index 00000000000..ee3d85599e4 --- /dev/null +++ b/desktop/src-tauri/src/huddle/relay_api_tests.rs @@ -0,0 +1,193 @@ +use super::*; + +#[test] +fn relay_auth_errors_preserve_stable_codes_for_ui_mapping() { + for (code, message) in [ + ("room_full", "room participant capacity reached"), + ("room_ended", "huddle has ended"), + ("huddle_relay_draining", "relay is draining; reconnect"), + ( + "huddle_owner_unreachable", + "could not reach the huddle owner", + ), + ("unsupported_version", "unsupported audio protocol version"), + ("upgrade_required", "audio protocol upgrade required"), + ] { + let payload = serde_json::json!({ + "type": "error", + "code": code, + "message": message, + }); + let error = format_audio_relay_error(&payload); + assert_eq!(error.code(), Some(code)); + assert_eq!( + error.to_string(), + format!("audio relay auth error [{code}]: {message}") + ); + } +} + +#[test] +fn audio_send_queue_drops_oldest_frame_when_full() { + let queue = AudioSendQueue::default(); + for value in 0..=AUDIO_SEND_QUEUE_DEPTH as u8 { + queue.push_latest(vec![value]); + } + let frames = queue + .state + .lock() + .expect("queue") + .frames + .iter() + .cloned() + .collect::>(); + assert_eq!(frames, vec![vec![1], vec![2], vec![3], vec![4]]); +} + +#[tokio::test] +async fn audio_send_queue_close_wakes_waiter_and_rejects_new_frames() { + let queue = std::sync::Arc::new(AudioSendQueue::default()); + let waiting_queue = std::sync::Arc::clone(&queue); + let waiter = tokio::spawn(async move { waiting_queue.pop().await }); + tokio::task::yield_now().await; + + queue.close(); + assert_eq!(waiter.await.expect("waiter"), None); + queue.push_latest(vec![1]); + assert_eq!(queue.pop().await, None); +} + +#[tokio::test] +async fn audio_send_queue_drains_before_reporting_closed() { + let queue = AudioSendQueue::default(); + queue.push_latest(vec![1]); + queue.close(); + + assert_eq!(queue.pop().await, Some(vec![1])); + assert_eq!(queue.pop().await, None); +} + +#[tokio::test] +async fn wire_send_failure_is_preserved_for_pipeline_owner() { + struct FailingSink; + impl futures_util::Sink for FailingSink { + type Error = &'static str; + + fn poll_ready( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Err("socket closed")) + } + fn start_send(self: std::pin::Pin<&mut Self>, _item: WsMsg) -> Result<(), Self::Error> { + Err("socket closed") + } + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + fn poll_close( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + } + + let queue = std::sync::Arc::new(AudioSendQueue::default()); + queue.push_latest(vec![1]); + let result = wire_send_loop( + queue, + std::sync::Arc::new(tokio::sync::Mutex::new(FailingSink)), + ) + .await; + assert!( + result.is_err_and(|error| error == "audio send: socket closed"), + "the reconnect owner must receive the socket send failure" + ); +} + +#[test] +fn authenticated_roster_parsing_preserves_only_authoritative_publisher_roles() { + let publisher = serde_json::json!({ + "peer_index": 7, + "pubkey": "agent", + "epoch": 3, + "role": "AgentTtsPublisher", + }); + assert_eq!( + parse_audio_roster_peer(&publisher), + Some((7, "agent".to_string(), 3, true)) + ); + + let participant = serde_json::json!({ + "peer_index": 8, + "pubkey": "human", + "role": "Participant", + }); + assert_eq!( + parse_audio_roster_peer(&participant), + Some((8, "human".to_string(), 0, false)) + ); + + let overflowing_index = serde_json::json!({ + "peer_index": 256, + "pubkey": "invalid", + "role": "AgentTtsPublisher", + }); + assert_eq!(parse_audio_roster_peer(&overflowing_index), None); +} + +#[test] +fn duplicate_identity_is_a_typed_lost_election() { + let error = format_audio_relay_error(&serde_json::json!({ + "code": "duplicate_identity", + "message": "publisher already connected", + })); + assert!(error.is_lost_election()); + + let other = format_audio_relay_error(&serde_json::json!({ + "code": "not_member", + "message": "membership required", + })); + assert!(!other.is_lost_election()); +} + +#[test] +fn tts_upsampling_doubles_rate_with_linear_midpoints() { + assert_eq!( + upsample_tts_24k_to_48k(&[0.0, 1.0, -1.0]), + vec![0.0, 0.5, 1.0, 0.0, -1.0, -1.0] + ); +} + +#[test] +fn tts_queue_rejects_cancelled_versions_and_pads_twenty_ms_frames() { + let mut queue = std::collections::VecDeque::new(); + queue_tts_broadcast_packet( + &mut queue, + super::super::tts::TtsBroadcastPacket { + epoch: 1, + speaker_generation: 7, + samples_24k: vec![0.25; 480], + }, + 1, + 7, + ); + assert_eq!(queue.len(), 1); + assert_eq!(queue[0].samples_48k.len(), 960); + + queue_tts_broadcast_packet( + &mut queue, + super::super::tts::TtsBroadcastPacket { + epoch: 1, + speaker_generation: 7, + samples_24k: vec![0.5; 480], + }, + 2, + 7, + ); + assert_eq!(queue.len(), 1, "cancelled epoch must not enqueue"); +} diff --git a/desktop/src-tauri/src/huddle/state.rs b/desktop/src-tauri/src/huddle/state.rs index c7aff1bf7e2..e57ab5e41e2 100644 --- a/desktop/src-tauri/src/huddle/state.rs +++ b/desktop/src-tauri/src/huddle/state.rs @@ -4,7 +4,7 @@ //! phase enum, voice input mode, and response types. use serde::{Deserialize, Serialize}; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, Arc, Mutex, Weak, @@ -43,9 +43,28 @@ pub enum HuddlePhase { Leaving, } +/// Health of the audio relay socket, independent of `phase`. +/// +/// `phase` stays `Active` while the socket is rebuilt so STT/TTS/agent voice +/// (which gate on `Connected | Active`) keep running through a blip. Only the +/// audio transport is being recovered; see `reconnect.rs`. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "status", rename_all = "snake_case")] +pub enum AudioLink { + #[default] + Live, + /// Rust is redialing the audio relay. `attempt` counts failed dials so far; + /// `draining` is true when the last refusal was `huddle_relay_draining`. + Reconnecting { attempt: u32, draining: bool }, + /// The retry window elapsed without a successful dial. The huddle is still + /// joined but carries no audio; the user must leave or rejoin. + Lost, +} + #[derive(Debug, Serialize, Deserialize)] pub struct HuddleState { pub phase: HuddlePhase, + pub audio_link: AudioLink, pub parent_channel_id: Option, pub ephemeral_channel_id: Option, /// Root event for the huddle's visible parent-channel thread. Transcript @@ -93,6 +112,12 @@ pub struct HuddleState { /// echo, never another socket authenticated as the same bot. #[serde(skip)] pub local_tts_publishers: tts::LocalTtsPublishers, + /// Relay-authoritative managed-agent TTS publishers, keyed by peer index. + /// The receive loop derives this only from publisher-role-bearing roster + /// snapshots/deltas. Removal immediately restores local fallback for the + /// next message. + #[serde(skip)] + pub audio_peer_pubkeys: Arc>>, /// Whether this client created the huddle (vs. joined it). /// Used to enforce that only the creator can end/archive the huddle. pub is_creator: bool, @@ -189,6 +214,7 @@ impl Clone for HuddleState { .clone(); Self { phase: self.phase.clone(), + audio_link: self.audio_link.clone(), parent_channel_id: self.parent_channel_id.clone(), ephemeral_channel_id: self.ephemeral_channel_id.clone(), huddle_thread_event_id: self.huddle_thread_event_id.clone(), @@ -201,6 +227,7 @@ impl Clone for HuddleState { remote_stt_pipeline: Arc::new(Mutex::new(None)), tts_pipeline: None, // Never clone the pipeline handle. local_tts_publishers: Arc::clone(&self.local_tts_publishers), + audio_peer_pubkeys: Arc::clone(&self.audio_peer_pubkeys), is_creator: self.is_creator, tts_enabled: self.tts_enabled, transcription_enabled: self.transcription_enabled, @@ -226,6 +253,7 @@ impl Default for HuddleState { let human_floor = HumanFloor::new(); Self { phase: HuddlePhase::Idle, + audio_link: AudioLink::Live, parent_channel_id: None, ephemeral_channel_id: None, huddle_thread_event_id: None, @@ -238,6 +266,7 @@ impl Default for HuddleState { remote_stt_pipeline: Arc::new(Mutex::new(None)), tts_pipeline: None, local_tts_publishers: tts::LocalTtsPublishers::default(), + audio_peer_pubkeys: Arc::new(Mutex::new(HashMap::new())), is_creator: false, tts_enabled: true, transcription_enabled: false, diff --git a/desktop/src-tauri/src/huddle/stt.rs b/desktop/src-tauri/src/huddle/stt.rs index c27bf38b649..3d6c35db2b2 100644 --- a/desktop/src-tauri/src/huddle/stt.rs +++ b/desktop/src-tauri/src/huddle/stt.rs @@ -100,7 +100,7 @@ impl SttPipeline { ptt_active: Option>, manual_mic_unmuted: Option>, human_floor: HumanFloor, - output_device: Option, + output_device_changes: tokio::sync::watch::Receiver>, ) -> Result<(Self, tokio_mpsc::Receiver), String> { let (audio_tx, audio_rx) = mpsc::sync_channel::(AUDIO_QUEUE_DEPTH); let (text_tx, text_rx) = tokio_mpsc::channel::(64); @@ -120,7 +120,7 @@ impl SttPipeline { ptt_active_worker, manual_mic_unmuted_worker, human_floor, - output_device, + output_device_changes, ) }) .map_err(|e| format!("failed to spawn stt-worker thread: {e}"))?; @@ -449,7 +449,7 @@ fn stt_worker( ptt_active: Option>, manual_mic_unmuted: Option>, human_floor: HumanFloor, - output_device: Option, + output_device_changes: tokio::sync::watch::Receiver>, ) { // ── 1. Initialise sherpa-onnx recognizer ───────────────────────────────── // @@ -566,7 +566,7 @@ fn stt_worker( manual_gate, &human_floor, local_barge_in_state, - output_device.as_deref(), + output_device_changes.borrow().as_deref(), track_local_floor, ); } diff --git a/desktop/src-tauri/src/huddle/tts.rs b/desktop/src-tauri/src/huddle/tts.rs index 3f12f883ba7..a1ca712e719 100644 --- a/desktop/src-tauri/src/huddle/tts.rs +++ b/desktop/src-tauri/src/huddle/tts.rs @@ -200,7 +200,7 @@ impl TtsPipeline { cancel: Arc, human_floor: HumanFloor, voice: &str, - output_device: Option, + output_device_changes: tokio::sync::watch::Receiver>, activity_app: Option, ) -> Result { let (text_tx, text_rx) = mpsc::sync_channel::(TEXT_QUEUE_DEPTH); @@ -255,7 +255,7 @@ impl TtsPipeline { worker_playback_probe, worker_broadcasters, ), - output_device, + output_device_changes, activity_app, startup_tx, ) @@ -318,6 +318,19 @@ fn authorize_or_defer_queued_text( } } +pub(super) fn apply_pending_output_device_change( + output_device_changes: &mut tokio::sync::watch::Receiver>, + mut apply: impl FnMut(Option<&str>) -> Result<(), String>, +) { + if !output_device_changes.has_changed().unwrap_or(false) { + return; + } + let selected = output_device_changes.borrow_and_update().clone(); + if let Err(error) = apply(selected.as_deref()) { + eprintln!("buzz-desktop: live TTS output device switch failed: {error}"); + } +} + #[allow(clippy::too_many_arguments)] fn tts_worker( model_dir: PathBuf, @@ -325,7 +338,7 @@ fn tts_worker( text_rx: mpsc::Receiver, human_floor: HumanFloor, control_state: WorkerControlState, - output_device: Option, + mut output_device_changes: tokio::sync::watch::Receiver>, activity_app: Option, startup_tx: mpsc::SyncSender>, ) { @@ -397,16 +410,17 @@ fn tts_worker( // ── 3. Initialise rodio output device ───────────────────────────────────── use rodio::buffer::SamplesBuffer; - let sink_handle = match super::audio_output::open_output_sink_by_name(output_device.as_deref()) - { - Ok(h) => h, - Err(e) => { - let error = format!("TTS audio output initialization failed: {e}"); - eprintln!("buzz-desktop: tts stage=startup status=failed reason=output_open"); - let _ = startup_tx.send(Err(error)); - return; - } - }; + let initial_output_device = output_device_changes.borrow().clone(); + let mut sink_handle = + match super::audio_output::open_output_sink_by_name(initial_output_device.as_deref()) { + Ok(h) => h, + Err(e) => { + let error = format!("TTS audio output initialization failed: {e}"); + eprintln!("buzz-desktop: tts stage=startup status=failed reason=output_open"); + let _ = startup_tx.send(Err(error)); + return; + } + }; let channels = match NonZero::new(1u16) { Some(c) => c, @@ -505,11 +519,17 @@ fn tts_worker( channels, rate, }; - let append_audio = |prepared: PreparedModelAudio, - route_id: u64, - speaker_pubkey: Option<&str>, - speaker_generation: u64, - floor_epoch: u64| { + let mut append_audio = |prepared: PreparedModelAudio, + route_id: u64, + speaker_pubkey: Option<&str>, + speaker_generation: u64, + floor_epoch: u64| { + apply_pending_output_device_change(&mut output_device_changes, |selected| { + let next_sink = super::audio_output::open_output_sink_by_name(selected)?; + playback.replace_output_mixer(next_sink.mixer()); + sink_handle = next_sink; + Ok(()) + }); let broadcast_samples = speaker_pubkey.map(|_| prepared.buffer.clone()); append_worker_audio( &append_context, diff --git a/desktop/src-tauri/src/huddle/tts_playback.rs b/desktop/src-tauri/src/huddle/tts_playback.rs index 8a90c018994..deac955f736 100644 --- a/desktop/src-tauri/src/huddle/tts_playback.rs +++ b/desktop/src-tauri/src/huddle/tts_playback.rs @@ -122,6 +122,17 @@ impl PlaybackCoordinator { } } + pub(super) fn replace_output_mixer(&self, mixer: &Mixer) { + *self.mixer.lock().unwrap_or_else(PoisonError::into_inner) = Some(mixer.clone()); + let old_player = { + let mut state = self.lock(); + state.first_append = true; + state.output_lease.begin_hangover(Instant::now()); + state.player.replace(Player::connect_new(mixer)) + }; + drop(old_player); + } + fn lock(&self) -> MutexGuard<'_, PlaybackState> { self.state.lock().unwrap_or_else(PoisonError::into_inner) } @@ -422,6 +433,44 @@ mod tests { ) } + #[test] + fn pending_output_watch_change_replaces_live_playback_before_next_append() { + let (playback, _old_source) = coordinator(); + append_second(&playback); + assert!(!playback.empty()); + + let (output_tx, mut output_rx) = tokio::sync::watch::channel(None); + output_tx.send_replace(Some("new route".to_string())); + let channels = NonZero::new(1).expect("nonzero channels"); + let rate = NonZero::new(24_000).expect("nonzero rate"); + let (next_mixer, _next_source) = rodio::mixer::mixer(channels, rate); + crate::huddle::tts::apply_pending_output_device_change(&mut output_rx, |selected| { + assert_eq!(selected, Some("new route")); + playback.replace_output_mixer(&next_mixer); + Ok(()) + }); + + assert!(playback.empty(), "the old-route queue must be dropped"); + append_second(&playback); + assert!(!playback.empty(), "subsequent audio must use the new mixer"); + } + + #[test] + fn replacing_output_mixer_drops_old_route_and_accepts_new_audio() { + let (playback, _old_source) = coordinator(); + append_second(&playback); + assert!(!playback.empty()); + + let channels = NonZero::new(1).expect("nonzero channels"); + let rate = NonZero::new(24_000).expect("nonzero rate"); + let (next_mixer, _next_source) = rodio::mixer::mixer(channels, rate); + playback.replace_output_mixer(&next_mixer); + + assert!(playback.empty()); + append_second(&playback); + assert!(!playback.empty()); + } + #[test] fn floor_authorized_append_does_not_reenter_the_coordinator_lock() { let (playback, _unpulled_source) = coordinator(); diff --git a/desktop/src-tauri/src/huddle/tts_settings.rs b/desktop/src-tauri/src/huddle/tts_settings.rs index 75cfef26e55..1e96cd6f09a 100644 --- a/desktop/src-tauri/src/huddle/tts_settings.rs +++ b/desktop/src-tauri/src/huddle/tts_settings.rs @@ -45,6 +45,8 @@ pub struct HuddleAudioSettingsState { pub tts_transition: tokio::sync::Mutex<()>, /// Selected huddle output device. `None` uses the system default. pub output_device: Mutex>, + /// Live output-route changes for an active huddle playout loop. + pub output_device_changes: tokio::sync::watch::Sender>, } #[derive(Debug, Clone, Serialize, PartialEq, Eq)] @@ -624,7 +626,7 @@ pub async fn preview_pocket_voice( cancel, super::human_floor::HumanFloor::new(), &voice_name, - output_device, + tokio::sync::watch::channel(output_device).1, None, )?; pipeline.speak("Hello! This is how I’ll read agent responses.".to_string())?; diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 613040b8095..46ae3b8e180 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -69,7 +69,6 @@ use huddle::{ check_pipeline_hotstart, close_huddle_companion, confirm_huddle_active, download_voice_models, end_huddle, get_huddle_agent_pubkeys, get_huddle_state, get_model_status, get_voice_input_mode, interrupt_huddle_speech, join_huddle, leave_huddle, open_huddle_window, push_audio_pcm, - reconnect::reconnect_huddle_audio, remove_agent_from_huddle, set_huddle_manual_mic_unmuted, set_huddle_transcription_enabled, set_tts_enabled, set_voice_input_mode, speak_agent_message, start_huddle, start_stt_pipeline, HuddlePhase, @@ -788,7 +787,6 @@ pub fn run() { close_huddle_companion, open_huddle_window, push_audio_pcm, - reconnect_huddle_audio, start_stt_pipeline, set_huddle_transcription_enabled, download_voice_models, diff --git a/desktop/src/features/channels/ui/useHuddleChannelMessages.ts b/desktop/src/features/channels/ui/useHuddleChannelMessages.ts index 5a3a7c40419..55856079064 100644 --- a/desktop/src/features/channels/ui/useHuddleChannelMessages.ts +++ b/desktop/src/features/channels/ui/useHuddleChannelMessages.ts @@ -4,8 +4,10 @@ import { huddleWindowChannelId } from "@/features/huddle/lib/huddleWindow"; import { mergeMessages } from "@/features/messages/hooks"; import { channelWindowThreadSummaries, + flattenChannelWindowEvents, type ChannelWindowStore, } from "@/features/messages/lib/channelWindowStore"; +import { getThreadReference } from "@/features/messages/lib/threading"; import { useThreadRepliesForRoots } from "@/features/messages/useThreadReplies"; import type { Channel, RelayEvent } from "@/shared/api/types"; @@ -25,6 +27,19 @@ type HuddleChannelMessagesOptions = { windowStore?: ChannelWindowStore; }; +export function seedHuddleThreadReplies( + windowStore: ChannelWindowStore | undefined, +): ReadonlyMap { + const repliesByRoot = new Map(); + if (!windowStore) return repliesByRoot; + for (const event of flattenChannelWindowEvents(windowStore)) { + const { parentId, rootId } = getThreadReference(event.tags); + if (!parentId || !rootId) continue; + repliesByRoot.set(rootId, [...(repliesByRoot.get(rootId) ?? []), event]); + } + return repliesByRoot; +} + export function useHuddleChannelMessages({ activeChannel, isHuddleTranscript, @@ -51,9 +66,16 @@ export function useHuddleChannelMessages({ : [], [isHuddleTranscript, threadSummaries], ); + const placeholderThreadRepliesByRoot = React.useMemo( + () => seedHuddleThreadReplies(windowStore), + [windowStore], + ); const huddleThreadReplies = useThreadRepliesForRoots( activeChannel, huddleThreadRootIds, + { + placeholderDataByRoot: placeholderThreadRepliesByRoot, + }, ); const resolvedMessages = React.useMemo( () => diff --git a/desktop/src/features/huddle/HuddleContext.tsx b/desktop/src/features/huddle/HuddleContext.tsx index 8e6ccbbd890..86c39adb521 100644 --- a/desktop/src/features/huddle/HuddleContext.tsx +++ b/desktop/src/features/huddle/HuddleContext.tsx @@ -814,62 +814,6 @@ export function HuddleProvider({ }; }, [ownsAudioSession]); - // Unexpected audio-owner/pod disconnects are recoverable: keep the huddle, - // mic, and voice pipelines live while Rust reconnects only the audio WS. - // `tokenRef` makes an intentional leave/start supersede this loop, and the - // in-flight guard collapses duplicate disconnect events from failed dials. - const audioReconnectInFlightRef = React.useRef(false); - React.useEffect(() => { - if (!ownsAudioSession) return; - - let cancelled = false; - let unlisten: (() => void) | null = null; - listen("huddle-audio-disconnected", () => { - if (cancelled || audioReconnectInFlightRef.current) return; - audioReconnectInFlightRef.current = true; - const reconnectToken = tokenRef.current; - - void (async () => { - // Keep a long enough tail for Kubernetes Service endpoint removal after - // a draining pod flips readiness. Early retries make remote-owner - // handoff fast; the two 2s attempts prevent a client connected to the - // draining pod itself from exhausting before kube-proxy converges. - const delaysMs = [0, 100, 250, 500, 1_000, 2_000, 2_000]; - for (const delayMs of delaysMs) { - if (cancelled || tokenRef.current !== reconnectToken) return; - if (delayMs > 0) { - await new Promise((resolve) => window.setTimeout(resolve, delayMs)); - } - if (cancelled || tokenRef.current !== reconnectToken) return; - try { - await invoke("reconnect_huddle_audio"); - // Success installs a live replacement pipeline. If it later fails, - // its Tauri event arrives after this loop releases the in-flight - // guard and starts a fresh bounded recovery cycle. Repeating those - // cycles is intentional while the relay remains connectable. - return; - } catch { - // A draining pod may still receive the first retry before Service - // endpoints converge. Keep the bounded backoff client-local. - } - } - - if (!cancelled && tokenRef.current === reconnectToken) { - await leaveHuddleRef.current(); - } - })().finally(() => { - audioReconnectInFlightRef.current = false; - }); - }).then((fn) => { - if (cancelled) fn(); - else unlisten = fn; - }); - return () => { - cancelled = true; - unlisten?.(); - }; - }, [ownsAudioSession]); - // High-frequency (20-30 Hz) audio levels live in their own context so their // churn re-renders only the meter components, not every useHuddle consumer. const levelsValue = React.useMemo( diff --git a/desktop/src/features/huddle/components/HuddleBar.tsx b/desktop/src/features/huddle/components/HuddleBar.tsx index d5a0423cf7c..382d2b7eed9 100644 --- a/desktop/src/features/huddle/components/HuddleBar.tsx +++ b/desktop/src/features/huddle/components/HuddleBar.tsx @@ -22,6 +22,7 @@ import type { RelayEvent } from "@/shared/api/types"; import { KIND_HUDDLE_REACTION } from "@/shared/constants/kinds"; import { cn } from "@/shared/lib/cn"; import { rewriteRelayUrl } from "@/shared/lib/mediaUrl"; +import { getStorageItem, setStorageItem } from "@/shared/lib/safeStorage"; import { useDocumentVisible } from "@/shared/lib/useDocumentVisible"; import { Button } from "@/shared/ui/button"; import { useEmojiBurst } from "@/shared/ui/EmojiBurstProvider"; @@ -29,13 +30,14 @@ import { Popover, PopoverContent, PopoverTrigger } from "@/shared/ui/popover"; import { Tooltip, TooltipContent, TooltipTrigger } from "@/shared/ui/tooltip"; import { useHuddle, useHuddleLevels } from "../HuddleContext"; import { useHuddleParticipantRoster } from "../hooks/useHuddleParticipantRoster"; +import { audioLinkNotice, type HuddleAudioLink } from "../lib/audioLink"; import { AddAgentDialog, type AgentAddResult } from "./AddAgentDialog"; import type { HuddleAgentVoiceSettings } from "./AgentVoiceMenu"; import { MicControls, SpeakerControls } from "./MicControls"; import { HuddleParticipantsControl } from "./ParticipantList"; import { truncatePubkey } from "@/shared/lib/pubkey"; -// Mirrors HuddleState in src-tauri/src/huddle/mod.rs. +// Mirrors HuddleState in src-tauri/src/huddle/state.rs. type HuddleState = { phase: | "idle" @@ -44,6 +46,7 @@ type HuddleState = { | "connected" | "active" | "leaving"; + audio_link: HuddleAudioLink; parent_channel_id: string | null; ephemeral_channel_id: string | null; huddle_thread_event_id: string | null; @@ -71,9 +74,14 @@ const HUDDLE_STATE_FALLBACK_INTERVAL_MS = 30_000; const HUDDLE_MODEL_STATUS_INTERVAL_MS = 10_000; const HUDDLE_REACTION_NAME_MAX = 48; const HEADPHONES_HINT_SEEN_STORAGE_KEY = "buzz.huddle.headphones-hint-seen"; +const PTT_HINT_SEEN_STORAGE_KEY = "buzz.huddle.ptt-hint-seen"; function hasSeenHeadphonesHint() { - return window.localStorage.getItem(HEADPHONES_HINT_SEEN_STORAGE_KEY) === "1"; + return getStorageItem(HEADPHONES_HINT_SEEN_STORAGE_KEY) === "1"; +} + +function hasSeenPttHint() { + return getStorageItem(PTT_HINT_SEEN_STORAGE_KEY) === "1"; } function isVisibleHuddleState(state: HuddleState | null) { @@ -188,6 +196,8 @@ export function HuddleBar({ const [headphonesHintDismissed, setHeadphonesHintDismissed] = React.useState( hasSeenHeadphonesHint, ); + const [pttHintDismissed, setPttHintDismissed] = + React.useState(hasSeenPttHint); const [isLeaving, setIsLeaving] = React.useState(false); const [showAddAgent, setShowAddAgent] = React.useState(false); const [agentAddError, setAgentAddError] = React.useState(null); @@ -330,9 +340,13 @@ export function HuddleBar({ const mainHadActiveHuddleRef = React.useRef(false); const dismissHeadphonesHint = React.useCallback(() => { - window.localStorage.setItem(HEADPHONES_HINT_SEEN_STORAGE_KEY, "1"); + setStorageItem(HEADPHONES_HINT_SEEN_STORAGE_KEY, "1"); setHeadphonesHintDismissed(true); }, []); + const dismissPttHint = React.useCallback(() => { + setStorageItem(PTT_HINT_SEEN_STORAGE_KEY, "1"); + setPttHintDismissed(true); + }, []); React.useEffect(() => { onVisibilityChange?.(isHuddleVisible); @@ -502,6 +516,7 @@ export function HuddleBar({ const hasAvailableMic = micConnected; const ttsEnabled = barState.tts_enabled; const transcriptionEnabled = barState.transcription_enabled; + const audioLinkBanner = audioLinkNotice(barState.audio_link); // Self-removing detection: remote-peer audio plays through native rodio // today (outside the WebView render graph), so the browser's AEC has no // far-end reference. The AEC follow-up PR flips this constant in the @@ -582,6 +597,28 @@ export function HuddleBar({ )} >
+ {/* Audio link banner — Rust is redialing, or gave up. */} + {audioLinkBanner && ( + + + {audioLinkBanner.message} + + + )} + {/* Error banner */} {huddleError && (
void; isMuted: boolean; onToggleMute: () => void; isPttMode: boolean; @@ -93,6 +95,8 @@ function usePrefersReducedMotion(): boolean { export function MicControls({ compact = false, + showPttHint = false, + onPttHintDismiss, isMuted, onToggleMute, isPttMode, @@ -146,7 +150,7 @@ export function MicControls({ "overflow-hidden border border-sidebar-border/80 bg-transparent text-sidebar-foreground/70 shadow-none", )} > - + - {isPttMode && !micUnavailable && isEffectivelyMuted ? ( + {showPttHint ? ( + + Hold + + {pushToTalkShortcut} + + to talk — click to switch to open mic + + ) : isPttMode && !micUnavailable && isEffectivelyMuted ? ( Click to unmute or hold diff --git a/desktop/src/features/huddle/lib/audioLink.test.mjs b/desktop/src/features/huddle/lib/audioLink.test.mjs new file mode 100644 index 00000000000..b183a320a3d --- /dev/null +++ b/desktop/src/features/huddle/lib/audioLink.test.mjs @@ -0,0 +1,27 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { audioLinkNotice } from "./audioLink.ts"; + +test("live audio shows no notice", () => { + assert.equal(audioLinkNotice({ status: "live" }), null); + assert.equal(audioLinkNotice(undefined), null); +}); + +test("reconnecting is informational and names a relay restart when draining", () => { + assert.deepEqual( + audioLinkNotice({ status: "reconnecting", attempt: 3, draining: false }), + { tone: "info", message: "Audio dropped. Reconnecting…" }, + ); + assert.deepEqual( + audioLinkNotice({ status: "reconnecting", attempt: 1, draining: true }), + { tone: "info", message: "The huddle relay is restarting. Reconnecting…" }, + ); +}); + +test("a lost link is an error that tells the user what to do", () => { + assert.deepEqual(audioLinkNotice({ status: "lost" }), { + tone: "error", + message: "Couldn’t reconnect audio. Leave and rejoin the huddle.", + }); +}); diff --git a/desktop/src/features/huddle/lib/audioLink.ts b/desktop/src/features/huddle/lib/audioLink.ts new file mode 100644 index 00000000000..d7c1eaf9c87 --- /dev/null +++ b/desktop/src/features/huddle/lib/audioLink.ts @@ -0,0 +1,30 @@ +// Mirrors `AudioLink` in src-tauri/src/huddle/state.rs. +export type HuddleAudioLink = + | { status: "live" } + | { status: "reconnecting"; attempt: number; draining: boolean } + | { status: "lost" }; + +/** + * Banner copy for a degraded audio link, or null while audio is live. Rust + * owns the retry loop; the renderer only reports what it is doing. + */ +export function audioLinkNotice( + link: HuddleAudioLink | undefined, +): { tone: "info" | "error"; message: string } | null { + switch (link?.status) { + case "reconnecting": + return { + tone: "info", + message: link.draining + ? "The huddle relay is restarting. Reconnecting…" + : "Audio dropped. Reconnecting…", + }; + case "lost": + return { + tone: "error", + message: "Couldn’t reconnect audio. Leave and rejoin the huddle.", + }; + default: + return null; + } +} diff --git a/desktop/src/features/huddle/lib/huddleError.test.mjs b/desktop/src/features/huddle/lib/huddleError.test.mjs index ec61b834678..fcea5ea6e7c 100644 --- a/desktop/src/features/huddle/lib/huddleError.test.mjs +++ b/desktop/src/features/huddle/lib/huddleError.test.mjs @@ -44,3 +44,26 @@ test("uses action-specific fallback copy for unknown errors", () => { "Couldn’t start the huddle.", ); }); + +test("maps relay room, drain, owner, and protocol errors to useful copy", () => { + const cases = [ + ["room_full", "This huddle is full."], + ["room_ended", "This huddle has ended."], + ["huddle_relay_draining", "The huddle relay is restarting. Reconnecting…"], + [ + "huddle_owner_unreachable", + "The huddle relay can’t be reached. Try again in a moment.", + ], + ["unsupported_version", "Update Buzz to join this huddle."], + [ + "upgrade_required", + "This huddle uses a newer audio version. Update Buzz, then try again.", + ], + ]; + for (const [code, expected] of cases) { + assert.equal( + formatHuddleActionError(`audio relay error: ${code}`, "join"), + expected, + ); + } +}); diff --git a/desktop/src/features/huddle/lib/huddleError.ts b/desktop/src/features/huddle/lib/huddleError.ts index 7ee47b21337..02b5c258fb2 100644 --- a/desktop/src/features/huddle/lib/huddleError.ts +++ b/desktop/src/features/huddle/lib/huddleError.ts @@ -1,7 +1,40 @@ export type HuddleAction = "join" | "start"; -const HUDDLE_AUDIO_UNAVAILABLE_MESSAGE = - "Huddle audio isn’t available on this server. Ask an administrator to turn it on."; +const HUDDLE_ERROR_MESSAGES: ReadonlyArray<{ + codes: readonly string[]; + message: string; +}> = [ + { + codes: [ + "huddle_audio_unavailable", + "huddle audio unavailable in this deployment", + ], + message: + "Huddle audio isn’t available on this server. Ask an administrator to turn it on.", + }, + { codes: ["room_full"], message: "This huddle is full." }, + { + codes: ["room_ended", "channel is archived"], + message: "This huddle has ended.", + }, + { + codes: ["huddle_relay_draining"], + message: "The huddle relay is restarting. Reconnecting…", + }, + { + codes: ["huddle_owner_unreachable"], + message: "The huddle relay can’t be reached. Try again in a moment.", + }, + { + codes: ["unsupported_version"], + message: "Update Buzz to join this huddle.", + }, + { + codes: ["upgrade_required"], + message: + "This huddle uses a newer audio version. Update Buzz, then try again.", + }, +]; function rawErrorMessage(error: unknown): string | null { if (error instanceof Error) { @@ -20,12 +53,10 @@ export function formatHuddleActionError( const message = rawErrorMessage(error)?.trim(); const normalized = message?.toLowerCase(); - if ( - normalized?.includes("huddle_audio_unavailable") || - normalized?.includes("huddle audio unavailable in this deployment") - ) { - return HUDDLE_AUDIO_UNAVAILABLE_MESSAGE; - } + const mapped = HUDDLE_ERROR_MESSAGES.find(({ codes }) => + codes.some((code) => normalized?.includes(code)), + ); + if (mapped) return mapped.message; if (message) { return message; diff --git a/desktop/src/features/messages/useThreadReplies.test.mjs b/desktop/src/features/messages/useThreadReplies.test.mjs index 53399cd2f39..8f06482c041 100644 --- a/desktop/src/features/messages/useThreadReplies.test.mjs +++ b/desktop/src/features/messages/useThreadReplies.test.mjs @@ -2,6 +2,12 @@ import assert from "node:assert/strict"; import { readFile } from "node:fs/promises"; import test from "node:test"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { JSDOM } from "jsdom"; +import React from "react"; + +import { useThreadRepliesForRoots } from "./useThreadReplies.ts"; + test("thread replies trust the relay-provided aux closure", async () => { const source = await readFile( new URL("./useThreadReplies.ts", import.meta.url), @@ -13,3 +19,117 @@ test("thread replies trust the relay-provided aux closure", async () => { ); assert.match(source, /replies\.push\(\.\.\.response\.events\)/); }); + +function reply(id, rootId) { + return { + id, + pubkey: "a".repeat(64), + kind: 9, + created_at: 1_700_000_000, + content: "reply", + tags: [["e", rootId, "", "reply"]], + sig: "sig", + }; +} + +async function withHookEnvironment(run) { + const dom = new JSDOM("", { + url: "http://localhost/", + }); + const previousGlobals = { + window: globalThis.window, + document: globalThis.document, + navigator: globalThis.navigator, + tauri: globalThis.__TAURI_INTERNALS__, + act: globalThis.IS_REACT_ACT_ENVIRONMENT, + }; + const pendingRequests = new Map(); + const tauriInternals = { + invoke: (command, args) => { + assert.equal(command, "get_thread_replies"); + return new Promise((resolve) => { + pendingRequests.set(args.rootEventId, resolve); + }); + }, + transformCallback: () => 1, + }; + Object.assign(globalThis, { + window: dom.window, + document: dom.window.document, + __TAURI_INTERNALS__: tauriInternals, + IS_REACT_ACT_ENVIRONMENT: true, + }); + Object.defineProperty(globalThis, "navigator", { + configurable: true, + value: dom.window.navigator, + }); + dom.window.__TAURI_INTERNALS__ = tauriInternals; + + try { + const { act, renderHook } = await import("@testing-library/react"); + await run({ act, pendingRequests, renderHook }); + } finally { + dom.window.close(); + Object.assign(globalThis, { + window: previousGlobals.window, + document: previousGlobals.document, + __TAURI_INTERNALS__: previousGlobals.tauri, + IS_REACT_ACT_ENVIRONMENT: previousGlobals.act, + }); + Object.defineProperty(globalThis, "navigator", { + configurable: true, + value: previousGlobals.navigator, + }); + } +} + +function hookWrapper(client) { + return ({ children }) => + React.createElement(QueryClientProvider, { client }, children); +} + +const channel = { id: "huddle", channelType: "stream" }; + +test("window-seeded huddle roots fetch and settle without a visible gap", async () => { + await withHookEnvironment(async ({ act, pendingRequests, renderHook }) => { + const seededRootId = "3".repeat(64); + const seededReply = reply("4".repeat(64), seededRootId); + const authoritativeReply = reply("5".repeat(64), seededRootId); + const client = new QueryClient({ + defaultOptions: { queries: { retry: false, gcTime: 0 } }, + }); + const placeholderDataByRoot = new Map([[seededRootId, [seededReply]]]); + const view = renderHook( + () => + useThreadRepliesForRoots(channel, [seededRootId], { + placeholderDataByRoot, + }), + { wrapper: hookWrapper(client) }, + ); + + try { + assert.deepEqual(view.result.current.events, [seededReply]); + assert.equal(pendingRequests.has(seededRootId), true); + await act(async () => { + pendingRequests.get(seededRootId)({ + events: [authoritativeReply], + next_cursor: null, + }); + for ( + let attempts = 0; + view.result.current.events[0]?.id !== authoritativeReply.id && + attempts < 50; + attempts += 1 + ) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + }); + assert.deepEqual(view.result.current.events, [authoritativeReply]); + } finally { + view.unmount(); + await client.cancelQueries(); + client.clear(); + client.unmount(); + } + }); +}); diff --git a/desktop/src/features/messages/useThreadReplies.ts b/desktop/src/features/messages/useThreadReplies.ts index 4d602348f95..aa82f2eb4b9 100644 --- a/desktop/src/features/messages/useThreadReplies.ts +++ b/desktop/src/features/messages/useThreadReplies.ts @@ -4,7 +4,6 @@ import { useQuery, useQueryClient, } from "@tanstack/react-query"; - import { threadRepliesKey, sortMessages, @@ -106,17 +105,24 @@ export function combineThreadRepliesResults( * replies into the chat timeline so companion and in-app presentations show the * same conversation without opening a transient thread surface. */ +export type ThreadRepliesForRootsOptions = { + placeholderDataByRoot?: ReadonlyMap; +}; + export function useThreadRepliesForRoots( activeChannel: Channel | null, rootIds: readonly string[], + options: ThreadRepliesForRootsOptions = {}, ) { const queryClient = useQueryClient(); const channelId = activeChannel?.id ?? "none"; + const placeholderDataByRoot = options.placeholderDataByRoot; return useQueries({ queries: rootIds.map((rootId) => ({ queryKey: threadRepliesKey(channelId, rootId), enabled: activeChannel !== null && activeChannel.channelType !== "forum", queryFn: () => loadThreadReplies(queryClient, channelId, rootId), + placeholderData: () => placeholderDataByRoot?.get(rootId), staleTime: 0, gcTime: 60 * 60 * 1_000, })), diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 0bd7bc6eecc..e4ce2a35522 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -18,6 +18,7 @@ import type { UnreadCatchUpChannelResult } from "@/shared/api/tauriUnreadCatchUp import { relayClient } from "@/shared/api/relayClient"; import { activateRateLimit } from "@/shared/api/relayRateLimitGate"; import { resolveAgentParallelism } from "@/features/agents/lib/agentParallelism"; +import type { HuddleAudioLink } from "@/features/huddle/lib/audioLink"; import type { ConnectionState } from "@/shared/api/relayClientShared"; import type { ChannelTemplate, @@ -3438,6 +3439,7 @@ type MockHuddleState = { transcription_enabled: boolean; is_creator: boolean; voice_input_mode: "push_to_talk" | "voice_activity"; + audio_link: HuddleAudioLink; }; type PersistedMockHuddle = { @@ -3553,6 +3555,7 @@ function initializeMockHuddle( transcription_enabled: seed.transcriptionEnabled ?? false, is_creator: seed.isCreator ?? true, voice_input_mode: "push_to_talk", + audio_link: { status: "live" }, }, }; } @@ -11198,6 +11201,7 @@ export function maybeInstallE2eTauriMocks() { transcription_enabled: false, is_creator: true, voice_input_mode: "push_to_talk", + audio_link: { status: "live" }, }, }; refreshMockHuddleMembership(activeConfig); @@ -11292,6 +11296,7 @@ export function maybeInstallE2eTauriMocks() { transcription_enabled: false, is_creator: false, voice_input_mode: "push_to_talk", + audio_link: { status: "live" }, }); return null; case "set_huddle_transcription_enabled":