fix(desktop): restore human barge-in over agent TTS in huddles (#6431)

## Problem

Humans cannot interrupt agents in huddles: agent TTS keeps playing over
a talking human (reported by @tlongwell-block, 2026-08-20). The full
requirement: **any human talking — local or remote, any input mode, any
audio rig — must interrupt any agent on the huddle.**

## This is a restoration, not a new feature

- **`b29c8cdaa` (#4281, 2026-08-04) deleted working local VAD
barge-in.** The pre-image shipped `BARGE_IN_DEBOUNCE_FRAMES = 20` (320
ms sustained speech cancels TTS), live at the production call site
(`pipeline.rs` passed `Some(tts_cancel)` unconditionally). The same day,
`ce3cf3cd2` (#4694) flipped the default input mode from VAD to
push-to-talk, which masked the loss.
- **`068a83b09` (#5671, 2026-08-13) removed the remaining TTS-awareness
plumbing from STT** (deliberately, to keep transcribing over agent audio
— a good change). Consequence: this PR is a re-plumb through the
`PlaybackCoordinator` from #6341, not a revert.
- A gap that **never** worked is also closed: PTT-mode users who open
the mic via the mute button (`manual_mic_unmuted` postdates the
deletion) transcribed fine but could not barge in.

## Design

All floor state lives under the single `PlaybackCoordinator` lock —
onset acceptance, epoch bump, synthesis invalidation, player
replacement, and output lease are one committed transition. No cancel
flag can be observed out of order with the playback state it describes.

- **Human floor**: local + per-peer remote ownership with epoch
invalidation. Onset cancels playback by queue replacement; late
synthesis for a stale epoch cannot append or restart.
- **Local onset (VAD)**: on an **isolated output route** (all CoreAudio
output-stream terminals report headphones), a confirmed short onset
interrupts immediately — no echo path exists. On a **coupled route**
(speakers, unknown, virtual, mixed), the restored **20-frame / 320 ms
sustained-speech debounce** discriminates a real human from speaker
bleed; the deleted code's comment records that 80 ms was tried and
false-triggered on laptop speakers. Route classification is queried
fresh at each onset (never cached — default-device re-routing mid-huddle
would strand a stale verdict).
- **Mic-open gate is per-frame**: barge-in observes on any frame where
the mic is actually open — pure VAD mode, or PTT with the mic manually
unmuted (key-held frames defer to the shortcut's own cancel).
- **Remote onset**: sustained non-DTX frames from a peer enter the same
persistent floor (independent of whether playback is live — a human
speaking while TTS is idle blocks late-arriving synthesis from starting
over them). Release on sustained DTX/absence, peer departure, and
recv-loop exit, with guards so a vanished peer cannot wedge the floor.
- **Output lease**: accepted appends renew an `Active` lease;
drain/cancel/onset start a 100 ms tail hangover (conservative against
measured ~12 ms/~1 ms CoreAudio tails), so speaker-tail bleed in the
just-drained window cannot self-trigger the coupled path.

## Known limitations (phase 2 pointers)

- Coupled-route mid-output barge-in pays the 320 ms debounce; a
playback-reference echo discriminator would shorten it.
- A speakers-rig participant's bleed can enter their mic and hold the
floor for other machines (bounded by release debounce).
- Non-macOS routes classify as coupled (fail-safe).

## Verification

- Full `buzz-desktop --lib` suite at head `b0459ae4a`: **2707 passed, 0
failed, 18 ignored** (pinned cargo 1.95.0).
- Exact CI recipe `just desktop-tauri-clippy`: PASS at head; base arm at
merge-base `b728a2af3` confirms the two
`#[allow(clippy::too_many_arguments)]`s cover branch-caused threshold
crossings (human_floor threading), not inherited noise.
- Regression tests pin: 20-frame threshold + reset-on-gap, short-coupled
rejection, sustained-coupled acceptance, coupled-idle acceptance,
remote-idle delayed-TTS rejection, output-tail hangover boundary (during
= rejected, after = accepted), isolated onset, per-frame mic-open gate
truth table, PTT+manual-unmute sustained coupled acquisition.
- `LocalBargeIn::observe` is covered as two joined halves (gate truth
table in `local_barge_in.rs`, floor transition in `tts_playback.rs`);
its body is a straight-line wrapper around a live CoreAudio query, left
uninjected deliberately.
- Coverage precision (mutation-verified):
`manual_open_ptt_sustained_speech_acquires_coupled_floor` pins the gate
→ 20-frame debounce → acquire → floor-blocked chain on the
coupled-**idle** cell. The live-output override leg is carried by
`sustained_coupled_speech_overrides_live_output_suppression`
(tts_playback.rs); the joiner's `sustained_coupled` argument is not
load-bearing there (flipping it to `false` leaves the test green, while
shortening the debounce by one frame turns it red).
- Live arms in progress: pre-regression build `b29c8cdaa^` staged to
confirm the deleted mechanism worked; two-endpoint remote-leg test
pending a second human.

## Commits

1. `fb681a5a9` — restore human barge-in (coordinator floor, lease, route
isolation, remote floor, 320 ms coupled debounce)
2. `b4265418e` — enable barge-in for manually opened mics (per-frame
gate; closes the PTT-open-mic gap)
3. `886489f2b` — extract local barge-in policy module (file-size
ratchet; also hoists the CoreAudio route query from per-frame to
per-onset, named in the commit message)
4. `b0459ae4a` — two targeted clippy allows for the widened worker
signatures

## Credits

Built by **Wren**. Regression archaeology and the PTT-open-mic gap by
**Dawn** (who also killed her own first fix as vacuous and caught a
clippy blocker before it hit CI). Review blockers (coordinator
serialization, idle-onset floors, output lease) by **Mari**. Live rig
verification by **Max**. Coordination and verification by **Eva**.
Opened by Eva with Tyler's explicit direction; commits carry agent
trailers.

---------

Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>

desktop/src-tauri/src/huddle/audio_output.rs

@@ -97,3 +97,70 @@ pub(crate) fn open_output_sink_by_name(
 
     rodio::DeviceSinkBuilder::open_default_sink().map_err(|e| format!("audio output: {e}"))
 }
+
+fn device_type_is_isolated(device_type: rodio::cpal::DeviceType) -> bool {
+    use rodio::cpal::DeviceType;
+    matches!(
+        device_type,
+        DeviceType::Headphones
+            | DeviceType::Headset
+            | DeviceType::Earpiece
+            | DeviceType::HearingAid
+    )
+}
+
+/// Conservative route-isolation query using cpal's safe structured device
+/// description. This is intentionally re-evaluated at confirmed local onset,
+/// so a route change cannot leave a stale isolated capability behind.
+pub(crate) fn output_route_is_isolated(preferred: Option<&str>) -> bool {
+    use rodio::cpal::traits::HostTrait;
+    use rodio::DeviceTrait;
+
+    let host = rodio::cpal::default_host();
+    let device = match preferred.filter(|name| !name.is_empty()) {
+        Some(name) => {
+            let Ok(devices) = host.output_devices() else {
+                return false;
+            };
+            let mut matches = devices.filter(|device| {
+                device
+                    .description()
+                    .ok()
+                    .map(|description| description.name().to_owned())
+                    == Some(name.to_owned())
+            });
+            let Some(device) = matches.next() else {
+                return false;
+            };
+            if matches.next().is_some() {
+                return false;
+            }
+            device
+        }
+        None => match host.default_output_device() {
+            Some(device) => device,
+            None => return false,
+        },
+    };
+
+    device
+        .description()
+        .is_ok_and(|description| device_type_is_isolated(description.device_type()))
+}
+
+#[cfg(test)]
+mod route_isolation_tests {
+    use super::device_type_is_isolated;
+    use rodio::cpal::DeviceType;
+
+    #[test]
+    fn only_positive_isolated_terminal_types_are_accepted() {
+        assert!(device_type_is_isolated(DeviceType::Headphones));
+        assert!(device_type_is_isolated(DeviceType::Headset));
+        assert!(device_type_is_isolated(DeviceType::Earpiece));
+        assert!(device_type_is_isolated(DeviceType::HearingAid));
+        assert!(!device_type_is_isolated(DeviceType::Speaker));
+        assert!(!device_type_is_isolated(DeviceType::Virtual));
+        assert!(!device_type_is_isolated(DeviceType::Unknown));
+    }
+}

desktop/src-tauri/src/huddle/human_floor.rs

@@ -0,0 +1,73 @@
+//! Shared human-floor handle backed by the TTS playback coordinator.
+
+use std::sync::Arc;
+
+use super::tts_playback::{HumanFloorAuthorization, PlaybackCoordinator};
+
+#[derive(Clone)]
+pub(crate) struct HumanFloor {
+    playback: Arc<PlaybackCoordinator>,
+}
+
+impl std::fmt::Debug for HumanFloor {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.debug_struct("HumanFloor").finish_non_exhaustive()
+    }
+}
+
+impl Default for HumanFloor {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl HumanFloor {
+    pub(crate) fn new() -> Self {
+        Self {
+            playback: Arc::new(PlaybackCoordinator::unbound()),
+        }
+    }
+
+    pub(super) fn playback(&self) -> Arc<PlaybackCoordinator> {
+        Arc::clone(&self.playback)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn is_blocked(&self) -> bool {
+        self.playback.human_floor_blocked()
+    }
+
+    pub(crate) fn epoch(&self) -> u64 {
+        self.playback.human_floor_epoch()
+    }
+
+    pub(super) fn authorization(&self, epoch: u64) -> HumanFloorAuthorization {
+        self.playback.human_floor_authorization(epoch)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn permits(&self, epoch: u64) -> bool {
+        self.authorization(epoch) == HumanFloorAuthorization::Permitted
+    }
+
+    pub(crate) fn enter_local(&self, route_isolated: bool, sustained_coupled_speech: bool) -> bool {
+        self.playback
+            .enter_local_human_floor(route_isolated, sustained_coupled_speech)
+    }
+
+    pub(crate) fn leave_local(&self) {
+        self.playback.leave_local_human_floor();
+    }
+
+    pub(crate) fn enter_remote(&self, peer: u8) {
+        self.playback.enter_remote_human_floor(peer);
+    }
+
+    pub(crate) fn leave_remote(&self, peer: u8) {
+        self.playback.leave_remote_human_floor(peer);
+    }
+
+    pub(crate) fn clear_remote(&self) {
+        self.playback.clear_remote_human_floor();
+    }
+}

desktop/src-tauri/src/huddle/latency_bench.rs

@@ -141,6 +141,7 @@ fn baseline_stt_fake_llm_tts_first_audio() {
         tts_dir,
         Arc::clone(&tts_active),
         Arc::clone(&tts_cancel),
+        super::human_floor::HumanFloor::new(),
         "eve",
         None, // default output device
         None, // no Tauri app handle
@@ -152,7 +153,14 @@ fn baseline_stt_fake_llm_tts_first_audio() {
     );
 
     let t = Instant::now();
-    let (stt, mut text_rx) = SttPipeline::new(stt_dir, None, None).expect("stt pipeline");
+    let (stt, mut text_rx) = SttPipeline::new(
+        stt_dir,
+        None,
+        None,
+        super::human_floor::HumanFloor::new(),
+        None,
+    )
+    .expect("stt pipeline");
     // Recognizer loads inside the worker thread; give it time, then verify
     // liveness via a first throwaway feed below.
     std::thread::sleep(Duration::from_secs(2));

desktop/src-tauri/src/huddle/local_barge_in.rs

@@ -0,0 +1,151 @@
+//! Local VAD barge-in policy and coupled-output debounce.
+
+use super::human_floor::HumanFloor;
+
+/// Whether local audio should use VAD barge-in for this frame.
+///
+/// This currently matches STT's `vad_flush_allowed`, but the two decisions are
+/// kept separate deliberately: one assigns cancellation ownership and the
+/// other controls utterance endpointing.
+pub(super) fn enabled(ptt_mode: bool, manually_open: bool, ptt_held: bool) -> bool {
+    !ptt_mode || (manually_open && !ptt_held)
+}
+
+/// Consecutive 16 ms VAD-positive frames required to restore local barge-in
+/// on acoustically coupled output. The prior implementation shipped 20 frames
+/// after 5 frames caused speaker-bleed self-cancellation (`b29c8cdaa^`).
+const COUPLED_BARGE_IN_FRAMES: usize = 20;
+
+#[derive(Debug, Default)]
+pub(super) struct LocalBargeIn {
+    acquired_floor: bool,
+    coupled_positive_frames: usize,
+}
+
+impl LocalBargeIn {
+    pub(super) fn observe(
+        &mut self,
+        probability: f32,
+        confirmed_onset: bool,
+        human_floor: &HumanFloor,
+        output_device: Option<&str>,
+        onset_threshold: f32,
+    ) {
+        if self.acquired_floor {
+            return;
+        }
+        let sustained_coupled = self.track_sustained_coupled(probability, onset_threshold);
+        if !confirmed_onset && !sustained_coupled {
+            return;
+        }
+        let route_isolated = super::audio_output::output_route_is_isolated(output_device);
+        self.acquire(human_floor, route_isolated, sustained_coupled);
+    }
+
+    pub(super) fn acquire(
+        &mut self,
+        human_floor: &HumanFloor,
+        route_isolated: bool,
+        sustained_coupled: bool,
+    ) {
+        self.acquired_floor = human_floor.enter_local(route_isolated, sustained_coupled);
+    }
+
+    fn track_sustained_coupled(&mut self, probability: f32, onset_threshold: f32) -> bool {
+        if probability > onset_threshold {
+            self.coupled_positive_frames = self.coupled_positive_frames.saturating_add(1);
+        } else {
+            self.coupled_positive_frames = 0;
+        }
+        self.coupled_positive_frames >= COUPLED_BARGE_IN_FRAMES
+    }
+
+    pub(super) fn release(&mut self, human_floor: &HumanFloor) {
+        if self.acquired_floor {
+            human_floor.leave_local();
+        }
+        *self = Self::default();
+    }
+}
+
+#[derive(Debug)]
+pub(super) struct WorkerLocalBargeIn {
+    state: LocalBargeIn,
+    human_floor: HumanFloor,
+}
+
+impl WorkerLocalBargeIn {
+    pub(super) fn new(human_floor: HumanFloor) -> Self {
+        Self {
+            state: LocalBargeIn::default(),
+            human_floor,
+        }
+    }
+}
+
+impl std::ops::Deref for WorkerLocalBargeIn {
+    type Target = LocalBargeIn;
+
+    fn deref(&self) -> &Self::Target {
+        &self.state
+    }
+}
+
+impl std::ops::DerefMut for WorkerLocalBargeIn {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.state
+    }
+}
+
+impl Drop for WorkerLocalBargeIn {
+    fn drop(&mut self) {
+        self.state.release(&self.human_floor);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn manual_open_mic_enables_vad_barge_in_in_ptt_mode() {
+        assert!(enabled(true, true, false));
+        assert!(!enabled(true, false, false));
+        assert!(!enabled(true, true, true));
+        assert!(enabled(false, false, false));
+    }
+
+    #[test]
+    fn manual_open_ptt_sustained_speech_acquires_coupled_floor() {
+        assert!(enabled(true, true, false));
+        let human_floor = HumanFloor::new();
+        let mut barge_in = LocalBargeIn::default();
+        for _ in 0..COUPLED_BARGE_IN_FRAMES {
+            let sustained = barge_in.track_sustained_coupled(0.9, 0.5);
+            if sustained {
+                barge_in.acquire(&human_floor, false, true);
+            }
+        }
+        assert!(barge_in.acquired_floor);
+        assert!(human_floor.is_blocked());
+    }
+
+    #[test]
+    fn coupled_barge_in_requires_twenty_consecutive_positive_frames() {
+        let mut barge_in = LocalBargeIn::default();
+        for _ in 0..COUPLED_BARGE_IN_FRAMES - 1 {
+            assert!(!barge_in.track_sustained_coupled(0.9, 0.5));
+        }
+        assert!(barge_in.track_sustained_coupled(0.9, 0.5));
+    }
+
+    #[test]
+    fn coupled_barge_in_debounce_resets_on_a_non_speech_frame() {
+        let mut barge_in = LocalBargeIn::default();
+        for _ in 0..COUPLED_BARGE_IN_FRAMES - 1 {
+            assert!(!barge_in.track_sustained_coupled(0.9, 0.5));
+        }
+        assert!(!barge_in.track_sustained_coupled(0.1, 0.5));
+        assert!(!barge_in.track_sustained_coupled(0.9, 0.5));
+    }
+}

desktop/src-tauri/src/huddle/mod.rs

@@ -28,9 +28,11 @@ pub mod agent_voice;
 pub mod agents;
 pub mod audio_output;
 mod commands;
+mod human_floor;
 pub mod jitter;
 #[cfg(test)]
 mod latency_bench;
+mod local_barge_in;
 pub mod models;
 pub mod pipeline;
 pub mod playout;
@@ -42,6 +44,8 @@ pub mod state;
 pub mod stt;
 pub mod transcription;
 pub mod tts;
+#[path = "tts_playback.rs"]
+mod tts_playback;
 pub mod tts_settings;
 mod tts_voice_import;
 mod tts_voice_registry;

desktop/src-tauri/src/huddle/pipeline.rs

@@ -311,6 +311,8 @@ pub(crate) async fn maybe_start_stt_pipeline(
         stt_starting,
         ptt_active_for_stt,
         manual_mic_unmuted_for_stt,
+        human_floor,
+        output_device,
         old_stt,
     ) = {
         let mut hs = state.huddle()?;
@@ -346,6 +348,13 @@ pub(crate) async fn maybe_start_stt_pipeline(
             stt_starting,
             ptt,
             manual_mic_unmuted,
+            hs.human_floor.clone(),
+            state
+                .huddle_audio
+                .output_device
+                .lock()
+                .unwrap_or_else(|e| e.into_inner())
+                .clone(),
             old,
         )
     };
@@ -353,7 +362,13 @@ pub(crate) async fn maybe_start_stt_pipeline(
     drop(old_stt);
 
     let constructed = tokio::task::spawn_blocking(move || {
-        stt::SttPipeline::new(model_dir, ptt_active_for_stt, manual_mic_unmuted_for_stt)
+        stt::SttPipeline::new(
+            model_dir,
+            ptt_active_for_stt,
+            manual_mic_unmuted_for_stt,
+            human_floor,
+            output_device,
+        )
     })
     .await;
     let (pipeline, text_rx) = match constructed {
@@ -457,7 +472,7 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result<bool, S
     // Atomically check preconditions and claim the construction slot.
     // The sentinel prevents a second caller from starting construction
     // while we're building outside the lock.
-    let (tts_active, tts_cancel, tts_starting) = {
+    let (tts_active, tts_cancel, human_floor, tts_starting) = {
         let hs = state.huddle()?;
         if hs.tts_pipeline.is_some() {
             return Ok(false);
@@ -471,6 +486,7 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result<bool, S
         (
             Arc::clone(&hs.tts_active),
             Arc::clone(&hs.tts_cancel),
+            hs.human_floor.clone(),
             Arc::clone(&hs.tts_starting),
         )
     };
@@ -484,6 +500,7 @@ pub(crate) async fn maybe_start_tts_pipeline(state: &AppState) -> Result<bool, S
             model_dir,
             tts_active,
             tts_cancel,
+            human_floor,
             &initial_voice,
             output_device,
             app,

desktop/src-tauri/src/huddle/playout.rs

@@ -30,6 +30,7 @@ use futures_util::{SinkExt, StreamExt};
 use tokio_tungstenite::tungstenite::Message as WsMsg;
 use tokio_util::sync::CancellationToken;
 
+use super::human_floor::HumanFloor;
 use super::jitter::{PeerJitterBuffer, SAMPLE_RATE_HZ};
 use super::relay_api::{WsStream, REMOTE_SPEECH_THRESHOLD};
 use super::wire::{FrameHeader, FLAG_DTX, V2_HEADER_LEN};
@@ -42,6 +43,7 @@ const SPEAKER_TICK_MS: u64 = 500;
 const SPEAKER_LEVEL_TICK_MS: u64 = 50;
 /// Per-peer arrival window for the TTS interrupt frame counter.
 const FRAME_WINDOW: std::time::Duration = std::time::Duration::from_millis(500);
+const REMOTE_RELEASE_DEBOUNCE: std::time::Duration = std::time::Duration::from_millis(500);
 /// Playout clock: NetEq emits 10 ms frames, so we tick at 10 ms.
 const PLAYOUT_TICK_MS: u64 = 10;
 
@@ -84,6 +86,39 @@ fn normalized_speaker_level(level_dbov: i8) -> f32 {
     ((f32::from(level_dbov) + 60.0) / 48.0).clamp(0.0, 1.0)
 }
 
+fn update_remote_release_deadline(
+    peer: u8,
+    is_dtx: bool,
+    remote_floor_owners: &std::collections::HashSet<u8>,
+    deadlines: &mut std::collections::HashMap<u8, tokio::time::Instant>,
+    now: tokio::time::Instant,
+) {
+    if !is_dtx {
+        deadlines.remove(&peer);
+    } else if remote_floor_owners.contains(&peer) {
+        deadlines
+            .entry(peer)
+            .or_insert(now + REMOTE_RELEASE_DEBOUNCE);
+    }
+}
+
+fn release_expired_remote_floors(
+    now: tokio::time::Instant,
+    owners: &mut std::collections::HashSet<u8>,
+    deadlines: &mut std::collections::HashMap<u8, tokio::time::Instant>,
+    human_floor: &HumanFloor,
+) {
+    let released: Vec<u8> = deadlines
+        .iter()
+        .filter_map(|(peer, deadline)| (*deadline <= now).then_some(*peer))
+        .collect();
+    for peer in released {
+        deadlines.remove(&peer);
+        owners.remove(&peer);
+        human_floor.leave_remote(peer);
+    }
+}
+
 fn should_recover_playout(depth: usize, currently_recovering: bool) -> bool {
     if currently_recovering {
         depth > PLAYOUT_QUEUE_RECOVERY_END
@@ -171,6 +206,7 @@ pub(crate) async fn run_playout_recv_loop(
     initial_peers: Vec<(u8, String)>,
     tts_active: Arc<AtomicBool>,
     tts_cancel: Arc<AtomicBool>,
+    human_floor: HumanFloor,
 ) {
     use rodio::buffer::SamplesBuffer;
     use std::num::NonZero;
@@ -183,9 +219,11 @@ pub(crate) async fn run_playout_recv_loop(
         initial_peers.into_iter().collect();
     let mut active_indices: std::collections::HashSet<u8> = std::collections::HashSet::new();
     let mut speaker_levels: std::collections::HashMap<u8, f32> = std::collections::HashMap::new();
+    let mut remote_release_deadlines: std::collections::HashMap<u8, tokio::time::Instant> =
+        std::collections::HashMap::new();
+    let mut remote_floor_owners: std::collections::HashSet<u8> = std::collections::HashSet::new();
     let mut frame_counts: std::collections::HashMap<u8, u16> = std::collections::HashMap::new();
     let mut last_frame_reset = tokio::time::Instant::now();
-    let mut tts_was_active = false;
 
     let mut speaker_tick = tokio::time::interval(std::time::Duration::from_millis(SPEAKER_TICK_MS));
     speaker_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
@@ -247,6 +285,12 @@ pub(crate) async fn run_playout_recv_loop(
                 }
             }
             _ = speaker_tick.tick() => {
+                release_expired_remote_floors(
+                    tokio::time::Instant::now(),
+                    &mut remote_floor_owners,
+                    &mut remote_release_deadlines,
+                    &human_floor,
+                );
                 if let Some(ref app) = app_handle {
                     use tauri::Emitter;
                     let pubkeys: Vec<String> = active_indices
@@ -303,6 +347,13 @@ pub(crate) async fn run_playout_recv_loop(
                         // by an idle peer to keep the codec alive — they
                         // don't mean the peer is speaking, and shouldn't
                         // make their tile flash for the 500 ms speaker tick.
+                        update_remote_release_deadline(
+                            peer_idx,
+                            is_dtx,
+                            &remote_floor_owners,
+                            &mut remote_release_deadlines,
+                            tokio::time::Instant::now(),
+                        );
                         if !is_dtx {
                             active_indices.insert(peer_idx);
                             let level = normalized_speaker_level(header.level_dbov);
@@ -312,14 +363,9 @@ pub(crate) async fn run_playout_recv_loop(
                                 .or_insert(level);
                         }
 
-                        // TTS interrupt frame counter — reset on TTS rising edge.
-                        let tts_now = tts_active.load(Ordering::Acquire);
-                        if tts_now && !tts_was_active {
-                            frame_counts.clear();
-                            last_frame_reset = tokio::time::Instant::now();
-                        }
-                        tts_was_active = tts_now;
-
+                        // Track remote speech independently of TTS liveness so a
+                        // human who starts while output is idle still owns the
+                        // floor and rejects delayed synthesis.
                         let slot = match peers.entry(peer_idx) {
                             std::collections::hash_map::Entry::Occupied(e) => e.into_mut(),
                             std::collections::hash_map::Entry::Vacant(e) => {
@@ -347,11 +393,10 @@ pub(crate) async fn run_playout_recv_loop(
                             slot.last_packet_at = tokio::time::Instant::now();
                         }
 
-                        // Count remote-speech frame arrivals for the TTS
-                        // interrupt. DTX/comfort frames don't count — they
-                        // mean the peer is silent, just keeping the codec
-                        // state alive.
-                        if tts_now && !is_dtx {
+                        // Count remote-speech frame arrivals for floor onset.
+                        // DTX/comfort frames don't count — they mean the peer
+                        // is silent, just keeping the codec state alive.
+                        if !is_dtx {
                             if last_frame_reset.elapsed() >= FRAME_WINDOW {
                                 frame_counts.clear();
                                 last_frame_reset = tokio::time::Instant::now();
@@ -359,7 +404,11 @@ pub(crate) async fn run_playout_recv_loop(
                             let count = frame_counts.entry(peer_idx).or_insert(0);
                             *count = count.saturating_add(1);
                             if *count >= REMOTE_SPEECH_THRESHOLD {
-                                tts_cancel.store(true, Ordering::Release);
+                                human_floor.enter_remote(peer_idx);
+                                remote_floor_owners.insert(peer_idx);
+                                if tts_active.load(Ordering::Acquire) {
+                                    tts_cancel.store(true, Ordering::Release);
+                                }
                             }
                         }
                     }
@@ -384,6 +433,9 @@ pub(crate) async fn run_playout_recv_loop(
                                                 {
                                                     peers.remove(&key);
                                                     frame_counts.remove(&key);
+                                                    remote_release_deadlines.remove(&key);
+                                                    remote_floor_owners.remove(&key);
+                                                    human_floor.leave_remote(key);
                                                     active_indices.remove(&key);
                                                     speaker_levels.remove(&key);
                                                 }
@@ -407,6 +459,16 @@ pub(crate) async fn run_playout_recv_loop(
                                             replacement.get(idx) == index_to_pubkey.get(idx)
                                         };
                                         peers.retain(|idx, _| identity_unchanged(idx));
+                                        for idx in index_to_pubkey
+                                            .keys()
+                                            .filter(|idx| !identity_unchanged(idx))
+                                            .copied()
+                                            .collect::<Vec<_>>()
+                                        {
+                                            human_floor.leave_remote(idx);
+                                            remote_release_deadlines.remove(&idx);
+                                            remote_floor_owners.remove(&idx);
+                                        }
                                         frame_counts.retain(|idx, _| identity_unchanged(idx));
                                         active_indices.retain(identity_unchanged);
                                         speaker_levels.retain(|idx, _| identity_unchanged(idx));
@@ -418,6 +480,9 @@ pub(crate) async fn run_playout_recv_loop(
                                         let key = idx as u8;
                                         index_to_pubkey.remove(&key);
                                         frame_counts.remove(&key);
+                                        remote_release_deadlines.remove(&key);
+                                        remote_floor_owners.remove(&key);
+                                        human_floor.leave_remote(key);
                                         active_indices.remove(&key);
                                         speaker_levels.remove(&key);
                                         // Dropping Player detaches its queue from the
@@ -441,6 +506,7 @@ pub(crate) async fn run_playout_recv_loop(
         }
     }
 
+    human_floor.clear_remote();
     if let Some(ref app) = app_handle {
         use tauri::Emitter;
         let _ = app.emit(
@@ -454,6 +520,50 @@ pub(crate) async fn run_playout_recv_loop(
 mod tests {
     use super::*;
 
+    #[test]
+    fn continuous_dtx_does_not_extend_remote_floor_deadline() {
+        let peer = 7;
+        let started = tokio::time::Instant::now();
+        let owners = std::collections::HashSet::from([peer]);
+        let mut deadlines = std::collections::HashMap::new();
+
+        update_remote_release_deadline(peer, true, &owners, &mut deadlines, started);
+        let armed = deadlines[&peer];
+        for elapsed_ms in [100, 200, 300, 400] {
+            update_remote_release_deadline(
+                peer,
+                true,
+                &owners,
+                &mut deadlines,
+                started + std::time::Duration::from_millis(elapsed_ms),
+            );
+        }
+
+        assert_eq!(deadlines[&peer], armed);
+        assert!(armed <= started + REMOTE_RELEASE_DEBOUNCE);
+
+        let human_floor = HumanFloor::new();
+        human_floor.enter_remote(peer);
+        let mut owners = owners;
+        release_expired_remote_floors(armed, &mut owners, &mut deadlines, &human_floor);
+        assert!(!human_floor.is_blocked());
+        assert!(owners.is_empty());
+        assert!(deadlines.is_empty());
+    }
+
+    #[test]
+    fn dtx_from_non_owner_does_not_arm_remote_floor_deadline() {
+        let mut deadlines = std::collections::HashMap::new();
+        update_remote_release_deadline(
+            7,
+            true,
+            &std::collections::HashSet::new(),
+            &mut deadlines,
+            tokio::time::Instant::now(),
+        );
+        assert!(deadlines.is_empty());
+    }
+
     #[test]
     fn speaker_level_maps_conversational_range() {
         assert_eq!(normalized_speaker_level(-127), 0.0);

desktop/src-tauri/src/huddle/relay_api.rs

@@ -58,9 +58,13 @@ pub(crate) async fn connect_audio_relay(
     let keys = state.keys.lock().map_err(|e| e.to_string())?.clone();
 
     // TTS interrupt flags — recv task cancels TTS when remote humans speak.
-    let (tts_cancel, tts_active) = {
+    let (tts_cancel, tts_active, human_floor) = {
         let hs = state.huddle()?;
-        (Arc::clone(&hs.tts_cancel), Arc::clone(&hs.tts_active))
+        (
+            Arc::clone(&hs.tts_cancel),
+            Arc::clone(&hs.tts_active),
+            hs.human_floor.clone(),
+        )
     };
 
     let app_handle = state.app_handle.lock().ok().and_then(|g| g.clone());
@@ -180,6 +184,7 @@ pub(crate) async fn connect_audio_relay(
             initial_peers,
             tts_cancel,
             tts_active,
+            human_floor,
             output_device_name,
         })
         .await
@@ -214,6 +219,7 @@ struct AudioRelayPipelineArgs {
     initial_peers: Vec<(u8, String)>,
     tts_cancel: Arc<AtomicBool>,
     tts_active: Arc<AtomicBool>,
+    human_floor: super::human_floor::HumanFloor,
     output_device_name: Option<String>,
 }
 
@@ -227,6 +233,7 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String
         initial_peers,
         tts_cancel,
         tts_active,
+        human_floor,
         output_device_name,
     } = args;
 
@@ -336,6 +343,7 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String
         initial_peers,
         tts_active,
         tts_cancel,
+        human_floor,
     ));
 
     // Wait for either task to finish, then abort the survivor.

desktop/src-tauri/src/huddle/state.rs

@@ -11,6 +11,7 @@ use std::sync::{
 };
 
 use super::agent_voice::AgentVoiceSettings;
+use super::human_floor::HumanFloor;
 use super::{stt, tts};
 
 /// Voice input mode: push-to-talk (PTT) or voice-activity detection (VAD).
@@ -106,6 +107,10 @@ pub struct HuddleState {
     /// restarts — both STT and TTS reference the same flag for the entire huddle.
     #[serde(skip)]
     pub tts_cancel: Arc<AtomicBool>,
+    /// Shared human-floor state. Confirmed local or remote human speech hard
+    /// cancels TTS and blocks stale/new playback until every source releases.
+    #[serde(skip)]
+    pub human_floor: HumanFloor,
     /// Sentinel: true while a TTS pipeline is being constructed (outside the lock).
     /// Prevents TOCTOU races where two concurrent callers both pass the `is_some()`
     /// check and both spawn TTS worker threads — the loser's thread would leak.
@@ -189,6 +194,7 @@ impl Clone for HuddleState {
             transcription_user_controlled: self.transcription_user_controlled,
             tts_active: Arc::clone(&self.tts_active),
             tts_cancel: Arc::clone(&self.tts_cancel),
+            human_floor: self.human_floor.clone(),
             tts_starting: Arc::clone(&self.tts_starting),
             stt_starting: Arc::clone(&self.stt_starting),
             last_agent_refresh: self.last_agent_refresh,
@@ -203,6 +209,8 @@ impl Clone for HuddleState {
 
 impl Default for HuddleState {
     fn default() -> Self {
+        let tts_cancel = Arc::new(AtomicBool::new(false));
+        let human_floor = HumanFloor::new();
         Self {
             phase: HuddlePhase::Idle,
             parent_channel_id: None,
@@ -220,7 +228,8 @@ impl Default for HuddleState {
             transcription_enabled: false,
             transcription_user_controlled: false,
             tts_active: Arc::new(AtomicBool::new(false)),
-            tts_cancel: Arc::new(AtomicBool::new(false)),
+            tts_cancel,
+            human_floor,
             tts_starting: Arc::new(AtomicBool::new(false)),
             stt_starting: Arc::new(AtomicBool::new(false)),
             last_agent_refresh: None,

desktop/src-tauri/src/huddle/stt.rs

@@ -32,6 +32,8 @@ use std::{
 
 use tokio::sync::mpsc as tokio_mpsc;
 
+use super::{human_floor::HumanFloor, local_barge_in};
+
 // ── Public pipeline handle ────────────────────────────────────────────────────
 
 /// Bounded audio queue capacity.
@@ -62,11 +64,11 @@ pub struct SttPipeline {
 impl SttPipeline {
     /// Spawn the pipeline thread.
     ///
-    /// Mic input is transcribed even while agent TTS is playing: the huddle UI
-    /// already tells users to wear headphones, so speaker bleed is accepted in
-    /// exchange for never dropping human speech that overlaps agent audio.
-    /// Local mic frames still never cancel TTS — push-to-talk and remote
-    /// participant speech remain the explicit barge-in paths.
+    /// Mic input is transcribed even while agent TTS is playing. In open-mic
+    /// VAD mode, confirmed speech acquires the shared human floor: immediately
+    /// on an isolated output route, or after the restored 320 ms sustained-
+    /// speech debounce on an acoustically coupled route. Push-to-talk retains
+    /// its explicit shortcut cancellation path.
     ///
     /// `ptt_active` and `manual_mic_unmuted` are present when the PTT shortcut
     /// is enabled. The pipeline accepts speech while either input path is open;
@@ -85,6 +87,8 @@ impl SttPipeline {
         model_dir: PathBuf,
         ptt_active: Option<Arc<AtomicBool>>,
         manual_mic_unmuted: Option<Arc<AtomicBool>>,
+        human_floor: HumanFloor,
+        output_device: Option<String>,
     ) -> Result<(Self, tokio_mpsc::Receiver<String>), String> {
         let (audio_tx, audio_rx) = mpsc::sync_channel::<Vec<u8>>(AUDIO_QUEUE_DEPTH);
         let (text_tx, text_rx) = tokio_mpsc::channel::<String>(64);
@@ -103,6 +107,8 @@ impl SttPipeline {
                     shutdown_worker,
                     ptt_active_worker,
                     manual_mic_unmuted_worker,
+                    human_floor,
+                    output_device,
                 )
             })
             .map_err(|e| format!("failed to spawn stt-worker thread: {e}"))?;
@@ -205,6 +211,7 @@ const MIN_VOICED_FRAMES: usize = 12;
 #[derive(Debug, PartialEq, Eq)]
 enum VadFrameAction {
     None,
+    ConfirmedOnset,
     Speech,
     FirstSilence,
     Flush,
@@ -268,7 +275,7 @@ impl VadEndpoint {
             for buffered in self.pre_roll.drain(..) {
                 self.speech_buf.extend_from_slice(&buffered);
             }
-            return VadFrameAction::Speech;
+            return VadFrameAction::ConfirmedOnset;
         }
 
         if probability > VAD_OFFSET_THRESHOLD {
@@ -342,6 +349,45 @@ fn stt_speculative_decode() -> bool {
     std::env::var("BUZZ_STT_SPECULATIVE").is_ok_and(|v| v == "1")
 }
 
+#[derive(Debug)]
+enum SttLoopInput {
+    Tick,
+    Batch(Vec<Vec<u8>>),
+}
+
+fn run_stt_receive_loop(
+    audio_rx: Receiver<Vec<u8>>,
+    shutdown: &AtomicBool,
+    human_floor: HumanFloor,
+    mut process: impl FnMut(SttLoopInput, &mut local_barge_in::LocalBargeIn),
+) {
+    let mut local_barge_in_state = local_barge_in::WorkerLocalBargeIn::new(human_floor);
+
+    loop {
+        // Check shutdown flag before blocking.
+        if shutdown.load(Ordering::Acquire) {
+            break;
+        }
+
+        process(SttLoopInput::Tick, &mut local_barge_in_state);
+
+        // Use recv_timeout so we can periodically check the shutdown flag.
+        let bytes = match audio_rx.recv_timeout(RECV_TIMEOUT) {
+            Ok(b) => b,
+            Err(mpsc::RecvTimeoutError::Timeout) => continue,
+            Err(mpsc::RecvTimeoutError::Disconnected) => break, // Sender dropped.
+        };
+
+        // Drain any additional pending messages to batch-process.
+        let mut batch = vec![bytes];
+        while let Ok(b) = audio_rx.try_recv() {
+            batch.push(b);
+        }
+        process(SttLoopInput::Batch(batch), &mut local_barge_in_state);
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
 fn stt_worker(
     model_dir: PathBuf,
     audio_rx: Receiver<Vec<u8>>,
@@ -349,6 +395,8 @@ fn stt_worker(
     shutdown: Arc<AtomicBool>,
     ptt_active: Option<Arc<AtomicBool>>,
     manual_mic_unmuted: Option<Arc<AtomicBool>>,
+    human_floor: HumanFloor,
+    output_device: Option<String>,
 ) {
     // ── 1. Initialise rubato resampler (48 kHz → 16 kHz, mono) ───────────────
     use rubato::{Fft, FixedSync, Resampler};
@@ -416,7 +464,6 @@ fn stt_worker(
     // computed at. Valid only while no new voiced frame has arrived since.
     let speculative_enabled = stt_speculative_decode();
     let mut speculative: Option<(String, usize)> = None;
-
     // ── 5. Main loop ──────────────────────────────────────────────────────────
     let mut transmit_was_active = ptt_active
         .as_ref()
@@ -424,73 +471,69 @@ fn stt_worker(
         || manual_mic_unmuted
             .as_ref()
             .is_some_and(|manual| manual.load(Ordering::Acquire));
-    loop {
-        // Check shutdown flag before blocking.
-        if shutdown.load(Ordering::Acquire) {
-            break;
-        }
+    run_stt_receive_loop(
+        audio_rx,
+        &shutdown,
+        human_floor.clone(),
+        |input, local_barge_in_state| {
+            match input {
+                SttLoopInput::Tick => {
+                    // Track the combined manual/PTT transmission edge. When both paths
+                    // close, the worklet stops sending frames, so flush here rather than
+                    // waiting for silence that will never arrive.
+                    if let Some(ref ptt) = ptt_active {
+                        let transmit_now = ptt.load(Ordering::Acquire)
+                            || manual_mic_unmuted
+                                .as_ref()
+                                .is_some_and(|manual| manual.load(Ordering::Acquire));
+                        if transmit_was_active
+                            && !transmit_now
+                            && endpoint.in_speech
+                            && !endpoint.speech_buf.is_empty()
+                        {
+                            flush_to_stt(
+                                &endpoint.speech_buf,
+                                endpoint.voiced_frames,
+                                &recognizer,
+                                &text_tx,
+                            );
+                            endpoint.reset_segment();
+                            local_barge_in_state.release(&human_floor);
+                        }
+                        transmit_was_active = transmit_now;
+                    }
+                }
+                SttLoopInput::Batch(batch) => {
+                    for bytes in batch {
+                        // Convert raw bytes to f32 samples (little-endian).
+                        let samples_48k = bytes_to_f32(&bytes);
+                        input_buf_48k.extend_from_slice(&samples_48k);
 
-        // Track the combined manual/PTT transmission edge. When both paths
-        // close, the worklet stops sending frames, so flush here rather than
-        // waiting for silence that will never arrive.
-        if let Some(ref ptt) = ptt_active {
-            let transmit_now = ptt.load(Ordering::Acquire)
-                || manual_mic_unmuted
-                    .as_ref()
-                    .is_some_and(|manual| manual.load(Ordering::Acquire));
-            if transmit_was_active
-                && !transmit_now
-                && endpoint.in_speech
-                && !endpoint.speech_buf.is_empty()
-            {
-                flush_to_stt(
-                    &endpoint.speech_buf,
-                    endpoint.voiced_frames,
-                    &recognizer,
-                    &text_tx,
-                );
-                endpoint.reset_segment();
+                        // Resample in chunk_in-sized blocks.
+                        while input_buf_48k.len() >= chunk_in {
+                            let chunk: Vec<f32> = input_buf_48k.drain(..chunk_in).collect();
+                            let resampled = resample_chunk(&mut resampler, &chunk);
+                            process_16k_samples(
+                                &resampled,
+                                &mut leftover_16k,
+                                &mut vad,
+                                &mut endpoint,
+                                flush_frames,
+                                (speculative_enabled, &mut speculative),
+                                &recognizer,
+                                &text_tx,
+                                ptt_active.as_ref(),
+                                manual_mic_unmuted.as_ref(),
+                                &human_floor,
+                                local_barge_in_state,
+                                output_device.as_deref(),
+                            );
+                        }
+                    }
+                }
             }
-            transmit_was_active = transmit_now;
-        }
-
-        // Use recv_timeout so we can periodically check the shutdown flag.
-        let bytes = match audio_rx.recv_timeout(RECV_TIMEOUT) {
-            Ok(b) => b,
-            Err(mpsc::RecvTimeoutError::Timeout) => continue,
-            Err(mpsc::RecvTimeoutError::Disconnected) => break, // Sender dropped.
-        };
-
-        // Drain any additional pending messages to batch-process.
-        let mut batch = vec![bytes];
-        while let Ok(b) = audio_rx.try_recv() {
-            batch.push(b);
-        }
-
-        for bytes in batch {
-            // Convert raw bytes to f32 samples (little-endian).
-            let samples_48k = bytes_to_f32(&bytes);
-            input_buf_48k.extend_from_slice(&samples_48k);
-
-            // Resample in chunk_in-sized blocks.
-            while input_buf_48k.len() >= chunk_in {
-                let chunk: Vec<f32> = input_buf_48k.drain(..chunk_in).collect();
-                let resampled = resample_chunk(&mut resampler, &chunk);
-                process_16k_samples(
-                    &resampled,
-                    &mut leftover_16k,
-                    &mut vad,
-                    &mut endpoint,
-                    flush_frames,
-                    (speculative_enabled, &mut speculative),
-                    &recognizer,
-                    &text_tx,
-                    ptt_active.as_ref(),
-                    manual_mic_unmuted.as_ref(),
-                );
-            }
-        }
-    }
+        },
+    );
 
     // No final flush — leave_huddle/end_huddle emit lifecycle events before
     // the STT worker exits, so a final flush would post a kind:9 message AFTER
@@ -546,6 +589,9 @@ fn process_16k_samples(
     text_tx: &tokio_mpsc::Sender<String>,
     ptt_active: Option<&Arc<AtomicBool>>,
     manual_mic_unmuted: Option<&Arc<AtomicBool>>,
+    human_floor: &HumanFloor,
+    local_barge_in_state: &mut local_barge_in::LocalBargeIn,
+    output_device: Option<&str>,
 ) {
     let (speculative_enabled, speculative) = speculative;
     leftover.extend_from_slice(samples);
@@ -562,7 +608,27 @@ fn process_16k_samples(
         // VAD mode, or with a manually open mic once the shortcut is up.
         let flush_allowed = vad_flush_allowed(ptt_active.is_some(), manually_open, ptt_held);
 
-        match endpoint.process_frame(frame, prob, accepts_audio, flush_allowed, flush_frames) {
+        let action =
+            endpoint.process_frame(frame, prob, accepts_audio, flush_allowed, flush_frames);
+        // Open-mic VAD semantics also apply when a PTT-mode user manually
+        // opens the mic. A held shortcut keeps its explicit key-down cancel.
+        let local_barge_in = local_barge_in::enabled(ptt_active.is_some(), manually_open, ptt_held);
+        if local_barge_in {
+            local_barge_in_state.observe(
+                prob,
+                action == VadFrameAction::ConfirmedOnset,
+                human_floor,
+                output_device,
+                VAD_ONSET_THRESHOLD,
+            );
+        } else {
+            local_barge_in_state.release(human_floor);
+        }
+
+        match action {
+            VadFrameAction::ConfirmedOnset => {
+                speculative.take();
+            }
             VadFrameAction::Speech => {
                 // New voiced audio invalidates any speculative decode.
                 speculative.take();
@@ -594,6 +660,9 @@ fn process_16k_samples(
                     ),
                 }
                 endpoint.reset_segment();
+                if local_barge_in {
+                    local_barge_in_state.release(human_floor);
+                }
             }
             VadFrameAction::None => {}
         }
@@ -607,6 +676,9 @@ fn process_16k_samples(
                 text_tx,
             );
             endpoint.reset_segment();
+            if local_barge_in {
+                local_barge_in_state.release(human_floor);
+            }
             speculative.take();
         }
     }
@@ -686,200 +758,5 @@ fn bytes_to_f32(bytes: &[u8]) -> Vec<f32> {
 use super::drain_until_shutdown;
 
 #[cfg(test)]
-mod tests {
-    use super::{
-        has_enough_voiced_audio, vad_flush_allowed, VadEndpoint, VadFrameAction, MIN_VOICED_FRAMES,
-        SILENCE_FLUSH_FRAMES, VAD_FRAME_SAMPLES, VAD_ONSET_FRAMES, VAD_PRE_ROLL_FRAMES,
-    };
-
-    fn frame(value: f32) -> Vec<f32> {
-        vec![value; VAD_FRAME_SAMPLES]
-    }
-
-    #[test]
-    fn short_vad_blips_do_not_reach_the_recognizer() {
-        assert!(!has_enough_voiced_audio(1));
-        assert!(!has_enough_voiced_audio(MIN_VOICED_FRAMES - 1));
-        assert!(has_enough_voiced_audio(MIN_VOICED_FRAMES));
-    }
-
-    #[test]
-    fn confirmed_onset_prepends_pre_roll_once() {
-        let mut endpoint = VadEndpoint::new();
-        for value in 0..VAD_PRE_ROLL_FRAMES - VAD_ONSET_FRAMES {
-            assert_eq!(
-                endpoint.process_frame(frame(value as f32), 0.0, true, true, SILENCE_FLUSH_FRAMES),
-                VadFrameAction::None
-            );
-        }
-        for value in 0..VAD_ONSET_FRAMES {
-            let action = endpoint.process_frame(
-                frame(100.0 + value as f32),
-                0.9,
-                true,
-                true,
-                SILENCE_FLUSH_FRAMES,
-            );
-            if value + 1 == VAD_ONSET_FRAMES {
-                assert_eq!(action, VadFrameAction::Speech);
-            } else {
-                assert_eq!(action, VadFrameAction::None);
-            }
-        }
-
-        assert_eq!(
-            endpoint.speech_buf.len(),
-            VAD_PRE_ROLL_FRAMES * VAD_FRAME_SAMPLES
-        );
-        assert_eq!(endpoint.speech_buf[0], 0.0);
-        assert_eq!(endpoint.speech_buf[VAD_FRAME_SAMPLES], 1.0);
-        assert_eq!(endpoint.pre_roll.len(), 0);
-        endpoint.process_frame(frame(200.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
-        assert_eq!(
-            endpoint.speech_buf.len(),
-            (VAD_PRE_ROLL_FRAMES + 1) * VAD_FRAME_SAMPLES
-        );
-    }
-
-    #[test]
-    fn onset_requires_consecutive_high_frames() {
-        let mut endpoint = VadEndpoint::new();
-        for probability in [0.9, 0.9, 0.2, 0.9, 0.9] {
-            assert_eq!(
-                endpoint.process_frame(frame(1.0), probability, true, true, SILENCE_FLUSH_FRAMES),
-                VadFrameAction::None
-            );
-        }
-        assert_eq!(
-            endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES),
-            VadFrameAction::Speech
-        );
-    }
-
-    #[test]
-    fn offset_hysteresis_preserves_borderline_speech() {
-        let mut endpoint = VadEndpoint::new();
-        for _ in 0..VAD_ONSET_FRAMES {
-            endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
-        }
-        assert_eq!(
-            endpoint.process_frame(frame(2.0), 0.4, true, true, SILENCE_FLUSH_FRAMES),
-            VadFrameAction::Speech
-        );
-        assert_eq!(endpoint.silence_frames, 0);
-    }
-
-    #[test]
-    fn below_offset_threshold_starts_silence() {
-        let mut endpoint = VadEndpoint::new();
-        for _ in 0..VAD_ONSET_FRAMES {
-            endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
-        }
-        assert_eq!(
-            endpoint.process_frame(frame(0.0), 0.3, true, true, SILENCE_FLUSH_FRAMES),
-            VadFrameAction::FirstSilence
-        );
-        assert_eq!(endpoint.silence_frames, 1);
-    }
-
-    #[test]
-    fn short_segment_reaches_the_visible_drop_path() {
-        let mut endpoint = VadEndpoint::new();
-        for _ in 0..VAD_ONSET_FRAMES {
-            endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
-        }
-        let mut action = VadFrameAction::None;
-        for _ in 0..SILENCE_FLUSH_FRAMES {
-            action = endpoint.process_frame(frame(0.0), 0.0, true, true, SILENCE_FLUSH_FRAMES);
-        }
-        assert_eq!(action, VadFrameAction::Flush);
-        assert!(!has_enough_voiced_audio(endpoint.voiced_frames));
-        assert!(!endpoint.speech_buf.is_empty());
-    }
-
-    #[test]
-    fn silence_flush_retains_only_hangover_audio() {
-        let mut endpoint = VadEndpoint::new();
-        for _ in 0..VAD_ONSET_FRAMES {
-            endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
-        }
-        let speech_len = endpoint.speech_buf.len();
-        for index in 1..=SILENCE_FLUSH_FRAMES {
-            let action = endpoint.process_frame(frame(0.0), 0.0, true, true, SILENCE_FLUSH_FRAMES);
-            if index == SILENCE_FLUSH_FRAMES {
-                assert_eq!(action, VadFrameAction::Flush);
-            }
-        }
-        assert_eq!(
-            endpoint.speech_buf.len(),
-            speech_len + 6 * VAD_FRAME_SAMPLES
-        );
-    }
-
-    #[test]
-    fn flush_boundary_never_double_includes_audio() {
-        const SEGMENT_N_MARKER: f32 = 777.0;
-        let mut endpoint = VadEndpoint::new();
-        for _ in 0..VAD_ONSET_FRAMES {
-            endpoint.process_frame(
-                frame(SEGMENT_N_MARKER),
-                0.9,
-                true,
-                true,
-                SILENCE_FLUSH_FRAMES,
-            );
-        }
-        for _ in 0..SILENCE_FLUSH_FRAMES {
-            endpoint.process_frame(
-                frame(SEGMENT_N_MARKER),
-                0.0,
-                true,
-                true,
-                SILENCE_FLUSH_FRAMES,
-            );
-        }
-        endpoint.reset_segment();
-
-        for _ in 0..VAD_ONSET_FRAMES {
-            endpoint.process_frame(frame(2.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
-        }
-        let leaked = endpoint
-            .speech_buf
-            .iter()
-            .filter(|sample| **sample == SEGMENT_N_MARKER)
-            .count();
-        assert_eq!(leaked, 0, "segment N audio leaked into segment N+1");
-    }
-
-    #[test]
-    fn reset_prevents_pre_roll_from_leaking_between_segments() {
-        const SEGMENT_N_MARKER: f32 = 777.0;
-        let mut endpoint = VadEndpoint::new();
-        endpoint.pre_roll.push_back(frame(SEGMENT_N_MARKER));
-        endpoint.reset_segment();
-        for _ in 0..VAD_ONSET_FRAMES {
-            endpoint.process_frame(frame(2.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
-        }
-        let leaked = endpoint
-            .speech_buf
-            .iter()
-            .filter(|sample| **sample == SEGMENT_N_MARKER)
-            .count();
-        assert_eq!(leaked, 0, "segment N pre-roll leaked into segment N+1");
-    }
-
-    #[test]
-    fn held_push_to_talk_never_silence_flushes() {
-        // Pure VAD mode: silence always ends the utterance.
-        assert!(vad_flush_allowed(false, false, false));
-        // Shortcut configured, nothing transmitting: nothing to flush anyway,
-        // but the pause path stays closed.
-        assert!(!vad_flush_allowed(true, false, false));
-        // Shortcut held: "I am not done talking" — never flush on silence,
-        // regardless of the manual mic state.
-        assert!(!vad_flush_allowed(true, false, true));
-        assert!(!vad_flush_allowed(true, true, true));
-        // Manually open mic with the shortcut up: normal VAD behavior.
-        assert!(vad_flush_allowed(true, true, false));
-    }
-}
+#[path = "stt_tests.rs"]
+mod tests;

desktop/src-tauri/src/huddle/stt_tests.rs

@@ -0,0 +1,255 @@
+use std::sync::{atomic::AtomicBool, mpsc, Arc, Barrier};
+
+use super::{
+    has_enough_voiced_audio, run_stt_receive_loop, vad_flush_allowed, HumanFloor, SttLoopInput,
+    VadEndpoint, VadFrameAction, MIN_VOICED_FRAMES, SILENCE_FLUSH_FRAMES, VAD_FRAME_SAMPLES,
+    VAD_ONSET_FRAMES, VAD_PRE_ROLL_FRAMES,
+};
+
+#[derive(Clone, Copy)]
+enum WorkerExit {
+    Shutdown,
+    SenderDisconnect,
+}
+
+fn assert_worker_exit_releases_floor(exit: WorkerExit) {
+    let human_floor = HumanFloor::new();
+    let shutdown = Arc::new(AtomicBool::new(false));
+    let (audio_tx, audio_rx) = mpsc::channel();
+    let acquired = Arc::new(Barrier::new(2));
+    let worker_floor = human_floor.clone();
+    let worker_shutdown = Arc::clone(&shutdown);
+    let worker_acquired = Arc::clone(&acquired);
+    let worker = std::thread::spawn(move || {
+        run_stt_receive_loop(
+            audio_rx,
+            &worker_shutdown,
+            worker_floor.clone(),
+            |input, local_barge_in_state| {
+                if matches!(input, SttLoopInput::Batch(_)) && !worker_floor.is_blocked() {
+                    local_barge_in_state.acquire(&worker_floor, true, false);
+                    worker_acquired.wait();
+                }
+            },
+        );
+    });
+
+    audio_tx.send(Vec::new()).expect("worker receiver is open");
+    acquired.wait();
+    assert!(human_floor.is_blocked());
+    match exit {
+        WorkerExit::Shutdown => {
+            shutdown.store(true, std::sync::atomic::Ordering::Release);
+        }
+        WorkerExit::SenderDisconnect => drop(audio_tx),
+    }
+    worker.join().expect("worker exits cleanly");
+
+    let replacement_epoch = human_floor.epoch();
+    assert!(
+        human_floor.permits(replacement_epoch),
+        "fresh TTS authorization must proceed after worker exit"
+    );
+    assert!(human_floor.enter_local(true, false));
+}
+
+#[test]
+fn worker_shutdown_releases_local_floor_for_replacement() {
+    assert_worker_exit_releases_floor(WorkerExit::Shutdown);
+}
+
+#[test]
+fn worker_channel_disconnect_releases_local_floor_for_replacement() {
+    assert_worker_exit_releases_floor(WorkerExit::SenderDisconnect);
+}
+
+fn frame(value: f32) -> Vec<f32> {
+    vec![value; VAD_FRAME_SAMPLES]
+}
+
+#[test]
+fn short_vad_blips_do_not_reach_the_recognizer() {
+    assert!(!has_enough_voiced_audio(1));
+    assert!(!has_enough_voiced_audio(MIN_VOICED_FRAMES - 1));
+    assert!(has_enough_voiced_audio(MIN_VOICED_FRAMES));
+}
+
+#[test]
+fn confirmed_onset_prepends_pre_roll_once() {
+    let mut endpoint = VadEndpoint::new();
+    for value in 0..VAD_PRE_ROLL_FRAMES - VAD_ONSET_FRAMES {
+        assert_eq!(
+            endpoint.process_frame(frame(value as f32), 0.0, true, true, SILENCE_FLUSH_FRAMES),
+            VadFrameAction::None
+        );
+    }
+    for value in 0..VAD_ONSET_FRAMES {
+        let action = endpoint.process_frame(
+            frame(100.0 + value as f32),
+            0.9,
+            true,
+            true,
+            SILENCE_FLUSH_FRAMES,
+        );
+        if value + 1 == VAD_ONSET_FRAMES {
+            assert_eq!(action, VadFrameAction::ConfirmedOnset);
+        } else {
+            assert_eq!(action, VadFrameAction::None);
+        }
+    }
+
+    assert_eq!(
+        endpoint.speech_buf.len(),
+        VAD_PRE_ROLL_FRAMES * VAD_FRAME_SAMPLES
+    );
+    assert_eq!(endpoint.speech_buf[0], 0.0);
+    assert_eq!(endpoint.speech_buf[VAD_FRAME_SAMPLES], 1.0);
+    assert_eq!(endpoint.pre_roll.len(), 0);
+    endpoint.process_frame(frame(200.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
+    assert_eq!(
+        endpoint.speech_buf.len(),
+        (VAD_PRE_ROLL_FRAMES + 1) * VAD_FRAME_SAMPLES
+    );
+}
+
+#[test]
+fn onset_requires_consecutive_high_frames() {
+    let mut endpoint = VadEndpoint::new();
+    for probability in [0.9, 0.9, 0.2, 0.9, 0.9] {
+        assert_eq!(
+            endpoint.process_frame(frame(1.0), probability, true, true, SILENCE_FLUSH_FRAMES),
+            VadFrameAction::None
+        );
+    }
+    assert_eq!(
+        endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES),
+        VadFrameAction::ConfirmedOnset
+    );
+}
+
+#[test]
+fn offset_hysteresis_preserves_borderline_speech() {
+    let mut endpoint = VadEndpoint::new();
+    for _ in 0..VAD_ONSET_FRAMES {
+        endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
+    }
+    assert_eq!(
+        endpoint.process_frame(frame(2.0), 0.4, true, true, SILENCE_FLUSH_FRAMES),
+        VadFrameAction::Speech
+    );
+    assert_eq!(endpoint.silence_frames, 0);
+}
+
+#[test]
+fn below_offset_threshold_starts_silence() {
+    let mut endpoint = VadEndpoint::new();
+    for _ in 0..VAD_ONSET_FRAMES {
+        endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
+    }
+    assert_eq!(
+        endpoint.process_frame(frame(0.0), 0.3, true, true, SILENCE_FLUSH_FRAMES),
+        VadFrameAction::FirstSilence
+    );
+    assert_eq!(endpoint.silence_frames, 1);
+}
+
+#[test]
+fn short_segment_reaches_the_visible_drop_path() {
+    let mut endpoint = VadEndpoint::new();
+    for _ in 0..VAD_ONSET_FRAMES {
+        endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
+    }
+    let mut action = VadFrameAction::None;
+    for _ in 0..SILENCE_FLUSH_FRAMES {
+        action = endpoint.process_frame(frame(0.0), 0.0, true, true, SILENCE_FLUSH_FRAMES);
+    }
+    assert_eq!(action, VadFrameAction::Flush);
+    assert!(!has_enough_voiced_audio(endpoint.voiced_frames));
+    assert!(!endpoint.speech_buf.is_empty());
+}
+
+#[test]
+fn silence_flush_retains_only_hangover_audio() {
+    let mut endpoint = VadEndpoint::new();
+    for _ in 0..VAD_ONSET_FRAMES {
+        endpoint.process_frame(frame(1.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
+    }
+    let speech_len = endpoint.speech_buf.len();
+    for index in 1..=SILENCE_FLUSH_FRAMES {
+        let action = endpoint.process_frame(frame(0.0), 0.0, true, true, SILENCE_FLUSH_FRAMES);
+        if index == SILENCE_FLUSH_FRAMES {
+            assert_eq!(action, VadFrameAction::Flush);
+        }
+    }
+    assert_eq!(
+        endpoint.speech_buf.len(),
+        speech_len + 6 * VAD_FRAME_SAMPLES
+    );
+}
+
+#[test]
+fn flush_boundary_never_double_includes_audio() {
+    const SEGMENT_N_MARKER: f32 = 777.0;
+    let mut endpoint = VadEndpoint::new();
+    for _ in 0..VAD_ONSET_FRAMES {
+        endpoint.process_frame(
+            frame(SEGMENT_N_MARKER),
+            0.9,
+            true,
+            true,
+            SILENCE_FLUSH_FRAMES,
+        );
+    }
+    for _ in 0..SILENCE_FLUSH_FRAMES {
+        endpoint.process_frame(
+            frame(SEGMENT_N_MARKER),
+            0.0,
+            true,
+            true,
+            SILENCE_FLUSH_FRAMES,
+        );
+    }
+    endpoint.reset_segment();
+
+    for _ in 0..VAD_ONSET_FRAMES {
+        endpoint.process_frame(frame(2.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
+    }
+    let leaked = endpoint
+        .speech_buf
+        .iter()
+        .filter(|sample| **sample == SEGMENT_N_MARKER)
+        .count();
+    assert_eq!(leaked, 0, "segment N audio leaked into segment N+1");
+}
+
+#[test]
+fn reset_prevents_pre_roll_from_leaking_between_segments() {
+    const SEGMENT_N_MARKER: f32 = 777.0;
+    let mut endpoint = VadEndpoint::new();
+    endpoint.pre_roll.push_back(frame(SEGMENT_N_MARKER));
+    endpoint.reset_segment();
+    for _ in 0..VAD_ONSET_FRAMES {
+        endpoint.process_frame(frame(2.0), 0.9, true, true, SILENCE_FLUSH_FRAMES);
+    }
+    let leaked = endpoint
+        .speech_buf
+        .iter()
+        .filter(|sample| **sample == SEGMENT_N_MARKER)
+        .count();
+    assert_eq!(leaked, 0, "segment N pre-roll leaked into segment N+1");
+}
+
+#[test]
+fn held_push_to_talk_never_silence_flushes() {
+    // Pure VAD mode: silence always ends the utterance.
+    assert!(vad_flush_allowed(false, false, false));
+    // Shortcut configured, nothing transmitting: nothing to flush anyway,
+    // but the pause path stays closed.
+    assert!(!vad_flush_allowed(true, false, false));
+    // Shortcut held: "I am not done talking" — never flush on silence,
+    // regardless of the manual mic state.
+    assert!(!vad_flush_allowed(true, false, true));
+    assert!(!vad_flush_allowed(true, true, true));
+    // Manually open mic with the shortcut up: normal VAD behavior.
+    assert!(vad_flush_allowed(true, true, false));
+}

desktop/src-tauri/src/huddle/tts.rs

@@ -46,6 +46,7 @@ use std::{
     time::{Duration, Instant},
 };
 
+use super::human_floor::HumanFloor;
 use super::pocket::{
     load_text_to_speech, load_voice_style, DEFAULT_VOICE, SAMPLE_RATE, VOICE_FILE_EXT,
 };
@@ -53,10 +54,8 @@ use super::preprocessing::preprocess_for_tts;
 
 #[path = "tts_voice_transition.rs"]
 mod voice_transition;
+use super::tts_playback::*;
 use voice_transition::*;
-#[path = "tts_playback.rs"]
-mod playback;
-use playback::*;
 #[path = "tts_startup.rs"]
 mod startup;
 use startup::await_worker_startup;
@@ -153,6 +152,7 @@ pub struct TtsPipeline {
     /// Kept alive here so the Arc isn't dropped — the worker holds a clone.
     #[allow(dead_code)]
     cancel: Arc<AtomicBool>,
+    human_floor: HumanFloor,
     /// Internal cancellation used only for voice changes. Kept separate so a
     /// concurrent human barge-in always clears every queued message.
     voice_cancel: Arc<AtomicBool>,
@@ -185,6 +185,7 @@ impl TtsPipeline {
         model_dir: PathBuf,
         tts_active: Arc<AtomicBool>,
         cancel: Arc<AtomicBool>,
+        human_floor: HumanFloor,
         voice: &str,
         output_device: Option<String>,
         activity_app: Option<tauri::AppHandle>,
@@ -196,6 +197,7 @@ impl TtsPipeline {
 
         let shutdown_worker = Arc::clone(&shutdown);
         let cancel_worker = Arc::clone(&cancel);
+        let worker_human_floor = human_floor.clone();
         let voice_cancel = Arc::new(AtomicBool::new(false));
         let worker_voice_cancel = Arc::clone(&voice_cancel);
         let tts_active_worker = Arc::clone(&tts_active);
@@ -227,6 +229,7 @@ impl TtsPipeline {
                         worker_voice_change_ack,
                     ),
                     text_rx,
+                    worker_human_floor,
                     (
                         tts_active_worker,
                         shutdown_worker,
@@ -249,6 +252,7 @@ impl TtsPipeline {
             tts_active,
             shutdown,
             cancel,
+            human_floor,
             voice_cancel,
             voice,
             voice_generation,
@@ -275,10 +279,130 @@ impl Drop for TtsPipeline {
 
 // ── Worker thread ─────────────────────────────────────────────────────────────
 
+fn authorize_or_defer_queued_text(
+    human_floor: &HumanFloor,
+    deferred_text: &mut VecDeque<QueuedText>,
+    queued_text: QueuedText,
+) -> Result<QueuedText, HumanFloorAuthorization> {
+    match human_floor.authorization(queued_text.floor_epoch) {
+        HumanFloorAuthorization::Blocked => {
+            deferred_text.push_front(queued_text);
+            Err(HumanFloorAuthorization::Blocked)
+        }
+        HumanFloorAuthorization::Stale => {
+            eprintln!(
+                "buzz-desktop: tts stage=queue status=dropped reason=barge_in route_id={}",
+                queued_text.route_id
+            );
+            Err(HumanFloorAuthorization::Stale)
+        }
+        HumanFloorAuthorization::Permitted => Ok(queued_text),
+    }
+}
+
+struct TtsAppendContext<'a> {
+    playback: &'a PlaybackCoordinator,
+    #[cfg(test)]
+    human_floor: &'a HumanFloor,
+    cancel: &'a AtomicBool,
+    voice_cancel: &'a AtomicBool,
+    shutdown: &'a AtomicBool,
+    tts_active: &'a AtomicBool,
+    speaker_generations: &'a SpeakerGenerations,
+    active_speaker: &'a ActiveSpeaker,
+    activity_frames: &'a Mutex<VecDeque<TtsSpeakerActivityFrame>>,
+    channels: NonZero<u16>,
+    rate: NonZero<u32>,
+}
+
+fn append_worker_audio(
+    context: &TtsAppendContext<'_>,
+    prepared: PreparedModelAudio,
+    route_id: u64,
+    speaker_pubkey: Option<&str>,
+    speaker_generation: u64,
+    floor_epoch: u64,
+) -> bool {
+    // Keep the shared floor in this context so the regression can mutation-check
+    // that authorization never moves back inside the coordinator callback.
+    #[cfg(test)]
+    let _ = context.human_floor;
+    let sample_count = prepared.sample_count;
+    let chunk_index = prepared.chunk_index;
+    let activity = speaker_pubkey.map(|pubkey| {
+        build_tts_speaker_activity_frames(&prepared.buffer, pubkey, SAMPLE_RATE as usize)
+    });
+    let floor_authorization = context.playback.append_if_human_floor_permits(
+        rodio::buffer::SamplesBuffer::new(context.channels, context.rate, prepared.buffer),
+        floor_epoch,
+        |player_empty| {
+            if context.cancel.load(Ordering::Acquire)
+                || context.voice_cancel.load(Ordering::Acquire)
+                || context.shutdown.load(Ordering::Acquire)
+            {
+                let reason = if context.shutdown.load(Ordering::Acquire) {
+                    "shutdown"
+                } else if context.cancel.load(Ordering::Acquire) {
+                    "barge_in"
+                } else {
+                    "voice_switch"
+                };
+                eprintln!(
+                    "buzz-desktop: tts stage=synthesis status=cancelled reason={reason} route_id={route_id}"
+                );
+                return false;
+            }
+            if speaker_pubkey.is_some_and(|pubkey| {
+                current_speaker_generation(context.speaker_generations, pubkey)
+                    != speaker_generation
+            }) {
+                eprintln!(
+                    "buzz-desktop: tts stage=synthesis status=cancelled reason=speaker_removed route_id={route_id}"
+                );
+                return false;
+            }
+            if let Some(pubkey) = speaker_pubkey {
+                let mut active = context
+                    .active_speaker
+                    .lock()
+                    .unwrap_or_else(|error| error.into_inner());
+                if player_empty {
+                    active.take();
+                }
+                if active
+                    .as_deref()
+                    .is_some_and(|current| !current.eq_ignore_ascii_case(pubkey))
+                {
+                    return false;
+                }
+                active.get_or_insert_with(|| pubkey.to_ascii_lowercase());
+                context
+                    .activity_frames
+                    .lock()
+                    .unwrap_or_else(|error| error.into_inner())
+                    .extend(activity.unwrap_or_default());
+            }
+            true
+        },
+        // Publish activity under the coordinator: an append and the mic gate it
+        // implies are one transition, so cancellation cannot overwrite it.
+        || context.tts_active.store(true, Ordering::Release),
+    );
+    if floor_authorization != HumanFloorAuthorization::Permitted {
+        return false;
+    }
+    eprintln!(
+        "buzz-desktop: tts stage=player status=append_accepted route_id={route_id} chunk_index={chunk_index} sample_count={sample_count}"
+    );
+    true
+}
+
+#[allow(clippy::too_many_arguments)]
 fn tts_worker(
     model_dir: PathBuf,
     voice_state: WorkerVoiceState,
     text_rx: mpsc::Receiver<QueuedText>,
+    human_floor: HumanFloor,
     control_state: WorkerControlState,
     output_device: Option<String>,
     activity_app: Option<tauri::AppHandle>,
@@ -377,10 +501,11 @@ fn tts_worker(
         }
     };
 
-    // One coordinator owns the current Player and every operation on it.
-    // Cancellation replaces its queue while the output stream and mixer stay
-    // alive, preserving cross-item pipelining without exposing Player handles.
-    let playback = Arc::new(PlaybackCoordinator::new(sink_handle.mixer()));
+    // One coordinator owns the current Player, floor state, and every operation.
+    // It was allocated with the huddle state so onset and playback share the
+    // same serialization boundary even before the TTS worker starts.
+    let playback = human_floor.playback();
+    playback.bind_mixer(sink_handle.mixer());
     playback_probe.install(Arc::clone(&playback));
 
     // Prime the audio output stream with a short silent buffer.
@@ -442,76 +567,33 @@ fn tts_worker(
     let tts_streaming = streaming_emit_frames();
     let mut last_route_id = 0;
     let mut deferred_text = VecDeque::new();
+    let append_context = TtsAppendContext {
+        playback: &playback,
+        #[cfg(test)]
+        human_floor: &human_floor,
+        cancel: &cancel,
+        voice_cancel: &voice_cancel,
+        shutdown: &shutdown,
+        tts_active: &tts_active,
+        speaker_generations: &speaker_generations,
+        active_speaker: &active_speaker,
+        activity_frames: &activity_frames,
+        channels,
+        rate,
+    };
     let append_audio = |prepared: PreparedModelAudio,
                         route_id: u64,
                         speaker_pubkey: Option<&str>,
-                        speaker_generation: u64| {
-        let sample_count = prepared.sample_count;
-        let chunk_index = prepared.chunk_index;
-        let activity = speaker_pubkey.map(|pubkey| {
-            build_tts_speaker_activity_frames(&prepared.buffer, pubkey, SAMPLE_RATE as usize)
-        });
-        let accepted = playback.append_if(
-            SamplesBuffer::new(channels, rate, prepared.buffer),
-            |player_empty| {
-                if cancel.load(Ordering::Acquire)
-                    || voice_cancel.load(Ordering::Acquire)
-                    || shutdown.load(Ordering::Acquire)
-                {
-                    let reason = if shutdown.load(Ordering::Acquire) {
-                        "shutdown"
-                    } else if cancel.load(Ordering::Acquire) {
-                        "barge_in"
-                    } else {
-                        "voice_switch"
-                    };
-                    eprintln!(
-                        "buzz-desktop: tts stage=synthesis status=cancelled reason={reason} route_id={route_id}"
-                    );
-                    return false;
-                }
-                if speaker_pubkey.is_some_and(|pubkey| {
-                    current_speaker_generation(&speaker_generations, pubkey) != speaker_generation
-                }) {
-                    eprintln!(
-                        "buzz-desktop: tts stage=synthesis status=cancelled reason=speaker_removed route_id={route_id}"
-                    );
-                    return false;
-                }
-                if let Some(pubkey) = speaker_pubkey {
-                    let mut active = active_speaker
-                        .lock()
-                        .unwrap_or_else(|error| error.into_inner());
-                    if player_empty {
-                        active.take();
-                    }
-                    if active
-                        .as_deref()
-                        .is_some_and(|current| !current.eq_ignore_ascii_case(pubkey))
-                    {
-                        return false;
-                    }
-                    active.get_or_insert_with(|| pubkey.to_ascii_lowercase());
-                    activity_frames
-                        .lock()
-                        .unwrap_or_else(|error| error.into_inner())
-                        .extend(activity.unwrap_or_default());
-                }
-                true
-            },
-            // Publish activity under the coordinator: an append and the mic
-            // gate it implies are one transition, so a cancellation that
-            // replaces this player cannot have its release overwritten by a
-            // `true` landing after the fact.
-            || tts_active.store(true, Ordering::Release),
-        );
-        if !accepted {
-            return false;
-        }
-        eprintln!(
-            "buzz-desktop: tts stage=player status=append_accepted route_id={route_id} chunk_index={chunk_index} sample_count={sample_count}"
-        );
-        true
+                        speaker_generation: u64,
+                        floor_epoch: u64| {
+        append_worker_audio(
+            &append_context,
+            prepared,
+            route_id,
+            speaker_pubkey,
+            speaker_generation,
+            floor_epoch,
+        )
     };
 
     loop {
@@ -632,7 +714,19 @@ fn tts_worker(
             thread::sleep(RECV_TIMEOUT);
             continue;
         }
-        let requested_voice = queued_text.voice_reference.unwrap_or_else(|| {
+        let mut queued_text =
+            match authorize_or_defer_queued_text(&human_floor, &mut deferred_text, queued_text) {
+                Ok(queued_text) => queued_text,
+                Err(HumanFloorAuthorization::Blocked) => {
+                    thread::sleep(RECV_TIMEOUT);
+                    continue;
+                }
+                Err(HumanFloorAuthorization::Stale) => continue,
+                Err(HumanFloorAuthorization::Permitted) => {
+                    unreachable!("permitted text is returned")
+                }
+            };
+        let requested_voice = queued_text.voice_reference.take().unwrap_or_else(|| {
             selected_voice
                 .lock()
                 .unwrap_or_else(|error| error.into_inner())
@@ -641,6 +735,7 @@ fn tts_worker(
         let raw_text = queued_text.text;
         let speaker_pubkey = queued_text.speaker_pubkey;
         let speaker_generation = queued_text.speaker_generation;
+        let floor_epoch = queued_text.floor_epoch;
         let route_id = queued_text.route_id;
         eprintln!("buzz-desktop: tts stage=synthesis status=started route_id={route_id}");
 
@@ -751,6 +846,7 @@ fn tts_worker(
                             route_id,
                             speaker_pubkey.as_deref(),
                             speaker_generation,
+                            floor_epoch,
                         ) {
                             return false;
                         }
@@ -832,6 +928,7 @@ fn tts_worker(
                                 route_id,
                                 speaker_pubkey.as_deref(),
                                 speaker_generation,
+                                floor_epoch,
                             ) {
                                 synthesis_outcome = "cancelled";
                                 break 'playback_chunks;
@@ -860,6 +957,7 @@ fn tts_worker(
                     route_id,
                     speaker_pubkey.as_deref(),
                     speaker_generation,
+                    floor_epoch,
                 ) {
                     synthesis_outcome = "cancelled";
                     break 'playback_chunks;

desktop/src-tauri/src/huddle/tts_pipeline_controls.rs

@@ -6,9 +6,11 @@ impl TtsPipeline {
     /// Non-blocking. Returns `Err` if the queue is full (bounded at
     /// `TEXT_QUEUE_DEPTH`) — caller may log and discard.
     pub fn speak(&self, text: String) -> Result<(), String> {
+        let floor_epoch = self.human_floor.epoch();
         self.text_tx
             .try_send(QueuedText {
                 generation: self.voice_generation.load(Ordering::Acquire),
+                floor_epoch,
                 route_id: 0,
                 speaker_pubkey: None,
                 speaker_generation: 0,
@@ -28,6 +30,7 @@ impl TtsPipeline {
         TtsTextSender {
             text_tx: self.text_tx.clone(),
             generation: self.voice_generation.load(Ordering::Acquire),
+            human_floor: self.human_floor.clone(),
             speaker_generations: Arc::clone(&self.speaker_generations),
         }
     }

desktop/src-tauri/src/huddle/tts_playback.rs

@@ -1,7 +1,17 @@
-use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
+use std::{
+    collections::HashSet,
+    sync::{Arc, Mutex, MutexGuard, PoisonError},
+    time::{Duration, Instant},
+};
 
 use rodio::{mixer::Mixer, Player, Source};
 
+/// Conservative guard after rodio reports drained. Max measured about 12 ms
+/// of cancellation tail on current-main CoreAudio and about 1 ms after player
+/// replacement; 100 ms safely bounds those observed paths while the phase-1
+/// route matrix determines whether this can be narrowed.
+const OUTPUT_TAIL_HANGOVER: Duration = Duration::from_millis(100);
+
 /// Serializes every operation on the TTS player and owns the utterance-boundary
 /// bookkeeping that must change atomically when playback is replaced.
 ///
@@ -10,18 +20,62 @@ use rodio::{mixer::Mixer, Player, Source};
 /// independently valid at either value, and no mutable reference to the state
 /// leaves the locked operation that created it.
 pub(super) struct PlaybackCoordinator {
-    mixer: Mixer,
+    mixer: Mutex<Option<Mixer>>,
     state: Mutex<PlaybackState>,
 }
 
 struct PlaybackState {
-    player: Player,
+    player: Option<Player>,
     /// `true` while no append has been committed since the last utterance
     /// boundary. Only `append_if` clears it, so it records appends that were
     /// actually queued — never one the authorization refused.
     first_append: bool,
     synthesis_in_flight: bool,
     synthesis_generation: u64,
+    output_lease: OutputLease,
+    human_floor: HumanFloorState,
+}
+
+#[derive(Default)]
+enum OutputLease {
+    #[default]
+    Inactive,
+    Active,
+    HangoverUntil(Instant),
+}
+
+impl OutputLease {
+    fn is_live_at(&mut self, now: Instant) -> bool {
+        match self {
+            Self::Inactive => false,
+            Self::Active => true,
+            Self::HangoverUntil(deadline) if now < *deadline => true,
+            Self::HangoverUntil(_) => {
+                *self = Self::Inactive;
+                false
+            }
+        }
+    }
+
+    fn begin_hangover(&mut self, now: Instant) {
+        if !matches!(self, Self::Inactive) {
+            *self = Self::HangoverUntil(now + OUTPUT_TAIL_HANGOVER);
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum HumanFloorAuthorization {
+    Permitted,
+    Blocked,
+    Stale,
+}
+
+#[derive(Default)]
+struct HumanFloorState {
+    epoch: u64,
+    local: bool,
+    remote: HashSet<u8>,
 }
 
 pub(super) struct SynthesisFlightGuard {
@@ -39,18 +93,35 @@ impl Drop for SynthesisFlightGuard {
 }
 
 impl PlaybackCoordinator {
+    #[cfg(test)]
     pub(super) fn new(mixer: &Mixer) -> Self {
+        let coordinator = Self::unbound();
+        coordinator.bind_mixer(mixer);
+        coordinator
+    }
+
+    pub(super) fn unbound() -> Self {
         Self {
-            mixer: mixer.clone(),
+            mixer: Mutex::new(None),
             state: Mutex::new(PlaybackState {
-                player: Player::connect_new(mixer),
+                player: None,
                 first_append: true,
                 synthesis_in_flight: false,
                 synthesis_generation: 0,
+                output_lease: OutputLease::Inactive,
+                human_floor: HumanFloorState::default(),
             }),
         }
     }
 
+    pub(super) fn bind_mixer(&self, mixer: &Mixer) {
+        *self.mixer.lock().unwrap_or_else(PoisonError::into_inner) = Some(mixer.clone());
+        let mut state = self.lock();
+        if state.player.is_none() {
+            state.player = Some(Player::connect_new(mixer));
+        }
+    }
+
     fn lock(&self) -> MutexGuard<'_, PlaybackState> {
         self.state.lock().unwrap_or_else(PoisonError::into_inner)
     }
@@ -61,6 +132,7 @@ impl PlaybackCoordinator {
     /// concurrent cancellation either replaces the queue before this append is
     /// authorized, or observes the committed state after it — never lands its
     /// own release between the two and gets overwritten.
+    #[cfg(test)]
     pub(super) fn append_if<S>(
         &self,
         source: S,
@@ -71,11 +143,15 @@ impl PlaybackCoordinator {
         S: Source<Item = f32> + Send + 'static,
     {
         let mut state = self.lock();
-        if !authorize(state.player.empty()) {
+        if !authorize(state.player.as_ref().is_none_or(Player::empty)) {
             return false;
         }
-        state.player.append(source);
+        let Some(player) = state.player.as_ref() else {
+            return false;
+        };
+        player.append(source);
         state.first_append = false;
+        state.output_lease = OutputLease::Active;
         commit();
         true
     }
@@ -84,28 +160,31 @@ impl PlaybackCoordinator {
     where
         S: Source<Item = f32> + Send + 'static,
     {
-        self.lock().player.append(source);
+        if let Some(player) = self.lock().player.as_ref() {
+            player.append(source);
+        }
     }
 
     pub(super) fn empty(&self) -> bool {
-        self.lock().player.empty()
+        self.lock().player.as_ref().is_none_or(Player::empty)
     }
 
     /// Observe playback emptiness under the coordinator so the onset decision
     /// for the audio being built is serialized with append and cancellation.
     pub(super) fn prepare_audio<R>(&self, prepare: impl FnOnce(bool) -> R) -> R {
         let state = self.lock();
-        let empty = state.player.empty();
+        let empty = state.player.as_ref().is_none_or(Player::empty);
         prepare(empty)
     }
 
     pub(super) fn release_if_drained(&self, release: impl FnOnce()) -> bool {
         let mut state = self.lock();
-        if !state.player.empty() || state.first_append {
+        if !state.player.as_ref().is_none_or(Player::empty) || state.first_append {
             return false;
         }
         release();
         state.first_append = true;
+        state.output_lease.begin_hangover(Instant::now());
         true
     }
 
@@ -124,7 +203,7 @@ impl PlaybackCoordinator {
 
     pub(super) fn with_playback_live<R>(&self, observe: impl FnOnce(bool) -> R) -> R {
         let state = self.lock();
-        observe(!state.player.empty() || state.synthesis_in_flight)
+        observe(!state.player.as_ref().is_none_or(Player::empty) || state.synthesis_in_flight)
     }
 
     /// Replace live playback with a fresh queue, publishing the replacement
@@ -142,21 +221,162 @@ impl PlaybackCoordinator {
         authorize: impl FnOnce() -> bool,
         commit: impl FnOnce(),
     ) -> bool {
+        let replacement = self
+            .mixer
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .as_ref()
+            .map(Player::connect_new);
         let old_player = {
             let mut state = self.lock();
-            if (state.player.empty() && !state.synthesis_in_flight) || !authorize() {
+            if (state.player.as_ref().is_none_or(Player::empty) && !state.synthesis_in_flight)
+                || !authorize()
+            {
                 return false;
             }
             state.first_append = true;
             state.synthesis_in_flight = false;
             state.synthesis_generation = state.synthesis_generation.wrapping_add(1);
-            let old_player = std::mem::replace(&mut state.player, Player::connect_new(&self.mixer));
+            state.output_lease.begin_hangover(Instant::now());
+            let old_player = std::mem::replace(&mut state.player, replacement);
             commit();
             old_player
         };
         drop(old_player);
         true
     }
+
+    #[cfg(test)]
+    pub(super) fn human_floor_blocked(&self) -> bool {
+        let state = self.lock();
+        state.human_floor.local || !state.human_floor.remote.is_empty()
+    }
+
+    pub(super) fn human_floor_epoch(&self) -> u64 {
+        self.lock().human_floor.epoch
+    }
+
+    pub(super) fn human_floor_authorization(&self, epoch: u64) -> HumanFloorAuthorization {
+        Self::human_floor_authorization_locked(&self.lock(), epoch)
+    }
+
+    fn human_floor_authorization_locked(
+        state: &PlaybackState,
+        epoch: u64,
+    ) -> HumanFloorAuthorization {
+        if state.human_floor.local || !state.human_floor.remote.is_empty() {
+            HumanFloorAuthorization::Blocked
+        } else if state.human_floor.epoch != epoch {
+            HumanFloorAuthorization::Stale
+        } else {
+            HumanFloorAuthorization::Permitted
+        }
+    }
+
+    #[cfg(test)]
+    pub(super) fn human_floor_permits(&self, epoch: u64) -> bool {
+        self.human_floor_authorization(epoch) == HumanFloorAuthorization::Permitted
+    }
+
+    pub(super) fn append_if_human_floor_permits<S>(
+        &self,
+        source: S,
+        epoch: u64,
+        authorize: impl FnOnce(bool) -> bool,
+        commit: impl FnOnce(),
+    ) -> HumanFloorAuthorization
+    where
+        S: Source<Item = f32> + Send + 'static,
+    {
+        let mut state = self.lock();
+        let floor_authorization = Self::human_floor_authorization_locked(&state, epoch);
+        if floor_authorization != HumanFloorAuthorization::Permitted {
+            return floor_authorization;
+        }
+        if !authorize(state.player.as_ref().is_none_or(Player::empty)) {
+            return HumanFloorAuthorization::Stale;
+        }
+        let Some(player) = state.player.as_ref() else {
+            return HumanFloorAuthorization::Stale;
+        };
+        player.append(source);
+        state.first_append = false;
+        state.output_lease = OutputLease::Active;
+        commit();
+        HumanFloorAuthorization::Permitted
+    }
+
+    pub(super) fn enter_local_human_floor(
+        &self,
+        route_isolated: bool,
+        sustained_coupled_speech: bool,
+    ) -> bool {
+        let replacement = self
+            .mixer
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .as_ref()
+            .map(Player::connect_new);
+        let old_player = {
+            let mut state = self.lock();
+            let output_live =
+                state.synthesis_in_flight || state.output_lease.is_live_at(Instant::now());
+            if state.human_floor.local
+                || (output_live && !route_isolated && !sustained_coupled_speech)
+            {
+                return false;
+            }
+            state.human_floor.local = true;
+            Self::commit_human_floor_onset(&mut state, replacement)
+        };
+        drop(old_player);
+        true
+    }
+
+    pub(super) fn leave_local_human_floor(&self) {
+        self.lock().human_floor.local = false;
+    }
+
+    pub(super) fn enter_remote_human_floor(&self, peer: u8) {
+        self.enter_human_floor(|floor| floor.remote.insert(peer));
+    }
+
+    pub(super) fn leave_remote_human_floor(&self, peer: u8) {
+        self.lock().human_floor.remote.remove(&peer);
+    }
+
+    pub(super) fn clear_remote_human_floor(&self) {
+        self.lock().human_floor.remote.clear();
+    }
+
+    fn enter_human_floor(&self, enter: impl FnOnce(&mut HumanFloorState) -> bool) {
+        let replacement = self
+            .mixer
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .as_ref()
+            .map(Player::connect_new);
+        let old_player = {
+            let mut state = self.lock();
+            if !enter(&mut state.human_floor) {
+                return;
+            }
+            Self::commit_human_floor_onset(&mut state, replacement)
+        };
+        drop(old_player);
+    }
+
+    fn commit_human_floor_onset(
+        state: &mut PlaybackState,
+        replacement: Option<Player>,
+    ) -> Option<Player> {
+        state.human_floor.epoch = state.human_floor.epoch.wrapping_add(1);
+        state.first_append = true;
+        state.synthesis_in_flight = false;
+        state.synthesis_generation = state.synthesis_generation.wrapping_add(1);
+        state.output_lease.begin_hangover(Instant::now());
+        std::mem::replace(&mut state.player, replacement)
+    }
 }
 
 #[cfg(test)]
@@ -194,6 +414,185 @@ mod tests {
         );
     }
 
+    fn one_second_source() -> SamplesBuffer {
+        SamplesBuffer::new(
+            NonZero::new(1).expect("nonzero channels"),
+            NonZero::new(24_000).expect("nonzero rate"),
+            vec![0.25; 24_000],
+        )
+    }
+
+    #[test]
+    fn floor_authorized_append_does_not_reenter_the_coordinator_lock() {
+        let (playback, _unpulled_source) = coordinator();
+        let epoch = playback.human_floor_epoch();
+        let (completed_tx, completed_rx) = std::sync::mpsc::sync_channel(1);
+        let worker = thread::spawn(move || {
+            let authorization =
+                playback.append_if_human_floor_permits(one_second_source(), epoch, |_| true, || {});
+            completed_tx
+                .send(authorization)
+                .expect("completion receiver");
+        });
+
+        assert_eq!(
+            completed_rx
+                .recv_timeout(Duration::from_secs(1))
+                .expect("floor-authorized append must not deadlock"),
+            HumanFloorAuthorization::Permitted
+        );
+        worker.join().expect("append worker");
+    }
+
+    #[test]
+    fn text_queued_during_a_held_floor_is_permitted_after_release() {
+        let (playback, _unpulled_source) = coordinator();
+        assert!(playback.enter_local_human_floor(true, false));
+        let queued_epoch = playback.human_floor_epoch();
+
+        assert_eq!(
+            playback.human_floor_authorization(queued_epoch),
+            HumanFloorAuthorization::Blocked
+        );
+        playback.leave_local_human_floor();
+        assert_eq!(
+            playback.human_floor_authorization(queued_epoch),
+            HumanFloorAuthorization::Permitted
+        );
+        assert_eq!(
+            playback.append_if_human_floor_permits(
+                one_second_source(),
+                queued_epoch,
+                |_| true,
+                || {},
+            ),
+            HumanFloorAuthorization::Permitted
+        );
+    }
+
+    #[test]
+    fn human_onset_replaces_playback_and_invalidates_late_append() {
+        let (playback, _unpulled_source) = coordinator();
+        append_second(&playback);
+        let stale_epoch = playback.human_floor_epoch();
+
+        assert!(playback.enter_local_human_floor(true, false));
+
+        assert!(playback.empty());
+        assert!(playback.human_floor_blocked());
+        assert!(!playback.human_floor_permits(stale_epoch));
+        playback.leave_local_human_floor();
+        assert!(!playback.human_floor_blocked());
+        assert!(!playback.human_floor_permits(stale_epoch));
+    }
+
+    #[test]
+    fn coupled_local_onset_while_idle_blocks_delayed_tts() {
+        let (playback, _unpulled_source) = coordinator();
+        let delayed_tts_epoch = playback.human_floor_epoch();
+
+        assert!(playback.enter_local_human_floor(false, false));
+
+        assert!(playback.human_floor_blocked());
+        assert!(!playback.human_floor_permits(delayed_tts_epoch));
+    }
+
+    #[test]
+    fn coupled_local_onset_during_output_is_rejected_as_ambiguous_echo() {
+        let (playback, _unpulled_source) = coordinator();
+        append_second(&playback);
+        let epoch = playback.human_floor_epoch();
+
+        assert!(!playback.enter_local_human_floor(false, false));
+
+        assert!(!playback.human_floor_blocked());
+        assert!(playback.human_floor_permits(epoch));
+        assert!(!playback.empty());
+    }
+
+    #[test]
+    fn sustained_coupled_speech_overrides_live_output_suppression() {
+        let (playback, _unpulled_source) = coordinator();
+        append_second(&playback);
+        let stale_epoch = playback.human_floor_epoch();
+
+        assert!(playback.enter_local_human_floor(false, true));
+
+        assert!(playback.empty());
+        assert!(playback.human_floor_blocked());
+        assert!(!playback.human_floor_permits(stale_epoch));
+    }
+
+    #[test]
+    fn coupled_local_onset_during_output_tail_hangover_is_rejected() {
+        let (playback, mut source) = coordinator();
+        append_second(&playback);
+        while !playback.empty() {
+            assert!(
+                source.next().is_some(),
+                "the mixer source outlives the queue"
+            );
+        }
+        assert!(playback.release_if_drained(|| {}));
+
+        assert!(!playback.enter_local_human_floor(false, false));
+        assert!(!playback.human_floor_blocked());
+    }
+
+    #[test]
+    fn coupled_local_onset_after_output_tail_hangover_is_accepted() {
+        let (playback, mut source) = coordinator();
+        append_second(&playback);
+        while !playback.empty() {
+            assert!(
+                source.next().is_some(),
+                "the mixer source outlives the queue"
+            );
+        }
+        assert!(playback.release_if_drained(|| {}));
+        playback.lock().output_lease =
+            OutputLease::HangoverUntil(Instant::now() - Duration::from_millis(1));
+
+        assert!(playback.enter_local_human_floor(false, false));
+        assert!(playback.human_floor_blocked());
+    }
+
+    #[test]
+    fn accepted_append_renews_an_expiring_output_lease() {
+        let (playback, _unpulled_source) = coordinator();
+        playback.lock().output_lease =
+            OutputLease::HangoverUntil(Instant::now() + Duration::from_millis(1));
+
+        append_second(&playback);
+
+        assert!(matches!(playback.lock().output_lease, OutputLease::Active));
+    }
+
+    #[test]
+    fn remote_onset_while_idle_blocks_delayed_tts() {
+        let (playback, _unpulled_source) = coordinator();
+        let delayed_tts_epoch = playback.human_floor_epoch();
+
+        playback.enter_remote_human_floor(7);
+
+        assert!(playback.human_floor_blocked());
+        assert!(!playback.human_floor_permits(delayed_tts_epoch));
+    }
+
+    #[test]
+    fn local_and_remote_sources_hold_the_same_floor_until_each_releases() {
+        let (playback, _unpulled_source) = coordinator();
+        assert!(playback.enter_local_human_floor(true, false));
+        let local_epoch = playback.human_floor_epoch();
+        playback.enter_remote_human_floor(7);
+        assert_ne!(playback.human_floor_epoch(), local_epoch);
+
+        playback.leave_local_human_floor();
+        assert!(playback.human_floor_blocked());
+        playback.leave_remote_human_floor(7);
+        assert!(!playback.human_floor_blocked());
+    }
+
     #[test]
     fn cancel_replaces_playback_without_waiting_for_the_mixer() {
         let (playback, _unpulled_source) = coordinator();

desktop/src-tauri/src/huddle/tts_settings.rs

@@ -622,6 +622,7 @@ pub async fn preview_pocket_voice(
             model_dir,
             active.clone(),
             cancel,
+            super::human_floor::HumanFloor::new(),
             &voice_name,
             output_device,
             None,

desktop/src-tauri/src/huddle/tts_tests.rs

@@ -12,6 +12,115 @@ use std::sync::{Arc, Mutex};
 #[path = "tts_tests/token_split.rs"]
 mod token_split;
 
+// ── Human-floor queue authorization ───────────────────────────────────────
+
+fn queued_text(route_id: u64, floor_epoch: u64) -> QueuedText {
+    QueuedText {
+        generation: 1,
+        floor_epoch,
+        route_id,
+        speaker_pubkey: None,
+        speaker_generation: 0,
+        voice_reference: None,
+        text: "queued while a human is speaking".to_string(),
+    }
+}
+
+#[test]
+fn production_worker_append_authorization_completes() {
+    let (completed_tx, completed_rx) = mpsc::sync_channel(1);
+    let worker = std::thread::spawn(move || {
+        let human_floor = HumanFloor::new();
+        let playback = human_floor.playback();
+        let channels = NonZero::new(1).expect("nonzero channels");
+        let rate = NonZero::new(SAMPLE_RATE).expect("nonzero rate");
+        let (mixer, _unpulled_source) = rodio::mixer::mixer(channels, rate);
+        playback.bind_mixer(&mixer);
+        let floor_epoch = human_floor.epoch();
+        let cancel = AtomicBool::new(false);
+        let voice_cancel = AtomicBool::new(false);
+        let shutdown = AtomicBool::new(false);
+        let tts_active = AtomicBool::new(false);
+        let speaker_generations = Arc::new(Mutex::new(HashMap::new()));
+        let active_speaker = Arc::new(Mutex::new(None));
+        let activity_frames = Mutex::new(VecDeque::new());
+        let context = TtsAppendContext {
+            playback: &playback,
+            human_floor: &human_floor,
+            cancel: &cancel,
+            voice_cancel: &voice_cancel,
+            shutdown: &shutdown,
+            tts_active: &tts_active,
+            speaker_generations: &speaker_generations,
+            active_speaker: &active_speaker,
+            activity_frames: &activity_frames,
+            channels,
+            rate,
+        };
+
+        let accepted = append_worker_audio(
+            &context,
+            PreparedModelAudio {
+                buffer: vec![0.25; SAMPLE_RATE as usize],
+                sample_count: SAMPLE_RATE as usize,
+                chunk_index: 0,
+            },
+            40,
+            None,
+            0,
+            floor_epoch,
+        );
+        completed_tx
+            .send((accepted, tts_active.load(Ordering::Acquire)))
+            .expect("completion receiver");
+    });
+
+    assert_eq!(
+        completed_rx
+            .recv_timeout(std::time::Duration::from_secs(1))
+            .expect("production worker append authorization must not deadlock"),
+        (true, true)
+    );
+    worker.join().expect("append worker");
+}
+
+#[test]
+fn worker_queue_defers_text_while_floor_is_held_then_releases_it() {
+    let human_floor = HumanFloor::new();
+    assert!(human_floor.enter_local(true, false));
+    let floor_epoch = human_floor.epoch();
+    let mut deferred = VecDeque::new();
+
+    assert!(matches!(
+        authorize_or_defer_queued_text(&human_floor, &mut deferred, queued_text(41, floor_epoch),),
+        Err(HumanFloorAuthorization::Blocked)
+    ));
+    assert_eq!(deferred.len(), 1, "held-floor text must stay queued");
+
+    human_floor.leave_local();
+    let queued = deferred.pop_front().expect("deferred text");
+    let released = authorize_or_defer_queued_text(&human_floor, &mut deferred, queued)
+        .expect("the same queue item is eligible after floor release");
+
+    assert_eq!(released.route_id, 41);
+    assert!(deferred.is_empty());
+}
+
+#[test]
+fn worker_queue_drops_text_from_before_human_onset() {
+    let human_floor = HumanFloor::new();
+    let stale_epoch = human_floor.epoch();
+    assert!(human_floor.enter_local(true, false));
+    human_floor.leave_local();
+    let mut deferred = VecDeque::new();
+
+    assert!(matches!(
+        authorize_or_defer_queued_text(&human_floor, &mut deferred, queued_text(42, stale_epoch),),
+        Err(HumanFloorAuthorization::Stale)
+    ));
+    assert!(deferred.is_empty(), "pre-barge-in text must not replay");
+}
+
 // ── Remote interrupt tracker ──────────────────────────────────────────────
 //
 // Models the per-peer frame counting logic in the recv task of

desktop/src-tauri/src/huddle/tts_voice_selection_tests.rs

@@ -16,6 +16,7 @@ fn inert_pipeline(cancel: Arc<AtomicBool>) -> TtsPipeline {
         tts_active: Arc::new(AtomicBool::new(false)),
         shutdown,
         cancel,
+        human_floor: HumanFloor::new(),
         voice_cancel: Arc::new(AtomicBool::new(false)),
         voice: Arc::new(std::sync::Mutex::new("reference_sample".to_string())),
         voice_generation: Arc::new(AtomicU64::new(1)),
@@ -169,6 +170,7 @@ fn an_in_hand_post_change_message_survives_cancellation() {
     ));
     text_tx
         .send(QueuedText {
+            floor_epoch: 0,
             generation: voice_generation.load(Ordering::Acquire),
             route_id: 1,
             speaker_pubkey: None,
@@ -183,6 +185,7 @@ fn an_in_hand_post_change_message_survives_cancellation() {
     let active = AtomicBool::new(true);
     let mut deferred_text = VecDeque::from([
         QueuedText {
+            floor_epoch: 0,
             generation: 1,
             route_id: 2,
             speaker_pubkey: None,
@@ -191,6 +194,7 @@ fn an_in_hand_post_change_message_survives_cancellation() {
             text: "old message".to_string(),
         },
         QueuedText {
+            floor_epoch: 0,
             generation: voice_generation.load(Ordering::Acquire),
             route_id: 3,
             speaker_pubkey: None,
@@ -250,6 +254,7 @@ fn superseding_voice_change_removes_earlier_deferred_messages() {
     )
     .expect("first voice change");
     deferred_text.push_back(QueuedText {
+        floor_epoch: 0,
         generation: voice_generation.load(Ordering::Acquire),
         route_id: 4,
         speaker_pubkey: None,
@@ -299,6 +304,7 @@ fn barge_in_clears_deferred_voice_change_messages() {
     let voice_change_ack = Arc::new(std::sync::Mutex::new(None));
     let (_text_tx, text_rx) = std::sync::mpsc::channel();
     let mut deferred_text = VecDeque::from([QueuedText {
+        floor_epoch: 0,
         generation: 2,
         route_id: 5,
         speaker_pubkey: None,
@@ -343,6 +349,7 @@ fn barge_in_during_a_voice_change_clears_post_change_messages() {
     )
     .expect("voice change");
     deferred_text.push_back(QueuedText {
+        floor_epoch: 0,
         generation: voice_generation.load(Ordering::Acquire),
         route_id: 6,
         speaker_pubkey: None,
@@ -375,6 +382,7 @@ fn a_sender_captured_before_voice_change_is_stale_even_if_it_sends_after_drain()
     let old_sender = TtsTextSender {
         text_tx,
         generation: voice_generation.load(Ordering::Acquire),
+        human_floor: HumanFloor::new(),
         speaker_generations: Arc::new(std::sync::Mutex::new(HashMap::new())),
     };
     let shutdown = AtomicBool::new(false);

desktop/src-tauri/src/huddle/tts_voice_transition.rs

@@ -9,7 +9,7 @@ use std::{
     },
 };
 
-use super::{PlaybackCoordinator, SynthesisFlightGuard};
+use super::{HumanFloor, PlaybackCoordinator, SynthesisFlightGuard};
 
 use crate::huddle::pocket::{load_voice_style, VoiceStyle, DEFAULT_VOICE, VOICE_FILE_EXT};
 
@@ -74,6 +74,7 @@ impl fmt::Debug for PlaybackProbe {
 #[derive(Debug)]
 pub(super) struct QueuedText {
     pub(super) generation: u64,
+    pub(super) floor_epoch: u64,
     pub(super) route_id: u64,
     pub(super) speaker_pubkey: Option<String>,
     pub(super) speaker_generation: u64,
@@ -85,6 +86,7 @@ pub(super) struct QueuedText {
 pub(crate) struct TtsTextSender {
     pub(super) text_tx: SyncSender<QueuedText>,
     pub(super) generation: u64,
+    pub(super) human_floor: HumanFloor,
     pub(super) speaker_generations: SpeakerGenerations,
 }
 
@@ -97,9 +99,11 @@ impl TtsTextSender {
         voice_reference: String,
         text: String,
     ) -> Result<(), String> {
+        let floor_epoch = self.human_floor.epoch();
         self.text_tx
             .send(QueuedText {
                 generation: self.generation,
+                floor_epoch,
                 route_id,
                 speaker_pubkey: Some(speaker_pubkey),
                 speaker_generation,
@@ -544,6 +548,7 @@ mod speaker_generation_tests {
 
     fn queued_speech(speaker_pubkey: &str, speaker_generation: u64) -> QueuedText {
         QueuedText {
+            floor_epoch: 0,
             generation: 1,
             route_id: 1,
             speaker_pubkey: Some(speaker_pubkey.to_string()),