fix(huddle): serialize barge-in monitor clears with worker player ops

Max's PR #997 review blocker: the monitor's check-then-clear was not
ordered against the worker's cancel consumption. Sequence: monitor loads
cancel=true → preempted → worker consumes cancel and appends a fresh
post-cancel utterance → monitor resumes its stale branch and clears the
fresh audio (and drops tts_active while it plays, un-gating echo).

Fix (Max's suggested shape): a player_ops mutex serializes all Player
mutations — monitor clear, worker cancel/shutdown clear, worker append —
and the monitor re-checks cancel while holding it. Either the clear runs
before fresh audio can be appended, or it observes cancel=false and
no-ops. Cancel consumption moved under the same lock so the false is
visible to the monitor's under-lock re-check.

The worker's post-synth stale-sentence check now also runs under the
lock together with its append, closing the symmetric window where the
monitor clears between the worker's check passing and the buffer
landing.

Lock is uncontended except during an actual barge-in (worker holds it
only for appends/clears, never across synth), so the hot path and the
~15 ms flag-to-silence are unchanged. Lock acquisition recovers from
poison — data is (), nothing inconsistent to observe — so a panicked
peer can't wedge the other thread.

New regression test models the exact stale-branch interleaving;
monitor-contract tests updated to the locked shape. 521 tests green.

Co-authored-by: Tyler Longwell <tlongwell@squareup.com>
Signed-off-by: Tyler Longwell <tlongwell@squareup.com>

Author: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d

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

@@ -18,7 +18,10 @@
 //!   → cancel flag: a 10 ms barge-in monitor thread silences the player and
 //!     releases tts_active on the flag's rising edge (~15 ms flag-to-silence,
 //!     even mid-sentence while the worker is blocked in synth_chunk); the
-//!     worker then consumes the flag — drain queue + clear + play (un-pause)
+//!     worker then consumes the flag — drain queue + clear + play (un-pause).
+//!     Monitor clears and worker player mutations are serialized through the
+//!     `player_ops` mutex, with the flag re-checked under the lock — see the
+//!     monitor block in `tts_worker` for the race this closes.
 //! ```
 //!
 //! Lookahead pipelining spans *items*, not just sentences within one item:
@@ -37,7 +40,7 @@ use std::{
     sync::{
         atomic::{AtomicBool, Ordering},
         mpsc::{self, SyncSender},
-        Arc,
+        Arc, Mutex, MutexGuard, PoisonError,
     },
     thread,
     time::Duration,
@@ -364,23 +367,42 @@ fn tts_worker(
     // monitor keeps re-clearing until the worker catches up, which also
     // covers a sentence appended in the race window after the worker's own
     // post-synthesis cancel check.
+    //
+    // `player_ops` closes the converse race (found in review): the monitor
+    // loads `cancel == true`, is preempted, the worker consumes the cancel
+    // and appends a fresh post-cancel utterance, then the monitor resumes
+    // from its stale branch and deletes audio that was meant to play. All
+    // worker player mutations (appends and cancel/shutdown clears) hold this
+    // lock, and the monitor re-checks `cancel` *while holding it* — so its
+    // clear either runs before fresh audio can be appended, or observes
+    // `cancel == false` and no-ops. The lock is uncontended except during an
+    // actual barge-in, so the hot path is unaffected.
+    let player_ops = Arc::new(Mutex::new(()));
     let monitor_stop = Arc::new(AtomicBool::new(false));
     let monitor = {
         let player = Arc::clone(&player);
         let cancel = Arc::clone(&cancel);
         let tts_active = Arc::clone(&tts_active);
         let stop = Arc::clone(&monitor_stop);
+        let player_ops = Arc::clone(&player_ops);
         thread::Builder::new()
             .name("tts-barge-in-monitor".into())
             .spawn(move || {
                 while !stop.load(Ordering::Acquire) {
                     if cancel.load(Ordering::Acquire) {
-                        // clear() pauses the persistent player; play() un-pauses
-                        // (see handle_cancel_or_shutdown). Idempotent — safe to
-                        // repeat every tick until the worker consumes the flag.
-                        player.clear();
-                        player.play();
-                        tts_active.store(false, Ordering::Release);
+                        let _ops = lock_player_ops(&player_ops);
+                        // Re-check under the lock: the worker may have
+                        // consumed this cancel (and appended fresh audio)
+                        // between the load above and the lock acquisition.
+                        if cancel.load(Ordering::Acquire) {
+                            // clear() pauses the persistent player; play()
+                            // un-pauses (see handle_cancel_or_shutdown).
+                            // Idempotent — safe to repeat every tick until
+                            // the worker consumes the flag.
+                            player.clear();
+                            player.play();
+                            tts_active.store(false, Ordering::Release);
+                        }
                     }
                     thread::sleep(MONITOR_TICK);
                 }
@@ -408,7 +430,13 @@ fn tts_worker(
     let mut first_append = true;
 
     loop {
-        if handle_cancel_or_shutdown(&cancel, &shutdown, &tts_active, &text_rx, Some(&player)) {
+        if handle_cancel_or_shutdown(
+            &cancel,
+            &shutdown,
+            &tts_active,
+            &text_rx,
+            Some((&player, &player_ops)),
+        ) {
             if shutdown.load(Ordering::Acquire) {
                 break;
             }
@@ -435,7 +463,13 @@ fn tts_worker(
 
         // Check cancel again after unblocking — a cancel may have arrived
         // while we were waiting.
-        if handle_cancel_or_shutdown(&cancel, &shutdown, &tts_active, &text_rx, Some(&player)) {
+        if handle_cancel_or_shutdown(
+            &cancel,
+            &shutdown,
+            &tts_active,
+            &text_rx,
+            Some((&player, &player_ops)),
+        ) {
             if shutdown.load(Ordering::Acquire) {
                 break;
             }
@@ -471,7 +505,13 @@ fn tts_worker(
             .collect();
 
         for sentence in &sentences {
-            if handle_cancel_or_shutdown(&cancel, &shutdown, &tts_active, &text_rx, Some(&player)) {
+            if handle_cancel_or_shutdown(
+                &cancel,
+                &shutdown,
+                &tts_active,
+                &text_rx,
+                Some((&player, &player_ops)),
+            ) {
                 first_append = true;
                 break;
             }
@@ -483,16 +523,6 @@ fn tts_worker(
 
             match engine.synth_chunk(text, "en", &style, SYNTH_STEPS, SYNTH_SPEED) {
                 Ok(samples) if !samples.is_empty() => {
-                    // A barge-in may have arrived during synthesis (the
-                    // blocking window the monitor thread exists for). Don't
-                    // append the now-stale sentence — the human interrupted;
-                    // speaking it anyway would talk over them. The flag is
-                    // deliberately NOT consumed here: the loop-top
-                    // handle_cancel_or_shutdown does the full consume
-                    // (drain queue, reset lead-in) on the next iteration.
-                    if cancel.load(Ordering::Acquire) {
-                        break;
-                    }
                     let mut boosted = apply_playback_gain(samples);
                     // Fade-out only — fading-in would attenuate the consonant
                     // onset (see `apply_fade_out` docstring + the
@@ -506,6 +536,25 @@ fn tts_worker(
                     // every chunk a quiet device warm-up window.
                     let buf =
                         build_sentence_append_buffer(&mut first_append, boosted, silence_buf_len);
+
+                    // Check-and-append under `player_ops`, serialized with
+                    // the monitor: a barge-in may have arrived during
+                    // synthesis (the blocking window the monitor thread
+                    // exists for). Don't append the now-stale sentence — the
+                    // human interrupted; speaking it anyway would talk over
+                    // them. Holding the lock for the check + append means the
+                    // monitor can never clear between our check passing and
+                    // the buffer landing. The flag is deliberately NOT
+                    // consumed here: the loop-top handle_cancel_or_shutdown
+                    // does the full consume (drain queue, reset lead-in) on
+                    // the next iteration.
+                    let _ops = lock_player_ops(&player_ops);
+                    if cancel.load(Ordering::Acquire) {
+                        // Nothing appended; the loop-top consume re-arms
+                        // `first_append` (the flag is still set — the worker
+                        // is its only consumer).
+                        break;
+                    }
                     player.append(SamplesBuffer::new(channels, rate, buf));
                     // NOTE: tts_active is set AFTER player.append(), not
                     // before. Setting it before synthesis would cause STT to
@@ -540,38 +589,61 @@ fn tts_worker(
 
 /// Check for cancel or shutdown. Returns `true` if the caller should break/continue.
 /// On cancel: drains the text queue and clears the cancel flag.
+///
+/// `player` pairs the Player with the `player_ops` mutex shared with the
+/// barge-in monitor thread; the cancel/shutdown clear runs under that lock so
+/// it is serialized with the monitor's stale-branch re-check (see the monitor
+/// block in `tts_worker`).
 fn handle_cancel_or_shutdown(
     cancel: &AtomicBool,
     shutdown: &AtomicBool,
     tts_active: &AtomicBool,
     text_rx: &mpsc::Receiver<String>,
-    player: Option<&rodio::Player>,
+    player: Option<(&rodio::Player, &Mutex<()>)>,
 ) -> bool {
     if shutdown.load(Ordering::Acquire) {
-        if let Some(p) = player {
+        if let Some((p, ops)) = player {
+            let _ops = lock_player_ops(ops);
             p.clear();
         }
         tts_active.store(false, Ordering::Release);
         return true;
     }
     if cancel.load(Ordering::Acquire) {
-        if let Some(p) = player {
+        if let Some((p, ops)) = player {
+            let _ops = lock_player_ops(ops);
             // `Player::clear()` removes queued sources AND pauses the player
             // (rodio 0.22 `clear()` ends with `self.pause()`). With one
             // persistent Player for the worker's lifetime, the un-pause is
             // mandatory: without `play()`, every append after a barge-in
             // would queue silently forever.
             p.clear();
             p.play();
+            // Consume the flag under the lock: once released with
+            // `cancel == false`, the monitor's stale branch no-ops instead
+            // of clearing the fresh post-cancel utterance.
+            while text_rx.try_recv().is_ok() {}
+            cancel.store(false, Ordering::Release);
+        } else {
+            while text_rx.try_recv().is_ok() {}
+            cancel.store(false, Ordering::Release);
         }
-        while text_rx.try_recv().is_ok() {}
-        cancel.store(false, Ordering::Release);
         tts_active.store(false, Ordering::Release);
         return true;
     }
     false
 }
 
+/// Acquire the `player_ops` lock, recovering from poison.
+///
+/// The data under the mutex is `()` — it only serializes Player mutations —
+/// so a panicked holder leaves nothing inconsistent to observe and recovery
+/// is always safe. Without this, a worker panic would wedge the monitor (or
+/// vice versa) on `unwrap()`.
+fn lock_player_ops(ops: &Mutex<()>) -> MutexGuard<'_, ()> {
+    ops.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
 /// Apply the fixed playback gain ([`PLAYBACK_GAIN`]), hard-clamped to ±1.0.
 ///
 /// Replaces the earlier per-sentence peak normalization — see the

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

@@ -7,7 +7,7 @@ use super::*;
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 // ── Remote interrupt tracker ──────────────────────────────────────────────
 //
@@ -468,19 +468,33 @@ fn on_receipt_check_releases_mic_gate_before_synthesis() {
 //
 // Models one tick of the tts-barge-in-monitor thread in `tts_worker`. The
 // contract (must match production):
-//   - cancel set   → silence player + release tts_active, flag NOT consumed
+//   - cancel set   → take `player_ops`, RE-CHECK cancel under the lock; if
+//     still set: silence player + release tts_active, flag NOT consumed
 //     (the worker owns consumption: queue drain + lead-in reset)
+//   - cancel cleared by the time the lock is held → no-op (stale branch)
 //   - cancel clear → no-op
 // Not consuming the flag is what makes the monitor idempotent across ticks
 // and closes the race where the worker appends a sentence after the
-// monitor's clear but before consuming the flag.
+// monitor's clear but before consuming the flag. The under-lock re-check
+// closes the converse race (PR #997 review blocker): worker consumes the
+// cancel and appends a fresh post-cancel utterance between the monitor's
+// initial load and its clear — the stale branch must not delete that audio.
 
 /// Test-side model of one monitor tick. `player_cleared` stands in for the
-/// `clear()+play()` pair on the real Player.
-fn simulate_monitor_tick(cancel: &AtomicBool, tts_active: &AtomicBool, player_cleared: &mut bool) {
+/// `clear()+play()` pair on the real Player; `player_ops` is the mutex
+/// serializing monitor clears with worker player mutations.
+fn simulate_monitor_tick(
+    cancel: &AtomicBool,
+    tts_active: &AtomicBool,
+    player_ops: &Mutex<()>,
+    player_cleared: &mut bool,
+) {
     if cancel.load(Ordering::Acquire) {
-        *player_cleared = true;
-        tts_active.store(false, Ordering::Release);
+        let _ops = player_ops.lock().unwrap();
+        if cancel.load(Ordering::Acquire) {
+            *player_cleared = true;
+            tts_active.store(false, Ordering::Release);
+        }
     }
 }
 
@@ -490,9 +504,10 @@ fn simulate_monitor_tick(cancel: &AtomicBool, tts_active: &AtomicBool, player_cl
 fn monitor_tick_silences_and_releases_without_consuming_cancel() {
     let cancel = AtomicBool::new(true);
     let tts_active = AtomicBool::new(true);
+    let player_ops = Mutex::new(());
     let mut player_cleared = false;
 
-    simulate_monitor_tick(&cancel, &tts_active, &mut player_cleared);
+    simulate_monitor_tick(&cancel, &tts_active, &player_ops, &mut player_cleared);
 
     assert!(player_cleared, "monitor must silence in-flight audio");
     assert!(
@@ -511,9 +526,10 @@ fn monitor_tick_silences_and_releases_without_consuming_cancel() {
 fn monitor_tick_noop_without_cancel() {
     let cancel = AtomicBool::new(false);
     let tts_active = AtomicBool::new(true);
+    let player_ops = Mutex::new(());
     let mut player_cleared = false;
 
-    simulate_monitor_tick(&cancel, &tts_active, &mut player_cleared);
+    simulate_monitor_tick(&cancel, &tts_active, &player_ops, &mut player_cleared);
 
     assert!(!player_cleared);
     assert!(
@@ -522,6 +538,53 @@ fn monitor_tick_noop_without_cancel() {
     );
 }
 
+/// Stale-branch race (PR #997 review blocker): monitor observes
+/// `cancel == true`, then the worker — under `player_ops` — consumes the
+/// cancel and appends a fresh post-cancel utterance before the monitor
+/// reaches its clear. The monitor's under-lock re-check must see
+/// `cancel == false` and no-op, leaving the fresh audio and its mic gate
+/// intact.
+#[test]
+fn monitor_stale_cancel_branch_must_not_clear_fresh_audio() {
+    let cancel = AtomicBool::new(true);
+    let tts_active = AtomicBool::new(true);
+    let player_ops = Mutex::new(());
+    let mut player_cleared = false;
+
+    // Monitor's initial (pre-lock) load observes the cancel…
+    let stale_observation = cancel.load(Ordering::Acquire);
+    assert!(stale_observation);
+
+    // …then the worker wins the lock: consumes the cancel, appends a fresh
+    // utterance, sets tts_active (mirrors handle_cancel_or_shutdown +
+    // the locked append in the sentence loop).
+    {
+        let _ops = player_ops.lock().unwrap();
+        cancel.store(false, Ordering::Release);
+        tts_active.store(true, Ordering::Release);
+    }
+
+    // Monitor resumes from its stale branch — the re-check under the lock
+    // must turn it into a no-op.
+    if stale_observation {
+        let _ops = player_ops.lock().unwrap();
+        if cancel.load(Ordering::Acquire) {
+            player_cleared = true;
+            tts_active.store(false, Ordering::Release);
+        }
+    }
+
+    assert!(
+        !player_cleared,
+        "stale monitor branch must not clear a fresh post-cancel utterance",
+    );
+    assert!(
+        tts_active.load(Ordering::Acquire),
+        "stale monitor branch must not release the mic gate while fresh \
+         audio is playing",
+    );
+}
+
 /// Full cycle: remote speech → cancel → TTS consumption → new TTS → cancel again.
 /// Validates the cancel mechanism is reusable across TTS sessions.
 /// The false→true transition on tts_active auto-clears counters — no