diff --git a/crates/tools/servers/agent_notify_server/README.md b/crates/tools/servers/agent_notify_server/README.md index 13af5aa897..5007b11556 100644 --- a/crates/tools/servers/agent_notify_server/README.md +++ b/crates/tools/servers/agent_notify_server/README.md @@ -45,14 +45,32 @@ Mixing rules: alert_beep_sound: test_beep.wav alert_done_sound: smrpg_flower.wav alert_await_user_input_sound: smrpg_ghost.wav + +extra_alert_beep_sounds: [] +extra_alert_done_sounds: + - smrpg_specialflower.wav + - smrpg_correct.wav +extra_alert_await_sounds: + - smrpg_wrong.wav + - smrpg_drybones_crumble.wav + loop_alert_timeout_millis: 2000 +escalate_wait_1: 15 +escalate_wait_2: 30 +escalate_wait_3: 45 ``` - Paths can be absolute, or relative to the YAML file's directory. -- `loop_alert_timeout_millis` is the gap *between* consecutive plays of a - looping sound (after one play ends, before the next begins). Omit to replay - back-to-back. - WAV and MP3 are both supported (decoded via rodio + symphonia). +- `loop_alert_timeout_millis` is the gap *between* consecutive plays of a + single voice. Omit to replay back-to-back. +- **Escalation**: a `/loop_*` request starts one voice immediately. At + `escalate_wait_1` / `escalate_wait_2` / `escalate_wait_3` seconds, a + second / third / fourth concurrent voice joins the mix. New voices are + taken from `extra_alert__sounds` in order; when that pool is + exhausted the supervisor cycles back through the full pool + (primary + extras), so existing voices double up. Each voice runs in + its own thread and drifts naturally relative to the others. ## Claude Code wiring diff --git a/crates/tools/servers/agent_notify_server/config/notify_config.yaml b/crates/tools/servers/agent_notify_server/config/notify_config.yaml index 7beeceb110..e36ba0bb88 100644 --- a/crates/tools/servers/agent_notify_server/config/notify_config.yaml +++ b/crates/tools/servers/agent_notify_server/config/notify_config.yaml @@ -4,11 +4,34 @@ # the file is missing) the corresponding endpoint becomes a no-op. # # `loop_alert_timeout_millis` is the gap *between* consecutive plays of a -# looping sound (measured from the moment one play finishes to the moment the -# next play begins). If omitted, looping sounds replay back-to-back. +# single voice (after one play finishes, before the next begins). If +# omitted, looping sounds replay back-to-back. +# +# Escalation: after a /loop_* endpoint starts a loop, additional concurrent +# voices join the mix at the configured wait times. Each new voice is +# picked from `extra_alert__sounds` in order; if the pool is +# exhausted the supervisor cycles back through it, so existing voices +# double up. Voices are independent threads and drift naturally relative +# to each other. # Paths can be absolute, or relative to this YAML file's directory. alert_beep_sound: test_beep.wav alert_done_sound: /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_flower.wav alert_await_user_input_sound: /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_ghost.wav + +extra_alert_beep_sounds: [] +extra_alert_done_sounds: + - /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_specialflower.wav + - /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_correct.wav + - /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_character_extrapower.wav +extra_alert_await_sounds: + - /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_wrong.wav + - /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_drybones_crumble.wav + - /Users/bt/dev/storyteller/artcraft/frontend/apps/artcraft/app/public/resources/sound/smrpg_enemy_spikethrow.wav + loop_alert_timeout_millis: 2000 + +# Seconds-from-loop-start at which voices 2, 3, and 4 join the mix. +escalate_wait_1: 15 +escalate_wait_2: 30 +escalate_wait_3: 45 diff --git a/crates/tools/servers/agent_notify_server/config/test_beep.wav b/crates/tools/servers/agent_notify_server/config/test_beep.wav new file mode 100644 index 0000000000..d41cf401c4 Binary files /dev/null and b/crates/tools/servers/agent_notify_server/config/test_beep.wav differ diff --git a/crates/tools/servers/agent_notify_server/src/audio_player.rs b/crates/tools/servers/agent_notify_server/src/audio_player.rs index 1363b35aa5..fe70d7750c 100644 --- a/crates/tools/servers/agent_notify_server/src/audio_player.rs +++ b/crates/tools/servers/agent_notify_server/src/audio_player.rs @@ -5,19 +5,21 @@ //! handlers — interact with it through [`AudioPlayerHandle`], which forwards //! commands over an mpsc channel. //! -//! The engine maintains two independent mixing channels: +//! The engine maintains two kinds of mixing channels at the cpal output: //! //! - **`oneshot_sink`** — fire-and-forget sounds queued by `play_once`. Reused //! for the lifetime of the engine via `clear() + play()` so subsequent //! appends keep working after a `StopAll`. -//! - **A per-loop iterator thread** — owns its own `Sink` and replays the -//! configured sound, sleeping `gap_millis` between iterations. Replaced -//! atomically when a new loop sound is requested; signalled to stop via -//! `Arc` polled every 50ms (kept short so `/stop` and Ctrl+C -//! feel instant). +//! - **A loop session** — a supervisor thread that escalates over time. It +//! starts one looping iterator immediately, then adds a second, third, and +//! fourth concurrent iterator at the configured escalation times. Each +//! iterator owns its own `Sink` and runs in its own thread, so the voices +//! drift naturally relative to one another. Replacing the loop session +//! (`PlayLoop` while one is running) or `StopAll` signals every iterator +//! and the supervisor via a shared `Arc` polled every 50ms. //! -//! Because the two channels share the same `OutputStreamHandle`, a one-shot -//! sound naturally mixes with whatever is looping. +//! Because every channel shares the same `OutputStreamHandle`, one-shots +//! mix on top of however many loop voices are currently running. use std::fs::File; use std::io::BufReader; @@ -42,8 +44,8 @@ impl AudioPlayerHandle { let _ = self.tx.send(AudioCommand::PlayOnce(path)); } - pub fn play_loop(&self, path: PathBuf, gap_millis: u64) { - let _ = self.tx.send(AudioCommand::PlayLoop(path, gap_millis)); + pub fn play_loop(&self, spec: LoopSpec) { + let _ = self.tx.send(AudioCommand::PlayLoop(spec)); } pub fn stop_all(&self) { @@ -55,6 +57,18 @@ impl AudioPlayerHandle { } } +#[derive(Clone, Debug)] +pub struct LoopSpec { + /// Ordered pool of sounds: primary first, then extras. The supervisor + /// indexes this with `layer % pool.len()`, so when extras run out it + /// cycles back through the pool (doubling already-playing voices). + pub pool: Vec, + /// Gap between consecutive plays of a single iterator's sound. + pub gap_millis: u64, + /// Wall-clock seconds from loop start when layers 2, 3, 4 should join. + pub escalate_waits_secs: [u64; 3], +} + pub fn spawn_audio_player() -> (AudioPlayerHandle, JoinHandle<()>) { let (tx, rx) = mpsc::channel::(); let thread = thread::Builder::new() @@ -66,7 +80,7 @@ pub fn spawn_audio_player() -> (AudioPlayerHandle, JoinHandle<()>) { enum AudioCommand { PlayOnce(PathBuf), - PlayLoop(PathBuf, u64), + PlayLoop(LoopSpec), StopAll, Shutdown, } @@ -76,7 +90,6 @@ fn run_audio_engine(rx: Receiver) { Ok(s) => s, Err(e) => { log::error!("audio engine: failed to open default output stream: {}", e); - // Drain the channel so callers don't block, then exit. while rx.recv().is_ok() {} return; } @@ -100,9 +113,13 @@ fn run_audio_engine(rx: Receiver) { log::warn!("play_once {}: {}", path.display(), e); } } - AudioCommand::PlayLoop(path, gap_millis) => { + AudioCommand::PlayLoop(spec) => { stop_current_loop(&mut current_loop); - current_loop = Some(start_loop(&stream_handle, path, gap_millis)); + if spec.pool.is_empty() { + log::warn!("play_loop: empty sound pool, ignoring"); + continue; + } + current_loop = Some(start_loop_supervisor(&stream_handle, spec)); } AudioCommand::StopAll => { stop_current_loop(&mut current_loop); @@ -117,23 +134,18 @@ fn run_audio_engine(rx: Receiver) { } } - // Channel closed; clean up anyway. stop_current_loop(&mut current_loop); oneshot_sink.stop(); } -fn start_loop( - stream_handle: &OutputStreamHandle, - path: PathBuf, - gap_millis: u64, -) -> LoopController { +fn start_loop_supervisor(stream_handle: &OutputStreamHandle, spec: LoopSpec) -> LoopController { let stop = Arc::new(AtomicBool::new(false)); let stop_for_thread = stop.clone(); let stream_handle = stream_handle.clone(); let thread = thread::Builder::new() - .name("agent-notify-loop".to_string()) - .spawn(move || run_loop_iterator(stream_handle, path, gap_millis, stop_for_thread)) - .expect("spawn loop iterator thread"); + .name("agent-notify-loop-supervisor".to_string()) + .spawn(move || run_loop_supervisor(stream_handle, spec, stop_for_thread)) + .expect("spawn loop supervisor thread"); LoopController { stop, thread: Some(thread) } } @@ -142,12 +154,82 @@ fn stop_current_loop(current_loop: &mut Option) { lc.stop.store(true, Ordering::SeqCst); if let Some(thread) = lc.thread.take() { if let Err(e) = thread.join() { - log::warn!("audio loop thread panicked while shutting down: {:?}", e); + log::warn!("audio loop supervisor panicked while shutting down: {:?}", e); } } } } +fn run_loop_supervisor( + stream_handle: OutputStreamHandle, + spec: LoopSpec, + stop: Arc, +) { + let pool_len = spec.pool.len(); + if pool_len == 0 { + return; + } + + let mut iterators: Vec> = Vec::with_capacity(4); + iterators.push(spawn_iterator( + &stream_handle, + spec.pool[0].clone(), + spec.gap_millis, + stop.clone(), + 0, + )); + + // Convert absolute escalation times into deltas relative to the previous + // escalation. `saturating_sub` handles out-of-order config (e.g. wait_2 + // smaller than wait_1) by collapsing the interval to zero. + let waits = spec.escalate_waits_secs; + let intervals = [ + waits[0], + waits[1].saturating_sub(waits[0]), + waits[2].saturating_sub(waits[1]), + ]; + + for layer in 0..3usize { + if !sleep_with_stop(Duration::from_secs(intervals[layer]), &stop) { + break; + } + let pool_idx = (layer + 1) % pool_len; + iterators.push(spawn_iterator( + &stream_handle, + spec.pool[pool_idx].clone(), + spec.gap_millis, + stop.clone(), + layer + 1, + )); + } + + // Wait until something signals stop, then drain iterator threads. + while !stop.load(Ordering::SeqCst) { + thread::sleep(STOP_POLL_INTERVAL); + } + + for t in iterators { + if let Err(e) = t.join() { + log::warn!("audio loop iterator panicked: {:?}", e); + } + } +} + +fn spawn_iterator( + stream_handle: &OutputStreamHandle, + path: PathBuf, + gap_millis: u64, + stop: Arc, + layer: usize, +) -> JoinHandle<()> { + log::info!("loop voice {} starting: {}", layer + 1, path.display()); + let stream_handle = stream_handle.clone(); + thread::Builder::new() + .name(format!("agent-notify-loop-{}", layer + 1)) + .spawn(move || run_loop_iterator(stream_handle, path, gap_millis, stop)) + .expect("spawn loop iterator thread") +} + fn run_loop_iterator( stream_handle: OutputStreamHandle, path: PathBuf, @@ -163,15 +245,11 @@ fn run_loop_iterator( }; while !stop.load(Ordering::SeqCst) { - match enqueue_sound(&sink, &path) { - Ok(()) => {} - Err(e) => { - log::warn!("loop iterator: failed to enqueue {}: {}", path.display(), e); - return; - } + if let Err(e) = enqueue_sound(&sink, &path) { + log::warn!("loop iterator: failed to enqueue {}: {}", path.display(), e); + return; } - // Wait for playback to finish, but stay responsive to stop. while !sink.empty() { if stop.load(Ordering::SeqCst) { sink.stop(); @@ -184,16 +262,10 @@ fn run_loop_iterator( return; } - if gap_millis > 0 { - let total = Duration::from_millis(gap_millis); - let started = Instant::now(); - while started.elapsed() < total { - if stop.load(Ordering::SeqCst) { - return; - } - let remaining = total.saturating_sub(started.elapsed()); - thread::sleep(remaining.min(STOP_POLL_INTERVAL)); - } + if gap_millis > 0 + && !sleep_with_stop(Duration::from_millis(gap_millis), &stop) + { + return; } } } @@ -205,6 +277,20 @@ fn enqueue_sound(sink: &Sink, path: &Path) -> anyhow::Result<()> { Ok(()) } +/// Sleep for `total` while polling `stop` every [`STOP_POLL_INTERVAL`]. +/// Returns `true` if the full duration elapsed, `false` if `stop` was set. +fn sleep_with_stop(total: Duration, stop: &AtomicBool) -> bool { + let started = Instant::now(); + while started.elapsed() < total { + if stop.load(Ordering::SeqCst) { + return false; + } + let remaining = total.saturating_sub(started.elapsed()); + thread::sleep(remaining.min(STOP_POLL_INTERVAL)); + } + !stop.load(Ordering::SeqCst) +} + struct LoopController { stop: Arc, thread: Option>, diff --git a/crates/tools/servers/agent_notify_server/src/config.rs b/crates/tools/servers/agent_notify_server/src/config.rs index b67edd5874..a16d35dd6b 100644 --- a/crates/tools/servers/agent_notify_server/src/config.rs +++ b/crates/tools/servers/agent_notify_server/src/config.rs @@ -9,12 +9,31 @@ use serde_derive::{Deserialize, Serialize}; pub const DEFAULT_CONFIG_PATH: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/config/notify_config.yaml"); +pub const DEFAULT_ESCALATE_WAIT_1_SECS: u64 = 15; +pub const DEFAULT_ESCALATE_WAIT_2_SECS: u64 = 30; +pub const DEFAULT_ESCALATE_WAIT_3_SECS: u64 = 45; + #[derive(Clone, Debug, Default, Deserialize, Serialize)] pub struct NotifyConfig { pub alert_beep_sound: Option, pub alert_done_sound: Option, pub alert_await_user_input_sound: Option, + + #[serde(default)] + pub extra_alert_beep_sounds: Vec, + #[serde(default)] + pub extra_alert_done_sounds: Vec, + #[serde(default)] + pub extra_alert_await_sounds: Vec, + pub loop_alert_timeout_millis: Option, + + /// Seconds after a loop starts before a second concurrent voice is added. + pub escalate_wait_1: Option, + /// Seconds before a third concurrent voice is added. + pub escalate_wait_2: Option, + /// Seconds before a fourth concurrent voice is added. + pub escalate_wait_3: Option, } impl NotifyConfig { @@ -46,17 +65,36 @@ impl NotifyConfig { } } + pub fn escalate_waits_secs(&self) -> [u64; 3] { + [ + self.escalate_wait_1.unwrap_or(DEFAULT_ESCALATE_WAIT_1_SECS), + self.escalate_wait_2.unwrap_or(DEFAULT_ESCALATE_WAIT_2_SECS), + self.escalate_wait_3.unwrap_or(DEFAULT_ESCALATE_WAIT_3_SECS), + ] + } + fn resolve_relative_paths(&mut self, base: &Path) { - resolve(&mut self.alert_beep_sound, base); - resolve(&mut self.alert_done_sound, base); - resolve(&mut self.alert_await_user_input_sound, base); + resolve_opt(&mut self.alert_beep_sound, base); + resolve_opt(&mut self.alert_done_sound, base); + resolve_opt(&mut self.alert_await_user_input_sound, base); + resolve_vec(&mut self.extra_alert_beep_sounds, base); + resolve_vec(&mut self.extra_alert_done_sounds, base); + resolve_vec(&mut self.extra_alert_await_sounds, base); } } -fn resolve(slot: &mut Option, base: &Path) { +fn resolve_opt(slot: &mut Option, base: &Path) { if let Some(p) = slot.as_ref() { if p.is_relative() { *slot = Some(base.join(p)); } } } + +fn resolve_vec(slot: &mut Vec, base: &Path) { + for p in slot.iter_mut() { + if p.is_relative() { + *p = base.join(&*p); + } + } +} diff --git a/crates/tools/servers/agent_notify_server/src/endpoints/loop_handlers.rs b/crates/tools/servers/agent_notify_server/src/endpoints/loop_handlers.rs index 1f474ce7ae..15d13b6956 100644 --- a/crates/tools/servers/agent_notify_server/src/endpoints/loop_handlers.rs +++ b/crates/tools/servers/agent_notify_server/src/endpoints/loop_handlers.rs @@ -2,35 +2,55 @@ use std::path::PathBuf; use actix_web::{HttpResponse, Responder, web}; +use crate::audio_player::LoopSpec; use crate::server_state::ServerState; pub async fn loop_beep_handler(state: web::Data) -> impl Responder { - play_loop_or_404(&state, state.config.alert_beep_sound.clone(), "alert_beep_sound") + play_loop_or_404( + &state, + state.config.alert_beep_sound.clone(), + state.config.extra_alert_beep_sounds.clone(), + "alert_beep_sound", + ) } pub async fn loop_done_handler(state: web::Data) -> impl Responder { - play_loop_or_404(&state, state.config.alert_done_sound.clone(), "alert_done_sound") + play_loop_or_404( + &state, + state.config.alert_done_sound.clone(), + state.config.extra_alert_done_sounds.clone(), + "alert_done_sound", + ) } pub async fn loop_await_handler(state: web::Data) -> impl Responder { play_loop_or_404( &state, state.config.alert_await_user_input_sound.clone(), + state.config.extra_alert_await_sounds.clone(), "alert_await_user_input_sound", ) } fn play_loop_or_404( state: &ServerState, - maybe_path: Option, + primary: Option, + extras: Vec, config_key: &str, ) -> HttpResponse { - match maybe_path { - Some(path) => { - let gap_millis = state.config.loop_alert_timeout_millis.unwrap_or(0); - state.audio.play_loop(path, gap_millis); - HttpResponse::Ok().body("ok\n") - } - None => HttpResponse::NotFound().body(format!("{} not configured\n", config_key)), - } + let Some(primary) = primary else { + return HttpResponse::NotFound().body(format!("{} not configured\n", config_key)); + }; + + let mut pool = Vec::with_capacity(1 + extras.len()); + pool.push(primary); + pool.extend(extras); + + let spec = LoopSpec { + pool, + gap_millis: state.config.loop_alert_timeout_millis.unwrap_or(0), + escalate_waits_secs: state.config.escalate_waits_secs(), + }; + state.audio.play_loop(spec); + HttpResponse::Ok().body("ok\n") } diff --git a/crates/tools/servers/agent_notify_server/static/index.html b/crates/tools/servers/agent_notify_server/static/index.html index 2c374b2c77..fd19a73c61 100644 --- a/crates/tools/servers/agent_notify_server/static/index.html +++ b/crates/tools/servers/agent_notify_server/static/index.html @@ -41,8 +41,16 @@ GET /loop_awaitLoop alert_await_user_input. Replaces any prior loop.

- Between iterations the server waits loop_alert_timeout_millis - (from the YAML config) before playing again. + Between iterations a single voice waits loop_alert_timeout_millis + before replaying. +

+

+ Escalation: after the loop starts, a second voice joins + at escalate_wait_1 seconds, a third at + escalate_wait_2, and a fourth at escalate_wait_3. + Each new voice is drawn from extra_alert_<state>_sounds; + when those run out the pool cycles, doubling already-playing voices. + Voices are independent threads — they drift relative to one another.