fix(desktop): preserve playback speed continuity

Signed-off-by: John Tennant <jtennant@squareup.com>

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

@@ -21,7 +21,9 @@ use crate::app_state::AppState;
 
 #[path = "playback_speed_dsp.rs"]
 mod playback_speed_dsp;
+#[cfg(test)]
 pub(crate) use playback_speed_dsp::process_complete_chunk;
+pub(crate) use playback_speed_dsp::process_complete_playback_chunk;
 use playback_speed_dsp::{validate_speed, DEFAULT_PLAYBACK_SPEED};
 
 const SETTINGS_FILE: &str = "tts-playback-settings.json";

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

@@ -71,6 +71,28 @@ pub fn process_complete_chunk(
     Ok(output[0][start..end].to_vec())
 }
 
+/// Pitch-preserve one complete playback chunk across its model-unit splits.
+///
+/// Pocket may divide a natural synthesis chunk into multiple model-valid
+/// units. Joining those units before processing keeps one stretcher timeline
+/// across the hidden boundaries instead of resetting the DSP at each unit.
+pub fn process_complete_playback_chunk(
+    model_units: &[Vec<f32>],
+    speed: f32,
+    sample_rate: u32,
+) -> Result<Vec<f32>, String> {
+    let sample_count = model_units.iter().try_fold(0_usize, |total, unit| {
+        total
+            .checked_add(unit.len())
+            .ok_or_else(|| "audio chunk is too large to process".to_string())
+    })?;
+    let mut samples = Vec::with_capacity(sample_count);
+    for unit in model_units {
+        samples.extend_from_slice(unit);
+    }
+    process_complete_chunk(&samples, speed, sample_rate)
+}
+
 /// Return Signalsmith's compensated output lookahead for descriptive reporting.
 #[allow(dead_code)]
 pub(crate) fn compensated_output_latency_samples(sample_rate: u32) -> usize {

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

@@ -47,7 +47,7 @@ use std::{
     time::Duration,
 };
 
-use super::playback_speed::{process_complete_chunk, PlaybackSpeedControl};
+use super::playback_speed::{process_complete_playback_chunk, PlaybackSpeedControl};
 use super::pocket::{
     load_text_to_speech, load_voice_style, DEFAULT_VOICE, SAMPLE_RATE, VOICE_FILE_EXT,
 };
@@ -714,7 +714,8 @@ fn tts_worker(
                 );
                 continue;
             }
-            let mut playback_audio = PlaybackChunkAudio::new();
+            let mut playback_units = Vec::with_capacity(model_chunks.len());
+            let mut first_model_unit_index = None;
             for model_chunk in &model_chunks {
                 let chunk_index = model_unit_index;
                 model_unit_index += 1;
@@ -758,32 +759,8 @@ fn tts_worker(
                 }
                 match synthesis {
                     Ok(samples) if !samples.is_empty() => {
-                        let speed = playback_speed.get();
-                        let samples = match process_complete_chunk(&samples, speed, SAMPLE_RATE) {
-                            Ok(processed) => processed,
-                            Err(error) => {
-                                eprintln!(
-                                    "buzz-desktop: TTS playback-speed processing failed at \
-                                     {speed:.2}x: {error}; using 1x playback"
-                                );
-                                samples
-                            }
-                        };
-                        if let Some(prepared) = playback_audio.push(
-                            samples,
-                            chunk_index,
-                            &mut first_append,
-                            silence_buf_len,
-                            player.empty(),
-                        ) {
-                            if !append_audio(prepared, route_id) {
-                                first_append = true;
-                                synthesis_outcome = "cancelled";
-                                break 'playback_chunks;
-                            }
-                            appended_audio = true;
-                            last_route_id = route_id;
-                        }
+                        first_model_unit_index.get_or_insert(chunk_index);
+                        playback_units.push(samples);
                     }
                     Ok(_) => {
                         eprintln!(
@@ -799,16 +776,40 @@ fn tts_worker(
                     }
                 }
             }
-            if let Some(prepared) =
-                playback_audio.finish(&mut first_append, silence_buf_len, player.empty())
-            {
-                if !append_audio(prepared, route_id) {
-                    first_append = true;
-                    synthesis_outcome = "cancelled";
-                    break 'playback_chunks;
+            if let Some(chunk_index) = first_model_unit_index {
+                let speed = playback_speed.get();
+                let samples =
+                    match process_complete_playback_chunk(&playback_units, speed, SAMPLE_RATE) {
+                        Ok(processed) => processed,
+                        Err(error) => {
+                            eprintln!(
+                                "buzz-desktop: TTS playback-speed processing failed at \
+                             {speed:.2}x: {error}; using 1x playback"
+                            );
+                            playback_units.into_iter().flatten().collect()
+                        }
+                    };
+                let mut playback_audio = PlaybackChunkAudio::new();
+                debug_assert!(playback_audio
+                    .push(
+                        samples,
+                        chunk_index,
+                        &mut first_append,
+                        silence_buf_len,
+                        player.empty(),
+                    )
+                    .is_none());
+                if let Some(prepared) =
+                    playback_audio.finish(&mut first_append, silence_buf_len, player.empty())
+                {
+                    if !append_audio(prepared, route_id) {
+                        first_append = true;
+                        synthesis_outcome = "cancelled";
+                        break 'playback_chunks;
+                    }
+                    appended_audio = true;
+                    last_route_id = route_id;
                 }
-                appended_audio = true;
-                last_route_id = route_id;
             }
             if synthesis_outcome == "failed" {
                 break 'playback_chunks;

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

@@ -1,4 +1,5 @@
 use super::*;
+use crate::huddle::playback_speed::process_complete_chunk;
 
 /// Regression guard: playback decoration happens after speed processing, so
 /// the fixed device cushion is never time-stretched.
@@ -23,3 +24,30 @@ fn playback_speed_preserves_production_sentence_lead_in() {
         "speech energy must remain after the fixed device cushion"
     );
 }
+
+#[test]
+fn playback_speed_keeps_one_timeline_across_model_units() {
+    let frequency = 220.0_f32;
+    let input: Vec<f32> = (0..SAMPLE_RATE)
+        .map(|sample| {
+            (2.0 * std::f32::consts::PI * frequency * sample as f32 / SAMPLE_RATE as f32).sin()
+        })
+        .collect();
+    let units = vec![input[..12_000].to_vec(), input[12_000..].to_vec()];
+
+    let joined = process_complete_playback_chunk(&units, 1.25, SAMPLE_RATE)
+        .expect("process complete playback chunk");
+    let expected =
+        process_complete_chunk(&input, 1.25, SAMPLE_RATE).expect("process already-joined input");
+
+    assert_eq!(joined, expected, "model units must share one stretcher run");
+    let boundary = joined.len() / 2;
+    let largest_step = joined[boundary - 240..boundary + 240]
+        .windows(2)
+        .map(|pair| (pair[1] - pair[0]).abs())
+        .fold(0.0_f32, f32::max);
+    assert!(
+        largest_step < 0.1,
+        "hidden model boundary introduced a {largest_step} sample discontinuity"
+    );
+}

desktop/src/features/settings/ui/SpeechPlaybackSettings.tsx

@@ -1,14 +1,10 @@
-import { useCallback, useEffect, useRef, useState } from "react";
+import { useCallback, useEffect, useState } from "react";
 import {
   getTtsPlaybackSpeed,
   setTtsPlaybackSpeed,
 } from "@/shared/api/ttsPlayback";
 import { Button } from "@/shared/ui/button";
-import {
-  applyLoadedPlaybackSpeed,
-  cancelPlaybackSpeedPersistence,
-  commitPlaybackSpeed,
-} from "./playbackSpeedPersistence";
+import { createPlaybackSpeedPersistence } from "./playbackSpeedPersistence";
 import { SettingsOptionGroup, SettingsOptionRow } from "./SettingsOptionGroup";
 
 const DEFAULT_SPEED = 1;
@@ -17,40 +13,31 @@ const SLIDER_MAX_SPEED = 2;
 const INPUT_MIN_SPEED = 0.25;
 const INPUT_MAX_SPEED = 4;
 const SPEED_STEP = 0.05;
+const playbackSpeedPersistence = createPlaybackSpeedPersistence(
+  setTtsPlaybackSpeed,
+  DEFAULT_SPEED,
+);
 
 export function SpeechPlaybackSettings() {
   const [speed, setSpeed] = useState(DEFAULT_SPEED);
   const [speedInput, setSpeedInput] = useState(String(DEFAULT_SPEED));
   const [error, setError] = useState<string | null>(null);
   const [loaded, setLoaded] = useState(false);
-  const active = useRef(true);
-  const confirmedSpeed = useRef(DEFAULT_SPEED);
-  const desiredSpeed = useRef(DEFAULT_SPEED);
-  const flushPromise = useRef<Promise<void> | null>(null);
-  const hasLocalIntent = useRef(false);
-
   const setDisplayedSpeed = useCallback((nextSpeed: number) => {
     setSpeed(nextSpeed);
     setSpeedInput(String(nextSpeed));
   }, []);
 
   useEffect(() => {
-    active.current = true;
     let loadActive = true;
+    const unsubscribe = playbackSpeedPersistence.subscribe({
+      setError,
+      setSpeed: setDisplayedSpeed,
+    });
     getTtsPlaybackSpeed()
       .then((savedSpeed) => {
         if (loadActive) {
-          applyLoadedPlaybackSpeed(
-            {
-              active,
-              confirmedSpeed,
-              desiredSpeed,
-              flushPromise,
-              hasLocalIntent,
-            },
-            { setSpeed: setDisplayedSpeed },
-            savedSpeed,
-          );
+          playbackSpeedPersistence.applyLoaded(savedSpeed);
         }
       })
       .catch((cause) => {
@@ -61,28 +48,12 @@ export function SpeechPlaybackSettings() {
       });
     return () => {
       loadActive = false;
-      cancelPlaybackSpeedPersistence({
-        active,
-        confirmedSpeed,
-        desiredSpeed,
-        flushPromise,
-        hasLocalIntent,
-      });
+      unsubscribe();
     };
   }, [setDisplayedSpeed]);
 
   const commitSpeed = (nextSpeed: number) => {
-    commitPlaybackSpeed(
-      {
-        active,
-        confirmedSpeed,
-        desiredSpeed,
-        flushPromise,
-        hasLocalIntent,
-      },
-      { persist: setTtsPlaybackSpeed, setError, setSpeed: setDisplayedSpeed },
-      nextSpeed,
-    );
+    playbackSpeedPersistence.commit(nextSpeed);
   };
 
   const commitTypedSpeed = () => {

desktop/src/features/settings/ui/playbackSpeedPersistence.test.mjs

@@ -1,11 +1,7 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 
-import {
-  applyLoadedPlaybackSpeed,
-  cancelPlaybackSpeedPersistence,
-  commitPlaybackSpeed,
-} from "./playbackSpeedPersistence.ts";
+import { createPlaybackSpeedPersistence } from "./playbackSpeedPersistence.ts";
 
 function deferred() {
   let resolve;
@@ -19,110 +15,99 @@ function deferred() {
 
 function harness() {
   const calls = [];
-  const errors = [];
   const saves = [];
-  const speeds = [];
-  const state = {
-    active: { current: true },
-    confirmedSpeed: { current: 1 },
-    desiredSpeed: { current: 1 },
-    flushPromise: { current: null },
-    hasLocalIntent: { current: false },
+  const persist = (speed) => {
+    calls.push(speed);
+    const save = deferred();
+    saves.push(save);
+    return save.promise;
   };
-  const callbacks = {
-    persist: (speed) => {
-      calls.push(speed);
-      const save = deferred();
-      saves.push(save);
-      return save.promise;
-    },
-    setError: (error) => errors.push(error),
-    setSpeed: (speed) => speeds.push(speed),
+  return {
+    calls,
+    controller: createPlaybackSpeedPersistence(persist),
+    saves,
   };
-  return { callbacks, calls, errors, saves, speeds, state };
 }
 
-async function settle(state) {
-  while (state.flushPromise.current) {
-    await state.flushPromise.current;
-  }
+function observe(controller) {
+  const errors = [];
+  const speeds = [];
+  const unsubscribe = controller.subscribe({
+    setError: (error) => errors.push(error),
+    setSpeed: (speed) => speeds.push(speed),
+  });
+  return { errors, speeds, unsubscribe };
 }
 
 test("rapid intents persist serially and finish at the latest speed", async () => {
   const h = harness();
-  commitPlaybackSpeed(h.state, h.callbacks, 1.25);
-  commitPlaybackSpeed(h.state, h.callbacks, 1.5);
+  const listener = observe(h.controller);
+  h.controller.commit(1.25);
+  h.controller.commit(1.5);
   assert.deepEqual(h.calls, [1.25]);
 
   h.saves[0].resolve();
   await Promise.resolve();
   assert.deepEqual(h.calls, [1.25, 1.5]);
   h.saves[1].resolve();
-  await settle(h.state);
+  await Promise.resolve();
 
-  assert.equal(h.state.confirmedSpeed.current, 1.5);
-  assert.equal(h.state.desiredSpeed.current, 1.5);
+  assert.equal(listener.speeds.at(-1), 1.5);
 });
 
 test("a stale failure does not roll back a newer successful intent", async () => {
   const h = harness();
-  commitPlaybackSpeed(h.state, h.callbacks, 1.25);
-  commitPlaybackSpeed(h.state, h.callbacks, 1.5);
+  const listener = observe(h.controller);
+  h.controller.commit(1.25);
+  h.controller.commit(1.5);
 
   h.saves[0].reject(new Error("first failed"));
   await Promise.resolve();
   assert.deepEqual(h.calls, [1.25, 1.5]);
   h.saves[1].resolve();
-  await settle(h.state);
+  await Promise.resolve();
 
-  assert.equal(h.state.confirmedSpeed.current, 1.5);
-  assert.equal(h.state.desiredSpeed.current, 1.5);
-  assert.deepEqual(h.speeds, [1.25, 1.5]);
-  assert.deepEqual(h.errors, [null, null]);
+  assert.equal(listener.speeds.at(-1), 1.5);
+  assert.equal(listener.errors.at(-1), null);
 });
 
 test("the latest failure rolls back to the last confirmed speed", async () => {
   const h = harness();
-  commitPlaybackSpeed(h.state, h.callbacks, 1.25);
+  const listener = observe(h.controller);
+  h.controller.commit(1.25);
   h.saves[0].reject(new Error("save failed"));
-  await settle(h.state);
+  await Promise.resolve();
+  await Promise.resolve();
 
-  assert.equal(h.state.confirmedSpeed.current, 1);
-  assert.equal(h.state.desiredSpeed.current, 1);
-  assert.deepEqual(h.speeds, [1.25, 1]);
-  assert.equal(h.errors.at(-1), "Error: save failed");
+  assert.equal(listener.speeds.at(-1), 1);
+  assert.equal(listener.errors.at(-1), "Error: save failed");
 });
 
 test("a delayed initial load cannot overwrite a local intent", async () => {
   const h = harness();
-  commitPlaybackSpeed(h.state, h.callbacks, 1.5);
-  applyLoadedPlaybackSpeed(h.state, h.callbacks, 1);
+  const listener = observe(h.controller);
+  h.controller.commit(1.5);
+  h.controller.applyLoaded(1);
 
-  assert.equal(h.state.desiredSpeed.current, 1.5);
-  assert.deepEqual(h.speeds, [1.5]);
+  assert.equal(listener.speeds.at(-1), 1.5);
+  h.saves[0].resolve();
+  await Promise.resolve();
+});
+
+test("the latest intent survives unmount and is visible after remount", async () => {
+  const h = harness();
+  const firstMount = observe(h.controller);
+  h.controller.commit(1.25);
+  h.controller.commit(1.5);
+  firstMount.unsubscribe();
 
   h.saves[0].resolve();
-  await settle(h.state);
-  assert.equal(h.state.confirmedSpeed.current, 1.5);
-  assert.equal(h.state.desiredSpeed.current, 1.5);
-});
-
-test("an unmounted instance cannot persist stale queued intent", async () => {
-  const oldInstance = harness();
-  commitPlaybackSpeed(oldInstance.state, oldInstance.callbacks, 1.25);
-  commitPlaybackSpeed(oldInstance.state, oldInstance.callbacks, 1.5);
-  cancelPlaybackSpeedPersistence(oldInstance.state);
-
-  const newInstance = harness();
-  commitPlaybackSpeed(newInstance.state, newInstance.callbacks, 0.75);
-  assert.deepEqual(oldInstance.calls, [1.25]);
-  assert.deepEqual(newInstance.calls, [0.75]);
-
-  oldInstance.saves[0].resolve();
-  await settle(oldInstance.state);
-  assert.deepEqual(oldInstance.calls, [1.25]);
-
-  newInstance.saves[0].resolve();
-  await settle(newInstance.state);
-  assert.equal(newInstance.state.confirmedSpeed.current, 0.75);
+  await Promise.resolve();
+  assert.deepEqual(h.calls, [1.25, 1.5]);
+  h.saves[1].resolve();
+  await Promise.resolve();
+
+  const secondMount = observe(h.controller);
+  assert.equal(secondMount.speeds.at(-1), 1.5);
+  assert.equal(secondMount.errors.at(-1), null);
 });

desktop/src/features/settings/ui/playbackSpeedPersistence.ts

@@ -1,91 +1,72 @@
-type MutableValue<T> = {
-  current: T;
-};
-
-export type PlaybackSpeedPersistenceState = {
-  active: MutableValue<boolean>;
-  confirmedSpeed: MutableValue<number>;
-  desiredSpeed: MutableValue<number>;
-  flushPromise: MutableValue<Promise<void> | null>;
-  hasLocalIntent: MutableValue<boolean>;
-};
-
 type PlaybackSpeedPersistenceCallbacks = {
-  persist: (speed: number) => Promise<void>;
   setSpeed: (speed: number) => void;
   setError: (error: string | null) => void;
 };
 
-export function commitPlaybackSpeed(
-  state: PlaybackSpeedPersistenceState,
-  callbacks: PlaybackSpeedPersistenceCallbacks,
-  nextSpeed: number,
+export function createPlaybackSpeedPersistence(
+  persist: (speed: number) => Promise<void>,
+  initialSpeed = 1,
 ) {
-  if (!state.active.current) return;
-  state.hasLocalIntent.current = true;
-  state.desiredSpeed.current = nextSpeed;
-  callbacks.setSpeed(nextSpeed);
-  callbacks.setError(null);
-  ensurePlaybackSpeedFlush(state, callbacks);
-}
+  let confirmedSpeed = initialSpeed;
+  let desiredSpeed = initialSpeed;
+  let error: string | null = null;
+  let flushPromise: Promise<void> | null = null;
+  let hasLocalIntent = false;
+  const listeners = new Set<PlaybackSpeedPersistenceCallbacks>();
 
-export function cancelPlaybackSpeedPersistence(
-  state: PlaybackSpeedPersistenceState,
-) {
-  state.active.current = false;
-}
-
-export function applyLoadedPlaybackSpeed(
-  state: PlaybackSpeedPersistenceState,
-  callbacks: Pick<PlaybackSpeedPersistenceCallbacks, "setSpeed">,
-  savedSpeed: number,
-) {
-  if (state.hasLocalIntent.current) return;
-  state.confirmedSpeed.current = savedSpeed;
-  state.desiredSpeed.current = savedSpeed;
-  callbacks.setSpeed(savedSpeed);
-}
-
-function ensurePlaybackSpeedFlush(
-  state: PlaybackSpeedPersistenceState,
-  callbacks: PlaybackSpeedPersistenceCallbacks,
-) {
-  if (!state.active.current || state.flushPromise.current) return;
-
-  const flush = flushPlaybackSpeed(state, callbacks).finally(() => {
-    if (state.flushPromise.current !== flush) return;
-    state.flushPromise.current = null;
-    if (
-      state.active.current &&
-      state.desiredSpeed.current !== state.confirmedSpeed.current
-    ) {
-      ensurePlaybackSpeedFlush(state, callbacks);
+  const notify = () => {
+    for (const listener of listeners) {
+      listener.setSpeed(desiredSpeed);
+      listener.setError(error);
     }
-  });
-  state.flushPromise.current = flush;
-}
+  };
 
-async function flushPlaybackSpeed(
-  state: PlaybackSpeedPersistenceState,
-  callbacks: PlaybackSpeedPersistenceCallbacks,
-) {
-  while (
-    state.active.current &&
-    state.desiredSpeed.current !== state.confirmedSpeed.current
-  ) {
-    const target = state.desiredSpeed.current;
-    try {
-      await callbacks.persist(target);
-      if (!state.active.current) return;
-      state.confirmedSpeed.current = target;
-    } catch (cause) {
-      if (!state.active.current) return;
-      if (state.desiredSpeed.current === target) {
-        state.desiredSpeed.current = state.confirmedSpeed.current;
-        callbacks.setSpeed(state.confirmedSpeed.current);
-        callbacks.setError(String(cause));
-        return;
+  const flush = async () => {
+    while (desiredSpeed !== confirmedSpeed) {
+      const target = desiredSpeed;
+      try {
+        await persist(target);
+        confirmedSpeed = target;
+      } catch (cause) {
+        if (desiredSpeed === target) {
+          desiredSpeed = confirmedSpeed;
+          error = String(cause);
+          notify();
+          return;
+        }
       }
     }
-  }
+  };
+
+  const ensureFlush = () => {
+    if (flushPromise) return;
+    const currentFlush = flush().finally(() => {
+      if (flushPromise !== currentFlush) return;
+      flushPromise = null;
+      if (desiredSpeed !== confirmedSpeed) ensureFlush();
+    });
+    flushPromise = currentFlush;
+  };
+
+  return {
+    applyLoaded(savedSpeed: number) {
+      if (hasLocalIntent) return;
+      confirmedSpeed = savedSpeed;
+      desiredSpeed = savedSpeed;
+      notify();
+    },
+    commit(nextSpeed: number) {
+      hasLocalIntent = true;
+      desiredSpeed = nextSpeed;
+      error = null;
+      notify();
+      ensureFlush();
+    },
+    subscribe(callbacks: PlaybackSpeedPersistenceCallbacks) {
+      listeners.add(callbacks);
+      callbacks.setSpeed(desiredSpeed);
+      callbacks.setError(error);
+      return () => listeners.delete(callbacks);
+    },
+  };
 }