From 6b5f796852c151dc0bb72741accea044bd7ea7c0 Mon Sep 17 00:00:00 2001 From: npub1223z34hd7vtwc6qj4s7flsxkj644nlre2nthu7lrrmkumhu3xddsrx9r6w <52a228d6edf316ec6812ac3c9fc0d696ab59fc7954d77e7be31eedcddf91335b@buzz.block.builderlab.xyz> Date: Mon, 27 Jul 2026 15:30:01 -0700 Subject: [PATCH] feat(desktop): add read-aloud playback for agent chat messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Let a user explicitly play/stop a delivered agent response from the message action bar, reusing the local Pocket TTS stack while keeping read-aloud fully independent from huddle orchestration. Backend (src-tauri): - New app-scoped message_tts module with its own TtsPipeline instance and fresh active/cancel flags — zero STT/barge-in/huddle-phase coupling. Commands: speak_chat_message / stop_chat_message_speech. - Message-keyed lifecycle (preparing/playing/completed/stopped/ superseded/failed) emitted as message-tts-state-changed events. Generation tokens guarantee stale watcher completions can never overwrite a newer request; one playback app-wide, newer play supersedes older. - Huddle exclusion both ways: speak refuses while huddle phase is not Idle, and start/join_huddle stop any message playback before taking the audio device. - Long-message limit mirrors the huddle cap (2000 chars, char-counted) with a spoken suffix and a truncated flag so the cut is disclosed, not silent. Missing voice model kicks off the download and surfaces a retryable error. No persisted audio cache. Frontend (desktop/src): - MessageTtsProvider (mounted in AppShell inside HuddleProvider) folds backend events through a pure reducer (playbackReducer.mjs) so stale terminal events for superseded messages can't clear newer playback. Errors surface as toasts + per-message state; truncation is disclosed via toast at playback start. - ReadAloudButton in MessageActionBar for delivered agent text only (isAgent, not pending, has body), with aria-label/aria-pressed, tooltip, and disabled-with-reason during active huddles. Tests: 16 Rust unit tests (truncation UTF-8 safety, supersede/stop races, concurrent stops, event payload shape) and 13 node:test reducer tests (out-of-order supersede delivery, detached errors, rapid play/stop). Full suites green: cargo test (1576), pnpm test (3414), typecheck, biome + file-size checks, desktop build. Co-authored-by: Taylor Ho Signed-off-by: Taylor Ho --- desktop/scripts/check-file-sizes.mjs | 5 +- desktop/src-tauri/src/app_state.rs | 4 + desktop/src-tauri/src/huddle/mod.rs | 4 + desktop/src-tauri/src/lib.rs | 3 + desktop/src-tauri/src/message_tts.rs | 497 ++++++++++++++++++ desktop/src-tauri/src/message_tts_tests.rs | 214 ++++++++ desktop/src/app/AppShell.tsx | 9 +- .../message-tts/lib/playbackReducer.d.mts | 39 ++ .../message-tts/lib/playbackReducer.mjs | 76 +++ .../message-tts/lib/playbackReducer.test.mjs | 134 +++++ .../message-tts/ui/MessageTtsProvider.tsx | 123 +++++ .../features/messages/ui/MessageActionBar.tsx | 73 +++ 12 files changed, 1179 insertions(+), 2 deletions(-) create mode 100644 desktop/src-tauri/src/message_tts.rs create mode 100644 desktop/src-tauri/src/message_tts_tests.rs create mode 100644 desktop/src/features/message-tts/lib/playbackReducer.d.mts create mode 100644 desktop/src/features/message-tts/lib/playbackReducer.mjs create mode 100644 desktop/src/features/message-tts/lib/playbackReducer.test.mjs create mode 100644 desktop/src/features/message-tts/ui/MessageTtsProvider.tsx diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs index c57e0e982e..7beccf4f41 100644 --- a/desktop/scripts/check-file-sizes.mjs +++ b/desktop/scripts/check-file-sizes.mjs @@ -396,7 +396,10 @@ const overrides = new Map([ // transition lock doc broadened to cover all protected-PID transitions, and // clear_agent_session_caches (per-pubkey retain) added alongside the // per-key clear. Load-bearing identity-contract change; queued to split. - ["src-tauri/src/app_state.rs", 1081], + // +4 (1081 -> 1085): message read-aloud experiment — the app-scoped + // MessageTtsState field (doc + init) added beside huddle_state. The state + // itself lives in message_tts.rs; only the wiring lands here. + ["src-tauri/src/app_state.rs", 1085], // multi-slot splitting + no-op suppression (#1309): the ReadStateManager // class grew from ~700 lines to ~1019 with the addition of // splitContextsIntoBudgetedSlots (pure fn + 5 tests), publishSplitSlots, diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index 778bd5fb3f..b4331328bb 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -48,6 +48,9 @@ pub struct AppState { pub channel_templates_store_lock: Mutex<()>, pub managed_agent_processes: Mutex>, pub huddle_state: Mutex, + /// App-scoped message read-aloud state (independent of huddle speech). + /// See `message_tts.rs` for the lifecycle contract. + pub message_tts: crate::message_tts::MessageTtsState, /// Tauri app handle — stored after setup so huddle commands can emit /// `huddle-state-changed` events without needing the handle threaded /// through every call site. @@ -211,6 +214,7 @@ pub fn build_app_state() -> AppState { managed_agent_processes: Mutex::new(HashMap::new()), session_config_cache: Mutex::new(HashMap::new()), huddle_state: Mutex::new(HuddleState::default()), + message_tts: crate::message_tts::MessageTtsState::default(), app_handle: Mutex::new(None), audio_output_device: Mutex::new(None), media_proxy_port: AtomicU16::new(0), diff --git a/desktop/src-tauri/src/huddle/mod.rs b/desktop/src-tauri/src/huddle/mod.rs index 8a8f1ec9ff..cb0ed319bb 100644 --- a/desktop/src-tauri/src/huddle/mod.rs +++ b/desktop/src-tauri/src/huddle/mod.rs @@ -197,6 +197,8 @@ pub async fn start_huddle( hs.phase = HuddlePhase::Creating; hs.parent_channel_id = Some(parent_channel_id.clone()); } + // The huddle owns the audio device now — end any message read-aloud. + crate::message_tts::stop_message_playback(&state); let ephemeral_uuid = Uuid::new_v4(); let ephemeral_channel_id = ephemeral_uuid.to_string(); @@ -357,6 +359,8 @@ pub async fn join_huddle( hs.parent_channel_id = Some(parent_channel_id.clone()); hs.ephemeral_channel_id = Some(ephemeral_channel_id.clone()); } + // The huddle owns the audio device now — end any message read-aloud. + crate::message_tts::stop_message_playback(&state); // Seed participant list with own pubkey as a fallback until relay responds. let own_pubkey = state diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 9017356ab5..1e517f1e67 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -13,6 +13,7 @@ mod media_proxy; mod mesh_llm; #[cfg(not(feature = "mesh-llm"))] mod mesh_llm_stubs; +mod message_tts; mod migration; #[cfg(test)] mod model_tests; @@ -857,6 +858,8 @@ pub fn run() { get_model_status, set_tts_enabled, speak_agent_message, + message_tts::speak_chat_message, + message_tts::stop_chat_message_speech, add_agent_to_huddle, check_pipeline_hotstart, confirm_huddle_active, diff --git a/desktop/src-tauri/src/message_tts.rs b/desktop/src-tauri/src/message_tts.rs new file mode 100644 index 0000000000..c448adecab --- /dev/null +++ b/desktop/src-tauri/src/message_tts.rs @@ -0,0 +1,497 @@ +//! App-scoped read-aloud playback for ordinary chat messages. +//! +//! Mental model: +//! +//! ```text +//! frontend: Play on a delivered agent message +//! → speak_chat_message(message_id, text) +//! 1. Reject while a huddle is live (one audio owner at a time) +//! 2. Reject when the Pocket TTS model is not ready (kick off download) +//! 3. Preprocess + truncate (MAX_MESSAGE_TTS_CHARS, disclosed to the UI) +//! 4. Supersede any active playback (bump generation, cancel, emit) +//! 5. Lazily build a dedicated TtsPipeline (NOT the huddle one) +//! 6. Queue text, emit "preparing", spawn a watcher task +//! → watcher polls the pipeline's active flag +//! active rises → emit "playing" +//! active falls → emit "completed" +//! generation changed → exit silently (a newer request owns the UI) +//! frontend: Stop → stop_chat_message_speech → cancel + emit "stopped" +//! ``` +//! +//! This module deliberately does NOT touch `HuddleState`. It owns its own +//! `TtsPipeline` instance with fresh `active`/`cancel` flags, so message +//! read-aloud carries none of the huddle's STT echo-gating, barge-in, or +//! phase lifecycle. The only huddle coupling is exclusion: `speak` refuses +//! while a huddle is non-idle, and huddle start/join calls +//! [`stop_message_playback`] so two pipelines never fight one output device. +//! +//! State machine (message-keyed, generation-guarded): +//! +//! ```text +//! preparing → playing → completed +//! │ │ +//! ├── stopped (explicit stop / huddle start) +//! ├── superseded (newer play request) +//! └── failed (engine/audio/queue error) +//! ``` +//! +//! Every transition is emitted as a `message-tts-state-changed` Tauri event. +//! Stale worker callbacks can never overwrite newer state: each play bumps +//! `generation`, and both the watcher and terminal emits check it first. + +use std::sync::{ + atomic::{AtomicBool, AtomicU64, Ordering}, + Arc, Mutex, +}; +use std::time::Duration; + +use serde::Serialize; +use tauri::State; + +use crate::app_state::AppState; +use crate::huddle::preprocessing::preprocess_for_tts; +use crate::huddle::tts::TtsPipeline; +use crate::huddle::{models, HuddlePhase}; + +// ── Constants ───────────────────────────────────────────────────────────────── + +/// Maximum characters synthesized per message (mirrors the huddle cap). +/// ~2000 chars ≈ 1–2 minutes of speech. The UI is told via `truncated` +/// so the cut is disclosed rather than silent. +pub(crate) const MAX_MESSAGE_TTS_CHARS: usize = 2000; + +/// Spoken suffix appended when a message is truncated, so the listener also +/// hears the disclosure. +pub(crate) const TRUNCATION_SUFFIX: &str = "... message truncated."; + +/// How often the watcher polls the pipeline's active flag. +const WATCH_TICK: Duration = Duration::from_millis(25); + +/// How long the watcher waits for audio to start before declaring failure. +/// First-chunk synthesis is a single sentence (~1s warm); 30s covers a cold +/// engine plus pathological first sentences with a wide margin. +const START_TIMEOUT: Duration = Duration::from_secs(30); + +/// How long `speak` waits for the worker to consume a cancel before deciding +/// the pipeline is wedged and rebuilding it. The worker's recv loop observes +/// the flag within ~100 ms. +const CANCEL_CONSUME_TIMEOUT: Duration = Duration::from_secs(2); + +/// Tauri event name for playback state transitions. +pub(crate) const STATE_EVENT: &str = "message-tts-state-changed"; + +// ── State ───────────────────────────────────────────────────────────────────── + +/// App-scoped message read-aloud state. Lives on [`AppState`]. +pub struct MessageTtsState { + /// Pipeline + current playback bookkeeping. Never held across `.await`. + inner: Mutex, + /// Monotonic play-request counter. Bumped by every `speak` and `stop`; + /// watchers capture it at spawn and exit when it moves on. + generation: Arc, + /// True while audio for the current request is queued or playing. + /// Shared with the pipeline worker thread (same contract as huddle TTS, + /// but nothing else reads it — no STT gating here). + active: Arc, + /// Cancel flag consumed by the pipeline worker: drains the queue and + /// silences the player. + cancel: Arc, + /// TOCTOU sentinel for pipeline construction (mirrors `tts_starting`). + starting: Arc, +} + +struct MessageTtsInner { + pipeline: Option>, + /// The message whose audio currently owns the player, if any. + current_message_id: Option, +} + +impl Default for MessageTtsState { + fn default() -> Self { + Self { + inner: Mutex::new(MessageTtsInner { + pipeline: None, + current_message_id: None, + }), + generation: Arc::new(AtomicU64::new(0)), + active: Arc::new(AtomicBool::new(false)), + cancel: Arc::new(AtomicBool::new(false)), + starting: Arc::new(AtomicBool::new(false)), + } + } +} + +impl MessageTtsState { + fn lock(&self) -> std::sync::MutexGuard<'_, MessageTtsInner> { + self.inner.lock().unwrap_or_else(|e| e.into_inner()) + } +} + +// ── Frontend event payload ──────────────────────────────────────────────────── + +/// Payload of every `message-tts-state-changed` emission. +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +pub(crate) struct PlaybackEvent { + pub message_id: String, + /// "preparing" | "playing" | "completed" | "stopped" | "superseded" | "failed" + pub status: &'static str, + /// True when the spoken text was cut at [`MAX_MESSAGE_TTS_CHARS`]. + pub truncated: bool, + pub error: Option, +} + +fn emit_playback_event(state: &AppState, event: &PlaybackEvent) { + use tauri::Emitter; + let app = match state.app_handle.lock() { + Ok(guard) => guard.clone(), + Err(_) => return, + }; + if let Some(app) = app { + let _ = app.emit(STATE_EVENT, event); + } +} + +// ── Pure helpers (unit-tested in message_tts_tests.rs) ──────────────────────── + +/// Preprocess and cap the raw message body for synthesis. +/// +/// Returns the speakable text plus whether it was truncated, or `Err` when +/// preprocessing leaves nothing to say (emoji-only, code-only, etc.). +/// Truncation counts chars (not bytes) so multi-byte UTF-8 never panics. +pub(crate) fn prepare_speech_text(raw: &str) -> Result<(String, bool), String> { + let truncated = raw.chars().count() > MAX_MESSAGE_TTS_CHARS; + let capped: String = if truncated { + let mut t: String = raw.chars().take(MAX_MESSAGE_TTS_CHARS).collect(); + t.push_str(TRUNCATION_SUFFIX); + t + } else { + raw.to_string() + }; + + let speakable = preprocess_for_tts(&capped); + if speakable.is_empty() { + return Err("message has no speakable text".to_string()); + } + Ok((speakable, truncated)) +} + +/// Whether a watcher spawned at `spawned_gen` still owns the UI state. +/// Split out so the supersede contract is directly testable. +pub(crate) fn watcher_owns_state(generation: &AtomicU64, spawned_gen: u64) -> bool { + generation.load(Ordering::Acquire) == spawned_gen +} + +// ── Playback control ────────────────────────────────────────────────────────── + +/// Stop any active message playback. Sync + lock-only, safe from any thread. +/// Used by `stop_chat_message_speech` and by huddle start/join (the huddle +/// takes ownership of the audio device). +/// +/// `status` is "stopped" for both paths — from the UI's perspective the +/// playback ended before completion at the user's request either way. +pub(crate) fn stop_message_playback(state: &AppState) { + let mtts = &state.message_tts; + let previous = { + let mut inner = mtts.lock(); + inner.current_message_id.take() + }; + // Invalidate watchers FIRST so a falling active flag can't emit + // "completed" concurrently with our "stopped". + mtts.generation.fetch_add(1, Ordering::AcqRel); + mtts.cancel.store(true, Ordering::Release); + + if let Some(message_id) = previous { + emit_playback_event( + state, + &PlaybackEvent { + message_id, + status: "stopped", + truncated: false, + error: None, + }, + ); + } +} + +/// Ensure a live dedicated pipeline exists, building one if needed. +/// Never called with the inner lock held across the construction await. +async fn ensure_pipeline(state: &AppState) -> Result, String> { + // Fast path: existing live pipeline. + { + let inner = state.message_tts.lock(); + if let Some(ref p) = inner.pipeline { + if !p.is_finished() { + return Ok(Arc::clone(p)); + } + } + } + + let model_dir = models::tts_model_dir().ok_or_else(|| { + // Kick the download so a retry can succeed without a huddle first. + if let Some(mgr) = models::global_model_manager() { + mgr.start_tts_download(state.http_client.clone()); + } + "voice model is not downloaded yet — download started, try again shortly".to_string() + })?; + + // Claim the construction slot (TOCTOU guard, mirrors huddle tts_starting). + if state.message_tts.starting.swap(true, Ordering::AcqRel) { + return Err("read-aloud is still starting up — try again in a moment".to_string()); + } + + let active = Arc::clone(&state.message_tts.active); + let cancel = Arc::clone(&state.message_tts.cancel); + let output_device = state + .audio_output_device + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + + let constructed = tokio::task::spawn_blocking(move || { + TtsPipeline::new(model_dir, active, cancel, output_device) + }) + .await; + + state.message_tts.starting.store(false, Ordering::Release); + + let pipeline = match constructed { + Ok(Ok(p)) => Arc::new(p), + Ok(Err(e)) => return Err(e), + Err(e) => return Err(format!("spawn_blocking failed: {e}")), + }; + + let mut inner = state.message_tts.lock(); + // A dead pipeline may still be stored; replace it. (A concurrent build is + // impossible — we held the `starting` sentinel.) + inner.pipeline = Some(Arc::clone(&pipeline)); + Ok(pipeline) +} + +/// Wait for the worker to consume the cancel flag so newly queued text is not +/// swallowed by the post-cancel queue drain. Returns `false` on timeout +/// (wedged worker — caller rebuilds the pipeline). +async fn wait_cancel_consumed(cancel: &AtomicBool) -> bool { + let deadline = std::time::Instant::now() + CANCEL_CONSUME_TIMEOUT; + while cancel.load(Ordering::Acquire) { + if std::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(WATCH_TICK).await; + } + true +} + +// ── Tauri commands ──────────────────────────────────────────────────────────── + +/// Read a delivered chat message aloud. One playback app-wide: a newer call +/// supersedes the previous one. +#[tauri::command] +pub async fn speak_chat_message( + message_id: String, + text: String, + state: State<'_, AppState>, +) -> Result<(), String> { + // Pre-flight failures (before any state is claimed) still emit a "failed" + // event so the frontend has one uniform error surface. They never touch + // generation/current_message, so an unrelated active playback is unharmed + // (the frontend reducer keys on message_id). + let preflight_fail = |err: String, state: &AppState| { + emit_playback_event( + state, + &PlaybackEvent { + message_id: message_id.clone(), + status: "failed", + truncated: false, + error: Some(err.clone()), + }, + ); + Err::<(), String>(err) + }; + + // Huddles own the audio device while live — refuse, don't mix. + { + let hs = state.huddle()?; + if hs.phase != HuddlePhase::Idle { + drop(hs); + return preflight_fail( + "read-aloud is unavailable during a huddle".to_string(), + &state, + ); + } + } + + let (speech_text, truncated) = match prepare_speech_text(&text) { + Ok(v) => v, + Err(e) => return preflight_fail(e, &state), + }; + + // Supersede any current playback and claim the new generation. + let superseded = { + let mut inner = state.message_tts.lock(); + inner.current_message_id.replace(message_id.clone()) + }; + let gen = state.message_tts.generation.fetch_add(1, Ordering::AcqRel) + 1; + if let Some(old_id) = superseded { + // The same message re-played also passes through supersede: the old + // audio stops and a fresh request starts — deterministic either way. + state.message_tts.cancel.store(true, Ordering::Release); + emit_playback_event( + &state, + &PlaybackEvent { + message_id: old_id, + status: "superseded", + truncated: false, + error: None, + }, + ); + } + + emit_playback_event( + &state, + &PlaybackEvent { + message_id: message_id.clone(), + status: "preparing", + truncated, + error: None, + }, + ); + + let fail = |err: String, state: &AppState| { + // Only report failure if no newer request took over meanwhile. + if watcher_owns_state(&state.message_tts.generation, gen) { + let mut inner = state.message_tts.lock(); + if inner.current_message_id.as_deref() == Some(message_id.as_str()) { + inner.current_message_id = None; + } + drop(inner); + emit_playback_event( + state, + &PlaybackEvent { + message_id: message_id.clone(), + status: "failed", + truncated, + error: Some(err.clone()), + }, + ); + } + Err::<(), String>(err) + }; + + let mut pipeline = match ensure_pipeline(&state).await { + Ok(p) => p, + Err(e) => return fail(e, &state), + }; + + // If we cancelled a previous playback, wait for the worker to consume the + // flag — otherwise its post-cancel queue drain would swallow our text. + if state.message_tts.cancel.load(Ordering::Acquire) + && !wait_cancel_consumed(&state.message_tts.cancel).await + { + // Worker wedged: drop it and rebuild once. Old pipeline drops (and + // joins its thread) outside the inner lock. + let old = { + let mut inner = state.message_tts.lock(); + inner.pipeline.take() + }; + if let Some(p) = old { + p.shutdown(); + drop(p); + } + state.message_tts.cancel.store(false, Ordering::Release); + pipeline = match ensure_pipeline(&state).await { + Ok(p) => p, + Err(e) => return fail(e, &state), + }; + } + + // A stop/supersede may have landed while we were constructing/waiting. + if !watcher_owns_state(&state.message_tts.generation, gen) { + return Ok(()); + } + + if let Err(e) = pipeline.speak(speech_text) { + return fail(e, &state); + } + + spawn_playback_watcher(&state, message_id, gen, truncated); + Ok(()) +} + +/// Stop the active message playback, if any. Always succeeds. +#[tauri::command] +pub fn stop_chat_message_speech(state: State<'_, AppState>) -> Result<(), String> { + stop_message_playback(&state); + Ok(()) +} + +// ── Watcher ─────────────────────────────────────────────────────────────────── + +/// Watch the pipeline's active flag and emit playing/completed/failed for +/// `message_id`. Exits silently the moment `generation` moves past +/// `spawned_gen` — the superseding/stopping call already emitted for us. +fn spawn_playback_watcher(state: &AppState, message_id: String, spawned_gen: u64, truncated: bool) { + let generation = Arc::clone(&state.message_tts.generation); + let active = Arc::clone(&state.message_tts.active); + let app = match state.app_handle.lock() { + Ok(guard) => guard.clone(), + Err(_) => None, + }; + let Some(app) = app else { return }; + + tauri::async_runtime::spawn(async move { + use tauri::Emitter; + let emit = |status: &'static str, error: Option| { + let _ = app.emit( + STATE_EVENT, + PlaybackEvent { + message_id: message_id.clone(), + status, + truncated, + error, + }, + ); + }; + + // Phase 1: wait for audio to start. + let start_deadline = std::time::Instant::now() + START_TIMEOUT; + loop { + if !watcher_owns_state(&generation, spawned_gen) { + return; + } + if active.load(Ordering::Acquire) { + break; + } + if std::time::Instant::now() >= start_deadline { + if watcher_owns_state(&generation, spawned_gen) { + emit( + "failed", + Some("audio did not start — the voice engine may be unavailable".into()), + ); + } + return; + } + tokio::time::sleep(WATCH_TICK).await; + } + emit("playing", None); + + // Phase 2: wait for the player to drain. + loop { + if !watcher_owns_state(&generation, spawned_gen) { + return; + } + if !active.load(Ordering::Acquire) { + break; + } + tokio::time::sleep(WATCH_TICK).await; + } + if watcher_owns_state(&generation, spawned_gen) { + emit("completed", None); + } + }); +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +#[path = "message_tts_tests.rs"] +mod tests; diff --git a/desktop/src-tauri/src/message_tts_tests.rs b/desktop/src-tauri/src/message_tts_tests.rs new file mode 100644 index 0000000000..7768157099 --- /dev/null +++ b/desktop/src-tauri/src/message_tts_tests.rs @@ -0,0 +1,214 @@ +//! Tests for message_tts.rs — the app-scoped chat read-aloud lifecycle. +//! +//! Covers the pure helpers (text preparation, truncation disclosure) and the +//! concurrency contracts (generation-based supersede, stop invalidation, +//! stale-watcher silence). The pipeline itself is exercised by huddle tests; +//! here we prove the *coordination* around it is deterministic. + +use super::*; + +// ── prepare_speech_text ─────────────────────────────────────────────────── + +#[test] +fn short_message_passes_through_untruncated() { + let (text, truncated) = prepare_speech_text("Hello there, friend.").unwrap(); + assert_eq!(text, "Hello there, friend."); + assert!(!truncated); +} + +#[test] +fn long_message_is_truncated_with_spoken_disclosure() { + let raw = "word ".repeat(1000); // 5000 chars + let (text, truncated) = prepare_speech_text(&raw).unwrap(); + assert!(truncated); + // The spoken suffix survives preprocessing so the listener hears the cut. + assert!( + text.ends_with("message truncated."), + "expected spoken disclosure, got tail: {:?}", + &text[text.len().saturating_sub(40)..] + ); + // Cap applies to the source text; result stays in the same ballpark. + assert!(text.chars().count() <= MAX_MESSAGE_TTS_CHARS + TRUNCATION_SUFFIX.chars().count()); +} + +#[test] +fn exactly_at_limit_is_not_truncated() { + let raw = "a".repeat(MAX_MESSAGE_TTS_CHARS); + let (_, truncated) = prepare_speech_text(&raw).unwrap(); + assert!(!truncated); +} + +#[test] +fn one_past_limit_is_truncated() { + let raw = "a".repeat(MAX_MESSAGE_TTS_CHARS + 1); + let (_, truncated) = prepare_speech_text(&raw).unwrap(); + assert!(truncated); +} + +#[test] +fn truncation_counts_chars_not_bytes() { + // é is 2 bytes in UTF-8; a byte-indexed slice at the limit would panic + // or split a code point. chars().take() must handle it. + let raw = "é".repeat(MAX_MESSAGE_TTS_CHARS + 10); + let (text, truncated) = prepare_speech_text(&raw).unwrap(); + assert!(truncated); + assert!(text.starts_with('é')); +} + +#[test] +fn unspeakable_message_is_an_error_not_silence() { + // URL-only content preprocesses to a placeholder, but emoji/whitespace-only + // reduces to nothing — must surface as Err so the UI shows "failed" + // instead of a playback that never starts. + let err = prepare_speech_text(" \n\t ").unwrap_err(); + assert!(err.contains("no speakable text"), "got: {err}"); +} + +#[test] +fn code_only_message_yields_omission_placeholder_not_error() { + // A fenced block preprocesses to "code block omitted" — still speakable. + let (text, _) = prepare_speech_text("```rust\nfn main() {}\n```").unwrap(); + assert!(text.contains("code block omitted"), "got: {text}"); +} + +#[test] +fn markdown_is_preprocessed_for_speech() { + let (text, _) = prepare_speech_text("**Done.** See https://example.com").unwrap(); + assert!(!text.contains("**"), "got: {text}"); + assert!(!text.contains("example.com"), "got: {text}"); + assert!(text.contains("link omitted"), "got: {text}"); +} + +// ── Generation / supersede contract ─────────────────────────────────────── + +#[test] +fn watcher_owns_state_until_generation_moves() { + let generation = AtomicU64::new(7); + assert!(watcher_owns_state(&generation, 7)); + generation.fetch_add(1, Ordering::AcqRel); + assert!(!watcher_owns_state(&generation, 7)); + assert!(watcher_owns_state(&generation, 8)); +} + +#[test] +fn stop_invalidates_watchers_and_clears_current_message() { + let state = crate::app_state::build_app_state(); + // Simulate an in-flight playback: watcher spawned at the current gen. + let spawned_gen = state.message_tts.generation.load(Ordering::Acquire); + state.message_tts.lock().current_message_id = Some("msg-1".into()); + state.message_tts.active.store(true, Ordering::Release); + + stop_message_playback(&state); + + // Watcher for the old playback must no longer own UI state — its + // falling-active "completed" emit is suppressed. + assert!(!watcher_owns_state( + &state.message_tts.generation, + spawned_gen + )); + // The worker is told to drain. + assert!(state.message_tts.cancel.load(Ordering::Acquire)); + // Message bookkeeping is cleared so a later stop is a no-op. + assert!(state.message_tts.lock().current_message_id.is_none()); +} + +#[test] +fn stop_with_nothing_playing_is_a_quiet_no_op() { + let state = crate::app_state::build_app_state(); + let gen_before = state.message_tts.generation.load(Ordering::Acquire); + stop_message_playback(&state); + // Generation still bumps (harmless — no watcher exists), no panic, and + // there was no message to clear. + assert!(state.message_tts.lock().current_message_id.is_none()); + assert!(state.message_tts.generation.load(Ordering::Acquire) > gen_before); +} + +#[test] +fn newer_play_supersedes_older_watcher_deterministically() { + // Models the exact interleaving speak_chat_message() produces: + // play A (gen 1) → play B (gen 2). A's watcher observes the moved + // generation and exits without emitting, regardless of when A's audio + // actually stops — so B's state can never be overwritten by A. + let state = crate::app_state::build_app_state(); + + // Play A. + state.message_tts.lock().current_message_id = Some("msg-a".into()); + let gen_a = state.message_tts.generation.fetch_add(1, Ordering::AcqRel) + 1; + assert!(watcher_owns_state(&state.message_tts.generation, gen_a)); + + // Play B supersedes: replaces the current message, bumps gen, cancels. + let old = state + .message_tts + .lock() + .current_message_id + .replace("msg-b".into()); + assert_eq!(old.as_deref(), Some("msg-a")); + let gen_b = state.message_tts.generation.fetch_add(1, Ordering::AcqRel) + 1; + state.message_tts.cancel.store(true, Ordering::Release); + + assert!(!watcher_owns_state(&state.message_tts.generation, gen_a)); + assert!(watcher_owns_state(&state.message_tts.generation, gen_b)); +} + +#[test] +fn rapid_stop_play_stop_leaves_consistent_state() { + let state = crate::app_state::build_app_state(); + for i in 0..50 { + let id = format!("msg-{i}"); + state.message_tts.lock().current_message_id = Some(id.clone()); + state.message_tts.generation.fetch_add(1, Ordering::AcqRel); + stop_message_playback(&state); + assert!(state.message_tts.lock().current_message_id.is_none()); + } + // Cancel flag is set (worker would consume it); no watcher owns any + // historical generation. + assert!(state.message_tts.cancel.load(Ordering::Acquire)); +} + +#[test] +fn concurrent_stops_from_many_threads_never_deadlock_or_double_clear() { + use std::thread; + let state = std::sync::Arc::new(crate::app_state::build_app_state()); + state.message_tts.lock().current_message_id = Some("msg-x".into()); + + let handles: Vec<_> = (0..8) + .map(|_| { + let state = std::sync::Arc::clone(&state); + thread::spawn(move || stop_message_playback(&state)) + }) + .collect(); + for h in handles { + h.join().unwrap(); + } + assert!(state.message_tts.lock().current_message_id.is_none()); +} + +// ── Event payload shape (frontend contract) ─────────────────────────────── + +#[test] +fn playback_event_serializes_with_stable_field_names() { + let ev = PlaybackEvent { + message_id: "abc".into(), + status: "playing", + truncated: true, + error: None, + }; + let json = serde_json::to_value(&ev).unwrap(); + assert_eq!(json["message_id"], "abc"); + assert_eq!(json["status"], "playing"); + assert_eq!(json["truncated"], true); + assert!(json["error"].is_null()); +} + +#[test] +fn failed_event_carries_the_error_text() { + let ev = PlaybackEvent { + message_id: "abc".into(), + status: "failed", + truncated: false, + error: Some("voice model is not downloaded yet".into()), + }; + let json = serde_json::to_value(&ev).unwrap(); + assert_eq!(json["status"], "failed"); + assert_eq!(json["error"], "voice model is not downloaded yet"); +} diff --git a/desktop/src/app/AppShell.tsx b/desktop/src/app/AppShell.tsx index f10f7febaf..2fc2e78af7 100644 --- a/desktop/src/app/AppShell.tsx +++ b/desktop/src/app/AppShell.tsx @@ -63,6 +63,7 @@ import { isSettingsSection, } from "@/features/settings/ui/SettingsPanels"; import { HuddleBar, HuddleProvider } from "@/features/huddle"; +import { MessageTtsProvider } from "@/features/message-tts/ui/MessageTtsProvider"; import { useDueReminderBadgeCount } from "@/features/reminders/hooks"; import { RemindMeLaterProvider } from "@/features/reminders/ui/RemindMeLaterProvider"; import { useReminderNotifications } from "@/features/reminders/useReminderNotifications"; @@ -875,7 +876,13 @@ export function AppShell() { style={chromeCssVarDefaults as React.CSSProperties} > - + {/* Message read-aloud state; scoped to routed + content — every message surface renders + under the Outlet. Inside HuddleProvider + (provider reads huddle state). */} + + + diff --git a/desktop/src/features/message-tts/lib/playbackReducer.d.mts b/desktop/src/features/message-tts/lib/playbackReducer.d.mts new file mode 100644 index 0000000000..bfff5645bd --- /dev/null +++ b/desktop/src/features/message-tts/lib/playbackReducer.d.mts @@ -0,0 +1,39 @@ +/** + * Type declarations for `playbackReducer.mjs`. Runtime lives in `.mjs` so the + * `node:test` runner can import it directly; this file gives TypeScript + * callers a typed view. + */ + +export type MessageTtsPhase = "idle" | "preparing" | "playing"; + +export type MessageTtsUiState = { + messageId: string | null; + phase: MessageTtsPhase; + truncated: boolean; + error: string | null; +}; + +export type MessageTtsBackendEvent = { + message_id: string; + status: + | "preparing" + | "playing" + | "completed" + | "stopped" + | "superseded" + | "failed"; + truncated: boolean; + error: string | null; +}; + +export const IDLE_STATE: MessageTtsUiState; + +export function applyPlaybackEvent( + state: MessageTtsUiState, + event: MessageTtsBackendEvent, +): { state: MessageTtsUiState; detachedError: string | null }; + +export function isMessagePlaying( + state: MessageTtsUiState, + messageId: string, +): boolean; diff --git a/desktop/src/features/message-tts/lib/playbackReducer.mjs b/desktop/src/features/message-tts/lib/playbackReducer.mjs new file mode 100644 index 0000000000..59a95798d3 --- /dev/null +++ b/desktop/src/features/message-tts/lib/playbackReducer.mjs @@ -0,0 +1,76 @@ +/** + * Pure state reducer for message read-aloud playback. + * + * The backend (`message_tts.rs`) emits `message-tts-state-changed` events; + * this reducer folds them into the single-playback UI state the provider + * exposes. Kept in `.mjs` (like `applyEditTagOverlay.mjs`) so the + * TS-loader-less `node:test` runner imports the same source production uses; + * typed access via the sibling `.d.mts`. + * + * Contract (mirrors the backend's generation guarantees): + * - At most one message is "current" at a time. + * - "preparing"/"playing" for message X makes X current, replacing anything. + * - Terminal events ("completed" | "stopped" | "superseded") only clear the + * state when they name the CURRENT message — a stale terminal event for an + * already-replaced message must not clear the newer playback. + * - "failed" for the current message keeps the id and carries the error so + * the row can surface it; "failed" for a non-current message becomes a + * detached error (toast-only), leaving current playback untouched. + */ + +export const IDLE_STATE = Object.freeze({ + messageId: null, + phase: "idle", // "idle" | "preparing" | "playing" + truncated: false, + error: null, +}); + +/** + * Fold one backend playback event into the current UI state. + * Returns `{ state, detachedError }`; `detachedError` is a failure for a + * message that no longer owns playback (surface as a toast, not row state). + */ +export function applyPlaybackEvent(state, event) { + const { message_id: messageId, status, truncated, error } = event; + + switch (status) { + case "preparing": + case "playing": + return { + state: { + messageId, + phase: status, + truncated: Boolean(truncated), + error: null, + }, + detachedError: null, + }; + case "completed": + case "stopped": + case "superseded": + if (state.messageId === messageId) { + return { state: IDLE_STATE, detachedError: null }; + } + return { state, detachedError: null }; + case "failed": { + const message = error || "playback failed"; + if (state.messageId === messageId || state.messageId === null) { + return { + state: { messageId, phase: "idle", truncated: false, error: message }, + detachedError: message, + }; + } + return { state, detachedError: message }; + } + default: + return { state, detachedError: null }; + } +} + +/** Whether the play affordance for `messageId` should render as "playing". */ +export function isMessagePlaying(state, messageId) { + return ( + state.messageId === messageId && + (state.phase === "preparing" || state.phase === "playing") + ); +} diff --git a/desktop/src/features/message-tts/lib/playbackReducer.test.mjs b/desktop/src/features/message-tts/lib/playbackReducer.test.mjs new file mode 100644 index 0000000000..6260f7a1e3 --- /dev/null +++ b/desktop/src/features/message-tts/lib/playbackReducer.test.mjs @@ -0,0 +1,134 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + applyPlaybackEvent, + IDLE_STATE, + isMessagePlaying, +} from "./playbackReducer.mjs"; + +function ev(messageId, status, extra = {}) { + return { + message_id: messageId, + status, + truncated: false, + error: null, + ...extra, + }; +} + +test("preparing makes the message current", () => { + const { state } = applyPlaybackEvent(IDLE_STATE, ev("a", "preparing")); + assert.equal(state.messageId, "a"); + assert.equal(state.phase, "preparing"); + assert.equal(isMessagePlaying(state, "a"), true); + assert.equal(isMessagePlaying(state, "b"), false); +}); + +test("playing keeps the message current and flips the phase", () => { + let { state } = applyPlaybackEvent(IDLE_STATE, ev("a", "preparing")); + ({ state } = applyPlaybackEvent(state, ev("a", "playing"))); + assert.equal(state.phase, "playing"); + assert.equal(isMessagePlaying(state, "a"), true); +}); + +test("completed for the current message returns to idle", () => { + let { state } = applyPlaybackEvent(IDLE_STATE, ev("a", "playing")); + ({ state } = applyPlaybackEvent(state, ev("a", "completed"))); + assert.deepEqual(state, IDLE_STATE); +}); + +test("stopped for the current message returns to idle", () => { + let { state } = applyPlaybackEvent(IDLE_STATE, ev("a", "playing")); + ({ state } = applyPlaybackEvent(state, ev("a", "stopped"))); + assert.deepEqual(state, IDLE_STATE); +}); + +test("supersede race: newer playback is never cleared by the old one's terminal event", () => { + // play a → play b arrives (preparing b + superseded a can arrive in either + // order) → a's late "superseded"/"completed" must not clear b. + let { state } = applyPlaybackEvent(IDLE_STATE, ev("a", "playing")); + ({ state } = applyPlaybackEvent(state, ev("b", "preparing"))); + assert.equal(state.messageId, "b"); + + ({ state } = applyPlaybackEvent(state, ev("a", "superseded"))); + assert.equal(state.messageId, "b", "stale superseded must not clear b"); + + ({ state } = applyPlaybackEvent(state, ev("a", "completed"))); + assert.equal(state.messageId, "b", "stale completed must not clear b"); + + assert.equal(isMessagePlaying(state, "b"), true); + assert.equal(isMessagePlaying(state, "a"), false); +}); + +test("superseded arriving before the new preparing also converges", () => { + let { state } = applyPlaybackEvent(IDLE_STATE, ev("a", "playing")); + ({ state } = applyPlaybackEvent(state, ev("a", "superseded"))); + assert.deepEqual(state, IDLE_STATE); + ({ state } = applyPlaybackEvent(state, ev("b", "preparing"))); + assert.equal(state.messageId, "b"); +}); + +test("failed for the current message carries the error and stops the playing affordance", () => { + let { state } = applyPlaybackEvent(IDLE_STATE, ev("a", "preparing")); + const result = applyPlaybackEvent( + state, + ev("a", "failed", { error: "voice model is not downloaded yet" }), + ); + state = result.state; + assert.equal(state.messageId, "a"); + assert.equal(state.phase, "idle"); + assert.equal(state.error, "voice model is not downloaded yet"); + assert.equal(result.detachedError, "voice model is not downloaded yet"); + assert.equal(isMessagePlaying(state, "a"), false); +}); + +test("failed for a non-current message leaves current playback untouched", () => { + const { state } = applyPlaybackEvent(IDLE_STATE, ev("b", "playing")); + const result = applyPlaybackEvent( + state, + ev("a", "failed", { error: "boom" }), + ); + assert.equal(result.state.messageId, "b"); + assert.equal(result.state.phase, "playing"); + assert.equal(result.detachedError, "boom"); +}); + +test("preflight failure from idle surfaces as a detached error", () => { + const result = applyPlaybackEvent( + IDLE_STATE, + ev("a", "failed", { error: "message has no speakable text" }), + ); + assert.equal(result.detachedError, "message has no speakable text"); + assert.equal(isMessagePlaying(result.state, "a"), false); +}); + +test("failed without error text falls back to a generic message", () => { + const result = applyPlaybackEvent(IDLE_STATE, ev("a", "failed")); + assert.equal(result.detachedError, "playback failed"); +}); + +test("truncated flag is carried on preparing/playing", () => { + const { state } = applyPlaybackEvent( + IDLE_STATE, + ev("a", "preparing", { truncated: true }), + ); + assert.equal(state.truncated, true); +}); + +test("unknown status is ignored", () => { + const before = applyPlaybackEvent(IDLE_STATE, ev("a", "playing")).state; + const { state } = applyPlaybackEvent(before, ev("a", "warming-up")); + assert.equal(state, before); +}); + +test("rapid play/stop/play across messages always tracks the latest", () => { + let state = IDLE_STATE; + for (let i = 0; i < 20; i++) { + ({ state } = applyPlaybackEvent(state, ev(`m${i}`, "preparing"))); + // Stale events from the previous message trickle in afterwards. + ({ state } = applyPlaybackEvent(state, ev(`m${i - 1}`, "superseded"))); + ({ state } = applyPlaybackEvent(state, ev(`m${i - 1}`, "completed"))); + assert.equal(state.messageId, `m${i}`); + } +}); diff --git a/desktop/src/features/message-tts/ui/MessageTtsProvider.tsx b/desktop/src/features/message-tts/ui/MessageTtsProvider.tsx new file mode 100644 index 0000000000..49c6d3807b --- /dev/null +++ b/desktop/src/features/message-tts/ui/MessageTtsProvider.tsx @@ -0,0 +1,123 @@ +import { invoke } from "@tauri-apps/api/core"; +import { listen } from "@tauri-apps/api/event"; +import * as React from "react"; +import { toast } from "sonner"; + +import { useHuddle } from "@/features/huddle"; +import { + applyPlaybackEvent, + IDLE_STATE, + isMessagePlaying, + type MessageTtsBackendEvent, + type MessageTtsUiState, +} from "@/features/message-tts/lib/playbackReducer.mjs"; + +/** + * App-scoped read-aloud state for chat messages. + * + * One playback app-wide: pressing play on message B while A speaks + * supersedes A (the backend cancels A's audio and emits "superseded"; the + * reducer keeps state keyed to B). All state transitions come FROM the + * backend via `message-tts-state-changed` — the provider never guesses; + * a play click shows feedback only when the backend confirms "preparing". + */ +type MessageTtsContextValue = { + /** Whether `messageId` currently owns playback (preparing or playing). */ + isPlaying: (messageId: string) => boolean; + /** True while the current playback's text was cut at the length limit. */ + isTruncated: (messageId: string) => boolean; + /** Start reading `body` aloud, superseding any current playback. */ + play: (messageId: string, body: string) => void; + /** Stop the current playback (no-op when nothing is playing). */ + stop: () => void; + /** Non-null while read-aloud is unavailable (e.g. during a huddle). */ + disabledReason: string | null; +}; + +const MessageTtsContext = React.createContext( + null, +); + +export function useMessageTts(): MessageTtsContextValue | null { + // Nullable on purpose: surfaces (like web builds or tests) without the + // provider simply hide the read-aloud affordance. + return React.useContext(MessageTtsContext); +} + +export function MessageTtsProvider({ + children, +}: { + children: React.ReactNode; +}) { + const [state, setState] = React.useState(IDLE_STATE); + // Provider is mounted inside HuddleProvider in AppShell. A live huddle owns + // the audio device, so read-aloud is disabled while connected (the backend + // enforces this too — the UI gate just explains it before the click). + const { activeEphemeralChannelId } = useHuddle(); + const disabledReason = activeEphemeralChannelId + ? "Read aloud is unavailable during a huddle" + : null; + + React.useEffect(() => { + let cancelled = false; + let unlisten: (() => void) | null = null; + + listen("message-tts-state-changed", (event) => { + if (cancelled) { + return; + } + setState((prev) => { + const { state: next, detachedError } = applyPlaybackEvent( + prev, + event.payload, + ); + if (detachedError) { + toast.error(`Read aloud failed: ${detachedError}`); + } + // Long-message limit disclosure: the backend caps very long messages + // and reports the cut via `truncated` (the audio also ends with a + // spoken "message truncated"). Announce once, when playback starts. + if (event.payload.status === "preparing" && event.payload.truncated) { + toast.info("This message is long — only the beginning will be read."); + } + return next; + }); + }).then((fn) => { + if (cancelled) fn(); + else unlisten = fn; + }); + + return () => { + cancelled = true; + unlisten?.(); + }; + }, []); + + const play = React.useCallback((messageId: string, body: string) => { + // Errors also arrive as "failed" events (handled above with a toast); + // the rejected promise carries the same text, so swallow it here. + invoke("speak_chat_message", { messageId, text: body }).catch(() => {}); + }, []); + + const stop = React.useCallback(() => { + invoke("stop_chat_message_speech").catch(() => {}); + }, []); + + const contextValue = React.useMemo( + () => ({ + isPlaying: (messageId: string) => isMessagePlaying(state, messageId), + isTruncated: (messageId: string) => + state.messageId === messageId && state.truncated, + play, + stop, + disabledReason, + }), + [state, play, stop, disabledReason], + ); + + return ( + + {children} + + ); +} diff --git a/desktop/src/features/messages/ui/MessageActionBar.tsx b/desktop/src/features/messages/ui/MessageActionBar.tsx index ba45bc62fb..2bafed3581 100644 --- a/desktop/src/features/messages/ui/MessageActionBar.tsx +++ b/desktop/src/features/messages/ui/MessageActionBar.tsx @@ -11,7 +11,9 @@ import { MailOpen, Pencil, SmilePlus, + Square, Trash2, + Volume2, } from "lucide-react"; import * as React from "react"; @@ -19,6 +21,7 @@ import { buildMessageLink } from "@/features/messages/lib/messageLink"; import { EmojiPicker } from "@/features/custom-emoji/ui/EmojiPicker"; import { useCustomEmoji } from "@/features/custom-emoji/hooks"; import { getThreadReference } from "@/features/messages/lib/threading"; +import { useMessageTts } from "@/features/message-tts/ui/MessageTtsProvider"; import { ReportMessageDialog } from "@/features/moderation/ui/ReportMessageDialog"; import { MessageModerationMenuItems } from "@/features/moderation/ui/MessageModerationMenuItems"; import type { @@ -365,6 +368,74 @@ function isCustomEmojiShortcode(emoji: string) { return emoji.startsWith(":") && emoji.endsWith(":"); } +/** + * Play/Stop read-aloud toggle for delivered agent messages. + * + * Renders nothing when the MessageTtsProvider is absent (web builds, tests) + * or when the message is not delivered agent text — read-aloud is scoped to + * agent responses in v1. While a huddle is live the button stays visible but + * disabled, with the tooltip explaining why (the backend refuses too). + */ +function ReadAloudButton({ message }: { message: TimelineMessage }) { + const messageTts = useMessageTts(); + + if (!messageTts) { + return null; + } + // Delivered agent text only: no pending sends, no system huddle rows, + // nothing without a body to speak. + if ( + !message.isAgent || + message.pending || + message.kind === KIND_HUDDLE_STARTED || + !message.body.trim() + ) { + return null; + } + + const playing = messageTts.isPlaying(message.id); + const disabled = messageTts.disabledReason !== null; + const label = playing ? "Stop reading aloud" : "Read message aloud"; + const tooltip = disabled + ? (messageTts.disabledReason ?? label) + : playing + ? "Stop" + : "Read aloud"; + + return ( + + + + + {tooltip} + + ); +} + export const MessageActionBar = React.memo(function MessageActionBar({ channelId, message, @@ -546,6 +617,8 @@ export const MessageActionBar = React.memo(function MessageActionBar({ ) : null} + + {hasReplyAction ? (