fix(app): don't wipe early-arriving peers on RoomJoined

Field-test regression from the F-06 fix (a17b930): joining a room via the
friends-list / Recents Join button sometimes landed in an empty roster even
though the peer was fully connected at the gossip layer.

Cause: `PeerJoined` (gossip event task) and `RoomJoined` (core command loop)
ride the same UI channel from different senders. The core emits `RoomJoined`
only after audio + echo-cancel setup, so `PeerJoined` for the new room
routinely arrives first. F-06 had added `reset_room_state()` to the
`RoomJoined` handler, which then cleared the peer that had already announced.
Echo cancellation widened the window and made it reliable; the roster
"self-healed" only on the peer's next periodic re-announce (`PeerUpdated`).

Fix: reset room-scoped UI state at join *initiation* (JoinPressed,
CreatePressed, JoinFriendRoom, JoinRecent) instead of on `RoomJoined`. From
Home that's a no-op (already cleared on leave), so nothing leaks, and an
early `PeerJoined` for the new room now survives. The in-call switch path
F-06 targeted is unreachable from the current UI (friends list + Recents
render only on the Home screen), so this fully covers the reachable case.

Field-verified on a 2-machine desktop<->dopedart call. 407 lib tests pass,
clippy --all-targets clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/app/mod.rs

@@ -529,6 +529,27 @@ pub struct AppState {
 }
 
 impl AppState {
+    fn reset_room_state(&mut self) {
+        self.clip_player.stop();
+        self.peers.clear();
+        self.audio_levels.clear();
+        self.locally_muted.clear();
+        self.chat_messages.clear();
+        self.chat_input.clear();
+        self.attachment_data.clear();
+        self.image_handle_cache.clear();
+        self.pending_saves.clear();
+        self.pending_plays.clear();
+        self.invalid_audio.clear();
+        self.connecting.clear();
+        self.ever_connected.clear();
+        self.recording = false;
+        self.recording_started = None;
+        self.call_started = None;
+        self.mic_level = 0.0;
+        self.self_sharing = false;
+    }
+
     fn custom_sound_path(&self, sound: Sound) -> &str {
         let opt = match sound {
             Sound::SelfJoin => &self.config.custom_sound_self_join,
@@ -985,6 +1006,10 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
             let output_device = state.selected_output.as_ref().map(|d| d.name.clone());
             if !state.ticket_input.is_empty() {
                 state.status_message = "Joining room...".to_string();
+                // Clear any prior room's UI state now, at initiation, so an early
+                // `PeerJoined` for the new room (which can beat `RoomJoined`) isn't
+                // wiped. From Home this is a no-op (already cleared on leave).
+                state.reset_room_state();
                 // Remember this nickname for next launch.
                 state.config.username = state.name.clone();
                 state.config.save();
@@ -1005,6 +1030,8 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
             let input_device = state.selected_input.as_ref().map(|d| d.name.clone());
             let output_device = state.selected_output.as_ref().map(|d| d.name.clone());
             state.status_message = "Creating room...".to_string();
+            // Clear any prior room's UI state at initiation (see JoinPressed).
+            state.reset_room_state();
             // Remember this nickname for next launch.
             state.config.username = state.name.clone();
             state.config.save();
@@ -1044,6 +1071,12 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
         AppMessage::UiEventReceived(event) => {
             match event {
                 UiEvent::RoomJoined { ticket, self_id } => {
+                    // NB: do NOT clear peers here. `PeerJoined` rides a separate
+                    // channel sender (the gossip event task) and routinely arrives
+                    // BEFORE this `RoomJoined` (which the core emits only after audio
+                    // + echo-cancel setup), so clearing here would wipe a peer that
+                    // already announced → an empty roster. Room-scoped state is reset
+                    // at join *initiation* instead (see the Join* handlers).
                     // Remember this gathering for one-click rejoin (W7 P5). The
                     // emitted ticket is the canonical room door (topic + member
                     // addr + label); push_recent de-dupes by topic and persists.
@@ -1065,29 +1098,18 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
                     notify::play(Sound::SelfJoin, state.config.custom_sound_self_join.as_deref());
                 }
                 UiEvent::RoomLeft => {
-                    state.clip_player.stop();
+                    state.reset_room_state();
                     state.ticket = "".to_string();
-                    state.peers.clear();
-                    state.audio_levels.clear();
-                    state.locally_muted.clear();
-                    state.call_started = None;
-                    state.recording = false;
-                    state.recording_started = None;
-                    state.chat_messages.clear();
-                    state.chat_input.clear();
-                    state.attachment_data.clear();
-                    state.image_handle_cache.clear();
-                    state.pending_saves.clear();
-                    state.pending_plays.clear();
-                    state.invalid_audio.clear();
-                    state.connecting.clear();
-                    state.ever_connected.clear();
                     state.status_message = "Ready to connect".to_string();
                     state.current_screen = Screen::Home;
-                    state.mic_level = 0.0;
-                    state.self_sharing = false;
                     notify::play(Sound::SelfLeave, state.config.custom_sound_self_leave.as_deref());
                 }
+                UiEvent::RoomReset => {
+                    state.reset_room_state();
+                    state.ticket = "".to_string();
+                    state.status_message = "Ready to connect".to_string();
+                    state.current_screen = Screen::Home;
+                }
                 UiEvent::PeerJoined { id, state: peer_state } => {
                     state.peers.insert(id, peer_state);
                     notify::play(Sound::PeerJoin, state.config.custom_sound_peer_join.as_deref());
@@ -1412,6 +1434,8 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
             let input_device = state.selected_input.as_ref().map(|d| d.name.clone());
             let output_device = state.selected_output.as_ref().map(|d| d.name.clone());
             state.status_message = "Joining your friend's room...".to_string();
+            // Clear any prior room's UI state at initiation (see JoinPressed).
+            state.reset_room_state();
             state.config.username = state.name.clone();
             state.config.save();
             state.mic_test_active = false;
@@ -1431,6 +1455,8 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
             let input_device = state.selected_input.as_ref().map(|d| d.name.clone());
             let output_device = state.selected_output.as_ref().map(|d| d.name.clone());
             state.status_message = "Rejoining a recent room...".to_string();
+            // Clear any prior room's UI state at initiation (see JoinPressed).
+            state.reset_room_state();
             state.config.username = state.name.clone();
             state.config.save();
             state.mic_test_active = false;
@@ -5769,10 +5795,82 @@ impl Program<AppMessage> for Icon {
 mod tests {
     use super::{
         format_duration, initial_window_position, reconnect_attempt_chime, reconnected_chime,
-        set_peer_gate_config, set_peer_volume_config, AppConfig, GateMeter, METER_MAX,
+        set_peer_gate_config, set_peer_volume_config, AppConfig, AppState, AttachmentState,
+        ChatEntry, GateMeter, METER_MAX,
     };
     use iroh::SecretKey;
 
+    #[test]
+    fn reset_room_state_clears_all_room_scoped_state() {
+        let mut state = AppState::default();
+        let peer = SecretKey::generate().public();
+        let attachment_id = [9u8; 32];
+        let now = std::time::Instant::now();
+
+        state.peers.insert(peer, crate::network::PeerState {
+            name: "Peer".to_string(),
+            is_muted: false,
+            addr: iroh::EndpointAddr::from(peer),
+            sharing: None,
+            avatar: crate::avatar::Avatar::default(),
+            game: None,
+        });
+        state.audio_levels.insert(peer, 0.5);
+        state.locally_muted.insert(peer);
+        state.chat_messages.push(ChatEntry {
+            name: "Peer".to_string(),
+            text: "old room".to_string(),
+            mine: false,
+            from: Some(peer.to_string()),
+            attachment: None,
+        });
+        state.chat_input = "draft".to_string();
+        state.attachment_data.insert(attachment_id, AttachmentState::Ready(vec![1]));
+        state.image_handle_cache.insert(
+            attachment_id,
+            iced::widget::image::Handle::from_bytes(vec![1]),
+        );
+        state.pending_saves.insert(attachment_id);
+        state.pending_plays.insert(attachment_id);
+        state.invalid_audio.insert(attachment_id);
+        state.connecting.insert(peer);
+        state.ever_connected.insert(peer);
+        state.recording = true;
+        state.recording_started = Some(now);
+        state.call_started = Some(now);
+        state.mic_level = 0.75;
+        state.self_sharing = true;
+        state.clip_status.lock().unwrap().playing_id = Some(attachment_id);
+
+        state.reset_room_state();
+
+        assert!(state.peers.is_empty());
+        assert!(state.audio_levels.is_empty());
+        assert!(state.locally_muted.is_empty());
+        assert!(state.chat_messages.is_empty());
+        assert!(state.chat_input.is_empty());
+        assert!(state.attachment_data.is_empty());
+        assert!(state.image_handle_cache.is_empty());
+        assert!(state.pending_saves.is_empty());
+        assert!(state.pending_plays.is_empty());
+        assert!(state.invalid_audio.is_empty());
+        assert!(state.connecting.is_empty());
+        assert!(state.ever_connected.is_empty());
+        assert!(!state.recording);
+        assert!(state.recording_started.is_none());
+        assert!(state.call_started.is_none());
+        assert_eq!(state.mic_level, 0.0);
+        assert!(!state.self_sharing);
+
+        for _ in 0..50 {
+            if crate::audio::clip_player::status_snapshot(&state.clip_status).playing_id.is_none() {
+                return;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(2));
+        }
+        panic!("clip player did not stop during room reset");
+    }
+
     #[test]
     fn peer_gate_persists_when_on_and_clears_when_off() {
         let mut config = AppConfig::default();

src/audio/echo_cancel.rs

@@ -34,6 +34,18 @@ const NODE_READY_TIMEOUT: Duration = Duration::from_secs(3);
 /// nodes never leak past the call that created them.
 pub struct EchoCancelGuard {
     module_index: String,
+    source_name: String,
+    sink_name: String,
+}
+
+impl EchoCancelGuard {
+    pub fn source_name(&self) -> &str {
+        &self.source_name
+    }
+
+    pub fn sink_name(&self) -> &str {
+        &self.sink_name
+    }
 }
 
 impl Drop for EchoCancelGuard {
@@ -58,12 +70,16 @@ pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result<Echo
     // don't stack duplicate modules / fight over the virtual node names.
     unload_stale();
 
+    let owner_pid = std::process::id();
+    let source_name = format!("{EC_SOURCE}.{owner_pid}");
+    let sink_name = format!("{EC_SINK}.{owner_pid}");
+
     let mut cmd = Command::new("pactl");
     cmd.arg("load-module")
         .arg("module-echo-cancel")
         .arg("aec_method=webrtc")
-        .arg(format!("source_name={EC_SOURCE}"))
-        .arg(format!("sink_name={EC_SINK}"));
+        .arg(format!("source_name={source_name}"))
+        .arg(format!("sink_name={sink_name}"));
     if let Some(src) = real_source.filter(|s| !s.is_empty()) {
         cmd.arg(format!("source_master={src}"));
     }
@@ -85,12 +101,12 @@ pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result<Echo
     if module_index.parse::<u64>().is_err() {
         return Err(format!("unexpected pactl output: {module_index:?}"));
     }
-    let guard = EchoCancelGuard { module_index };
+    let guard = EchoCancelGuard { module_index, source_name, sink_name };
 
     // The virtual nodes appear shortly after the module loads; wait for both so
     // the subsequent capture/playback streams can actually target them. If they
     // never show, drop the guard (unloads) and report failure.
-    if !wait_for_nodes() {
+    if !wait_for_nodes(guard.source_name(), guard.sink_name()) {
         return Err("echo-cancel virtual nodes did not appear in time".to_string());
     }
 
@@ -102,10 +118,10 @@ pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result<Echo
 }
 
 /// Polls until both virtual nodes exist or the timeout elapses.
-fn wait_for_nodes() -> bool {
+fn wait_for_nodes(source_name: &str, sink_name: &str) -> bool {
     let deadline = Instant::now() + NODE_READY_TIMEOUT;
     loop {
-        if node_present("sources", EC_SOURCE) && node_present("sinks", EC_SINK) {
+        if node_present("sources", source_name) && node_present("sinks", sink_name) {
             return true;
         }
         if Instant::now() >= deadline {
@@ -126,8 +142,30 @@ fn node_present(kind: &str, name: &str) -> bool {
         .any(|line| line.split('\t').nth(1) == Some(name))
 }
 
-/// Unloads any leftover `module-echo-cancel` instance we previously created
-/// (identified by our virtual node names in its argument string). Best-effort.
+fn pid_from_ec_args(args: &str) -> Option<u32> {
+    let source_prefix = format!("source_name={EC_SOURCE}.");
+    args.split_whitespace()
+        .find_map(|arg| arg.strip_prefix(&source_prefix))?
+        .parse()
+        .ok()
+}
+
+fn ec_module_is_stale(args: &str, is_alive: impl Fn(u32) -> bool) -> bool {
+    pid_from_ec_args(args).is_some_and(|pid| !is_alive(pid))
+}
+
+#[cfg(target_os = "linux")]
+fn process_is_alive(pid: u32) -> bool {
+    std::path::Path::new("/proc").join(pid.to_string()).exists()
+}
+
+#[cfg(not(target_os = "linux"))]
+fn process_is_alive(_pid: u32) -> bool {
+    true
+}
+
+/// Unloads leftover PeerSpeak `module-echo-cancel` instances only when their
+/// owning process is gone. Best-effort and conservative on non-Linux platforms.
 fn unload_stale() {
     let Ok(out) = Command::new("pactl").arg("list").arg("modules").arg("short").output() else {
         return;
@@ -137,7 +175,10 @@ fn unload_stale() {
         let index = cols.next().unwrap_or("");
         let name = cols.next().unwrap_or("");
         let args = cols.next().unwrap_or("");
-        if name == "module-echo-cancel" && args.contains(EC_SOURCE) && index.parse::<u64>().is_ok() {
+        if name == "module-echo-cancel"
+            && ec_module_is_stale(args, process_is_alive)
+            && index.parse::<u64>().is_ok()
+        {
             let _ = Command::new("pactl").arg("unload-module").arg(index).output();
             crate::log_msg(&format!("Echo cancel: cleaned up stale module {index}"));
         }
@@ -155,12 +196,42 @@ mod tests {
     #[ignore]
     fn enable_creates_and_unloads_nodes() {
         let guard = enable(None, None).expect("module-echo-cancel should load");
-        assert!(node_present("sources", EC_SOURCE), "cleaned source must exist");
-        assert!(node_present("sinks", EC_SINK), "reference sink must exist");
+        let source_name = guard.source_name().to_string();
+        let sink_name = guard.sink_name().to_string();
+        assert!(node_present("sources", &source_name), "cleaned source must exist");
+        assert!(node_present("sinks", &sink_name), "reference sink must exist");
         drop(guard);
         // Give pactl a moment to tear the nodes down.
         std::thread::sleep(Duration::from_millis(300));
-        assert!(!node_present("sources", EC_SOURCE), "source must be gone after unload");
-        assert!(!node_present("sinks", EC_SINK), "sink must be gone after unload");
+        assert!(!node_present("sources", &source_name), "source must be gone after unload");
+        assert!(!node_present("sinks", &sink_name), "sink must be gone after unload");
+    }
+
+    #[test]
+    fn parses_owner_pid_only_from_our_source_name() {
+        assert_eq!(
+            pid_from_ec_args(
+                "aec_method=webrtc source_name=peerspeak_echocancel_source.4242 sink_name=peerspeak_echocancel_sink.4242"
+            ),
+            Some(4242)
+        );
+        assert_eq!(pid_from_ec_args("aec_method=webrtc"), None);
+        assert_eq!(
+            pid_from_ec_args("source_name=peerspeak_echocancel_source.not-a-pid"),
+            None
+        );
+        assert_eq!(pid_from_ec_args("source_name=someone_elses_source.4242"), None);
+    }
+
+    #[test]
+    fn stale_decision_keeps_live_and_foreign_modules() {
+        let ours = "source_name=peerspeak_echocancel_source.4242";
+        assert!(!ec_module_is_stale(ours, |pid| pid == 4242));
+        assert!(ec_module_is_stale(ours, |_| false));
+        assert!(!ec_module_is_stale("source_name=foreign.4242", |_| false));
+        assert!(!ec_module_is_stale(
+            "source_name=peerspeak_echocancel_source.malformed",
+            |_| false
+        ));
     }
 }

src/audio/multitrack.rs

@@ -29,6 +29,32 @@ const SILENCE_CHUNK: usize = FRAME_SAMPLES * 256;
 /// can drift if the capture clock runs ahead of the mixer cycle; past it the
 /// oldest mic audio is dropped. Mirrors `recorder::MAX_MIC_FIFO`.
 const MAX_MIC_FIFO: usize = 48_000 / 5;
+const MAX_SESSION_DIR_ATTEMPTS: usize = 1_000;
+
+/// Create a collision-free session directory for a timestamp. The base
+/// timestamp is tried first, followed by `-2`, `-3`, and so on; an existing
+/// recording is never reopened or overwritten.
+pub fn create_session_dir(base: &Path, now_unix_secs: u64) -> io::Result<PathBuf> {
+    let filename = crate::audio::recorder::timestamp_filename(now_unix_secs);
+    let stem = filename.trim_end_matches(".wav");
+    for attempt in 1..=MAX_SESSION_DIR_ATTEMPTS {
+        let name = if attempt == 1 {
+            stem.to_string()
+        } else {
+            format!("{stem}-{attempt}")
+        };
+        let path = base.join(name);
+        match std::fs::create_dir(&path) {
+            Ok(()) => return Ok(path),
+            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
+            Err(e) => return Err(e),
+        }
+    }
+    Err(io::Error::new(
+        io::ErrorKind::AlreadyExists,
+        "multitrack directory suffixes exhausted",
+    ))
+}
 
 /// One output track: its WAV writer plus whether it has been written *this*
 /// cycle (so `end_cycle` knows which tracks to pad with silence).
@@ -263,6 +289,19 @@ mod tests {
         assert_eq!(track_filename("!!!", &id), format!("peer-{short}.wav"));
     }
 
+    #[test]
+    fn same_second_sessions_get_unique_directories_without_reuse() {
+        let base = tmpdir("collision");
+        let first = create_session_dir(&base, 1_700_000_000).unwrap();
+        std::fs::write(first.join("sentinel"), b"keep me").unwrap();
+
+        let second = create_session_dir(&base, 1_700_000_000).unwrap();
+
+        assert_ne!(second, first);
+        assert_eq!(std::fs::read(first.join("sentinel")).unwrap(), b"keep me");
+        let _ = std::fs::remove_dir_all(&base);
+    }
+
     #[test]
     fn all_tracks_equal_length_after_n_cycles() {
         let dir = tmpdir("equal");

src/audio/pipewire_impl.rs

@@ -151,11 +151,9 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender<Vec<i16>>, target_n
                     let data = &mut datas[0];
                     let size = data.chunk().size() as usize;
                     if let Some(slice) = data.data() {
-                        // Each sample is 2 bytes (S16LE)
-                        for chunk in slice[..size].chunks_exact(2) {
-                            let sample = i16::from_le_bytes([chunk[0], chunk[1]]);
+                        for_each_capture_sample(slice, size, |sample| {
                             let _ = user_data.producer.try_push(sample);
-                        }
+                        });
                     }
                 }
             }
@@ -224,6 +222,16 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender<Vec<i16>>, target_n
     Ok(())
 }
 
+/// Visit the complete S16LE samples in the portion PipeWire reports as filled.
+/// Clamp the reported byte count to the mapped slice before indexing: a bad
+/// chunk size must not panic from the realtime capture callback.
+fn for_each_capture_sample(slice: &[u8], size: usize, mut visit: impl FnMut(i16)) {
+    let size = size.min(slice.len());
+    for chunk in slice[..size].chunks_exact(2) {
+        visit(i16::from_le_bytes([chunk[0], chunk[1]]));
+    }
+}
+
 /// Frames the playback RT callback should produce this cycle.
 ///
 /// `requested` is the graph's per-cycle quantum from `Buffer::requested()` (0 if
@@ -263,6 +271,25 @@ fn drain_loop(
     }
 }
 
+/// Reserve exact occupancy before making a frame visible to the consumer.
+/// `after_reserve` is empty in production and lets the regression test force a
+/// consumer interleaving at the critical ordering boundary.
+fn publish_frame<P: Producer<Item = i16>>(
+    fill: &AtomicUsize,
+    dropped: &AtomicU64,
+    producer: &mut P,
+    frame: &[i16],
+    after_reserve: impl FnOnce(),
+) {
+    fill.fetch_add(frame.len(), Ordering::Relaxed);
+    after_reserve();
+    let pushed = producer.push_slice(frame);
+    if pushed != frame.len() {
+        fill.fetch_sub(frame.len() - pushed, Ordering::Relaxed);
+        dropped.fetch_add(1, Ordering::Relaxed);
+    }
+}
+
 fn frames_to_produce(requested: usize, mapped_frames: usize) -> usize {
     /// Safe per-cycle fallback when the graph doesn't report a quantum.
     const FALLBACK_FRAMES: usize = 1024;
@@ -522,10 +549,12 @@ fn run_playback(
                 worker_dropped.fetch_add(1, Ordering::Relaxed);
                 return;
             }
-            for &sample in &frame {
-                let _ = producer.try_push(sample);
-            }
-            worker_fill.fetch_add(frame.len(), Ordering::Relaxed);
+            // Reserve occupancy BEFORE publishing samples. Otherwise the RT
+            // consumer can pop a newly-visible sample before it is counted and
+            // wrap the exact fill gauge to usize::MAX, wedging mixer pacing.
+            // `push_slice` also publishes the frame as one operation rather than
+            // exposing a half-written stereo pair.
+            publish_frame(&worker_fill, &worker_dropped, &mut producer, &frame, || {});
         });
     });
 
@@ -577,8 +606,9 @@ fn run_playback(
 
 #[cfg(test)]
 mod tests {
-    use super::{drain_loop, frames_to_produce};
-    use std::sync::atomic::{AtomicBool, Ordering};
+    use super::{drain_loop, for_each_capture_sample, frames_to_produce, publish_frame};
+    use ringbuf::{HeapRb, traits::{Consumer, Producer, Split}};
+    use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
     use std::sync::{Arc, Mutex};
     use std::time::Duration;
     use std::{sync::mpsc, thread};
@@ -614,6 +644,39 @@ mod tests {
         assert_eq!(frames_to_produce(1024, 0), 0);
     }
 
+    #[test]
+    fn capture_size_larger_than_mapping_is_clamped() {
+        let mut samples = Vec::new();
+        for_each_capture_sample(&[1, 0, 2, 0, 3], usize::MAX, |sample| {
+            samples.push(sample)
+        });
+        assert_eq!(samples, vec![1, 2]);
+    }
+
+    #[test]
+    fn occupancy_is_reserved_before_frame_is_published() {
+        let rb = HeapRb::<i16>::new(8);
+        let (mut producer, mut consumer) = rb.split();
+        assert!(producer.try_push(7).is_ok());
+
+        let fill = AtomicUsize::new(1);
+        let dropped = AtomicU64::new(0);
+        publish_frame(&fill, &dropped, &mut producer, &[10, 11], || {
+            // Force the consumer to drain the old sample after the new frame's
+            // occupancy is reserved but before that frame is published.
+            assert_eq!(consumer.try_pop(), Some(7));
+            assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 3);
+        });
+
+        assert_eq!(fill.load(Ordering::Relaxed), 2);
+        assert_eq!(consumer.try_pop(), Some(10));
+        assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 2);
+        assert_eq!(consumer.try_pop(), Some(11));
+        assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 1);
+        assert_eq!(fill.load(Ordering::Relaxed), 0);
+        assert_eq!(dropped.load(Ordering::Relaxed), 0);
+    }
+
     // --- drain_loop (A7: worker must not hang shutdown) ---
 
     #[test]

src/audio/recorder.rs

@@ -14,7 +14,7 @@
 //! and patches the two size fields on [`Recorder::finalize`].
 
 use std::collections::VecDeque;
-use std::fs::File;
+use std::fs::{File, OpenOptions};
 use std::io::{self, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 
@@ -24,6 +24,7 @@ const BITS_PER_SAMPLE: u16 = 16;
 const CHANNELS: u16 = 1;
 const RIFF_DATA_OVERHEAD: u64 = 36;
 const MAX_RIFF_DATA_BYTES: u64 = u32::MAX as u64 - RIFF_DATA_OVERHEAD;
+const MAX_NAME_ATTEMPTS: usize = 1_000;
 
 /// Cap on buffered mic samples (~200ms). Bounds how far recording lag can drift
 /// if the capture clock runs persistently faster than playout — past this we drop
@@ -42,7 +43,12 @@ pub struct WavWriter {
 impl WavWriter {
     /// Create the file and write the 44-byte header with zeroed size fields.
     pub fn new(path: &Path) -> io::Result<Self> {
-        let mut file = File::create(path)?;
+        Self::from_file(File::create(path)?)
+    }
+
+    /// Start a WAV in an already-opened file. This lets callers choose atomic
+    /// create-new semantics instead of the truncating behavior of `File::create`.
+    fn from_file(mut file: File) -> io::Result<Self> {
         file.write_all(&Self::header(0))?;
         Ok(Self {
             file,
@@ -125,13 +131,31 @@ impl Recorder {
     /// Create a recording at `dir/<timestamped>.wav`. The directory is assumed to
     /// exist (the caller creates it).
     pub fn create(dir: &Path, now_unix_secs: u64) -> io::Result<Self> {
-        let path = dir.join(timestamp_filename(now_unix_secs));
-        let writer = WavWriter::new(&path)?;
-        Ok(Self {
-            writer,
-            mic_fifo: VecDeque::new(),
-            path,
-        })
+        let filename = timestamp_filename(now_unix_secs);
+        let stem = filename.trim_end_matches(".wav");
+        for attempt in 1..=MAX_NAME_ATTEMPTS {
+            let name = if attempt == 1 {
+                filename.clone()
+            } else {
+                format!("{stem}-{attempt}.wav")
+            };
+            let path = dir.join(name);
+            match OpenOptions::new().write(true).create_new(true).open(&path) {
+                Ok(file) => {
+                    return Ok(Self {
+                        writer: WavWriter::from_file(file)?,
+                        mic_fifo: VecDeque::new(),
+                        path,
+                    });
+                }
+                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
+                Err(e) => return Err(e),
+            }
+        }
+        Err(io::Error::new(
+            io::ErrorKind::AlreadyExists,
+            "recording filename suffixes exhausted",
+        ))
     }
 
     /// The path being written.
@@ -210,6 +234,30 @@ mod tests {
         assert_eq!(timestamp_filename(0), "peerspeak-1970-01-01_000000.wav");
     }
 
+    #[test]
+    fn same_second_recordings_get_unique_files_without_truncation() {
+        let dir = std::env::temp_dir().join(format!(
+            "peerspeak-collision-{}",
+            std::process::id()
+        ));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+
+        let mut first = Recorder::create(&dir, 1_700_000_000).unwrap();
+        first.write_frame(&[123, 456]).unwrap();
+        let first_path = first.path().to_path_buf();
+        first.finalize().unwrap();
+        let original = std::fs::read(&first_path).unwrap();
+
+        let second = Recorder::create(&dir, 1_700_000_000).unwrap();
+        let second_path = second.path().to_path_buf();
+        assert_ne!(second_path, first_path);
+        assert_eq!(std::fs::read(&first_path).unwrap(), original);
+        second.finalize().unwrap();
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     fn wav_header_round_trips_sizes() {
         let dir = std::env::temp_dir();

src/core/messages.rs

@@ -106,6 +106,9 @@ pub enum CoreCommand {
 pub enum UiEvent {
     RoomJoined { ticket: String, self_id: String },
     RoomLeft,
+    /// Clear room-scoped UI state after a failed in-call room switch, without a
+    /// leave chime. The persistent identity remains unchanged.
+    RoomReset,
     PeerJoined { id: EndpointId, state: PeerState },
     PeerLeft { id: EndpointId },
     /// The fixed reconnect grace expired and bounded background gossip recovery

src/core/mod.rs

@@ -91,6 +91,22 @@ fn game_presence_label(game: Option<&crate::game::DetectedGame>) -> Option<Strin
         .filter(|s| !s.is_empty())
 }
 
+/// Wait for the next game update. A closed sender permanently disables the
+/// source by clearing the receiver; subsequent calls remain pending instead of
+/// leaving an always-ready closed branch in the core `select!` loop.
+async fn next_game_change(
+    game_rx: &mut Option<tokio::sync::watch::Receiver<Option<crate::game::DetectedGame>>>,
+) -> Option<Option<crate::game::DetectedGame>> {
+    let Some(rx) = game_rx.as_mut() else {
+        return std::future::pending().await;
+    };
+    if rx.changed().await.is_err() {
+        *game_rx = None;
+        return None;
+    }
+    Some(rx.borrow_and_update().clone())
+}
+
 fn arm_discovery_retry(
     discovery_deadline: &mut Option<tokio::time::Instant>,
     now: tokio::time::Instant,
@@ -116,14 +132,14 @@ type GraceTimers = Arc<std::sync::Mutex<HashMap<EndpointId, tokio::task::JoinHan
 type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;
 
 type KnownPeers =
-    Arc<std::sync::Mutex<HashMap<String, HashMap<EndpointId, EndpointAddr>>>>;
+    Arc<std::sync::Mutex<HashMap<[u8; 32], HashMap<EndpointId, EndpointAddr>>>>;
 
 #[derive(Clone)]
 struct RecoveryContext {
     coordinator: RecoveryCoordinator,
     room_state: Arc<IrohGossipState>,
     known_peers: KnownPeers,
-    ticket: String,
+    topic_id: [u8; 32],
 }
 
 impl RecoveryContext {
@@ -131,7 +147,7 @@ impl RecoveryContext {
         self.known_peers
             .lock()
             .unwrap()
-            .get(&self.ticket)
+            .get(&self.topic_id)
             .and_then(|peers| peers.get(peer_id))
             .cloned()
     }
@@ -141,7 +157,7 @@ impl RecoveryContext {
     }
 
     fn forget(&self, peer_id: EndpointId) {
-        if let Some(peers) = self.known_peers.lock().unwrap().get_mut(&self.ticket) {
+        if let Some(peers) = self.known_peers.lock().unwrap().get_mut(&self.topic_id) {
             peers.remove(&peer_id);
         }
         self.coordinator.cancel(peer_id);
@@ -943,11 +959,19 @@ async fn run_core_loop(
     // the game is *broadcast* as presence is gated by `game_presence_enabled` (opt-in,
     // seeded false; the GUI sends `SetGamePresenceEnabled` from config at startup).
     // The override + process map start at their defaults and are set via commands.
-    let game_detector = crate::game::detector::GameDetector::spawn(
+    let (game_detector, mut game_rx) = match crate::game::detector::GameDetector::spawn(
         crate::game::ManualOverride::Auto,
         std::collections::BTreeMap::new(),
-    );
-    let mut game_rx = game_detector.subscribe();
+    ) {
+        Ok(detector) => {
+            let rx = detector.subscribe();
+            (Some(detector), Some(rx))
+        }
+        Err(e) => {
+            crate::log_msg(&format!("game detector unavailable: {e}"));
+            (None, None)
+        }
+    };
     let mut game_presence_enabled = false;
     // The latest debounced detection, kept regardless of the broadcast toggle so a
     // later opt-in can immediately publish whatever is currently running.
@@ -1058,13 +1082,14 @@ async fn run_core_loop(
                 Some(cmd) => cmd,
                 None => break,
             },
-            changed = game_rx.changed() => {
+            game_change = next_game_change(&mut game_rx) => {
                 // The detector worker published a new debounced game (or `None`).
-                if changed.is_err() {
-                    // Worker gone (shouldn't happen before shutdown); stop watching.
+                let Some(detected) = game_change else {
+                    // Worker gone unexpectedly. The helper fused this source, so
+                    // this logs once and the closed channel cannot spin select!.
+                    crate::log_msg("game detector stopped; disabling game detection");
                     continue;
-                }
-                let detected = game_rx.borrow_and_update().clone();
+                };
                 current_game = detected.clone();
                 // Always tell the GUI for the local per-game background + indicator.
                 let _ = ui_tx.send(UiEvent::GameChanged(detected.clone())).await;
@@ -1182,6 +1207,7 @@ async fn run_core_loop(
             CoreCommand::Join { name, ticket, room_name, input_device, output_device, echo_cancellation, avatar } => {
                 presence.name = name.clone();
                 presence.avatar = avatar;
+                let was_in_room = active_session.is_some();
 
                 // Finalize any recording before tearing down the old session — its
                 // capture/mixer feeders are about to stop.
@@ -1194,6 +1220,7 @@ async fn run_core_loop(
                     session.shutdown(audio_backend.clone()).await;
                     net.audio_router.clear();
                     net.file_router.clear();
+                    *current_room.lock().unwrap() = None;
                 }
 
                 // If a network-mode / identity change was deferred while a call was
@@ -1243,6 +1270,17 @@ async fn run_core_loop(
                     ));
                     ticket_str
                 };
+                let topic_id = match PeerSpeakTicket::topic_of(&ticket_str) {
+                    Some(topic_id) => topic_id,
+                    None => {
+                        crate::log_msg("Error invalid room ticket");
+                        if was_in_room {
+                            let _ = ui_tx.send(UiEvent::RoomReset).await;
+                        }
+                        let _ = ui_tx.send(UiEvent::Error("invalid room ticket".to_string())).await;
+                        continue;
+                    }
+                };
 
                 // Per-session transport over the persistent endpoint, bound to the
                 // persistent audio router so this call's inbound audio links route
@@ -1267,14 +1305,14 @@ async fn run_core_loop(
                     None,
                 );
 
-                // Snapshot THIS room's retained peers (by ticket) as extra bootstrap
+                // Snapshot THIS room's retained peers (by topic) as extra bootstrap
                 // targets so a rejoin can dial them (A8) — including after a detour
-                // through another room, since the per-ticket archive isn't cleared.
+                // through another room, since the per-topic archive isn't cleared.
                 // Resolution rides the persistent address book.
                 let extra_bootstrap: Vec<EndpointAddr> = known_peers
                     .lock()
                     .unwrap()
-                    .get(&ticket_str)
+                    .get(&topic_id)
                     .map(|peers| peers.values().cloned().collect())
                     .unwrap_or_default();
 
@@ -1291,6 +1329,9 @@ async fn run_core_loop(
                 ));
                 if let Err(e) = room_state.join(&ticket_str, self_state.clone(), extra_bootstrap).await {
                     crate::log_msg(&format!("Error room_state.join failed: {:?}", e));
+                    if was_in_room {
+                        let _ = ui_tx.send(UiEvent::RoomReset).await;
+                    }
                     let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await;
                     net.audio_router.clear();
                     net.file_router.clear();
@@ -1317,12 +1358,11 @@ async fn run_core_loop(
                         output_device.as_deref(),
                     ) {
                         Ok(guard) => {
+                            let source_name = guard.source_name().to_string();
+                            let sink_name = guard.sink_name().to_string();
                             echo_cancel_guard = Some(guard);
                             crate::log_msg("Echo cancellation enabled");
-                            (
-                                Some(crate::audio::echo_cancel::EC_SOURCE.to_string()),
-                                Some(crate::audio::echo_cancel::EC_SINK.to_string()),
-                            )
+                            (Some(source_name), Some(sink_name))
                         }
                         Err(e) => {
                             crate::log_msg(&format!(
@@ -1343,6 +1383,9 @@ async fn run_core_loop(
                 let (capture_target, playback_target) = (input_device.clone(), output_device.clone());
 
                 if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) {
+                    if was_in_room {
+                        let _ = ui_tx.send(UiEvent::RoomReset).await;
+                    }
                     let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
                     let _ = room_state.leave().await;
                     net.audio_router.clear();
@@ -1355,6 +1398,9 @@ async fn run_core_loop(
                 // production to the hardware clock instead of a fixed timer.
                 let ring_fill = Arc::new(AtomicUsize::new(0));
                 if let Err(e) = audio_backend.start_playback(playback_rx, playback_target, ring_fill.clone()) {
+                    if was_in_room {
+                        let _ = ui_tx.send(UiEvent::RoomReset).await;
+                    }
                     let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await;
                     let _ = audio_backend.stop();
                     let _ = room_state.leave().await;
@@ -1656,23 +1702,57 @@ async fn run_core_loop(
                         // track in Both mode) one aligned frame per cycle; Mixed mode
                         // writes the single blended file as before.
                         if mt_active {
-                            if let Some(mt) = multitrack_mixer.lock().unwrap().as_mut() {
-                                let res = (|| -> std::io::Result<()> {
+                            let write_err = multitrack_mixer
+                                .lock()
+                                .unwrap()
+                                .as_mut()
+                                .map(|mt| -> std::io::Result<()> {
                                     for (id, f) in &stems {
                                         mt.write_peer(*id, f)?;
                                     }
                                     mt.write_mix(&record_mix)?;
                                     mt.end_cycle()
-                                })();
-                                if let Err(e) = res {
-                                    crate::log_msg(&format!("Multitrack write failed: {e}"));
-                                }
+                                })
+                                .transpose()
+                                .err();
+                            if let Some(e) = write_err {
+                                crate::log_msg(&format!("Multitrack write failed: {e}"));
+                                stop_recording(
+                                    &recorder_mixer,
+                                    &is_recording_mixer,
+                                    &multitrack_mixer,
+                                    &is_multitrack_mixer,
+                                    &ui_tx_mixer,
+                                ).await;
+                                let _ = ui_tx_mixer
+                                    .send(UiEvent::Error(format!(
+                                        "Recording stopped — write failed: {e}"
+                                    )))
+                                    .await;
+                            }
+                        } else if is_recording_mixer.load(Ordering::Relaxed) {
+                            let write_err = recorder_mixer
+                                .lock()
+                                .unwrap()
+                                .as_mut()
+                                .map(|rec| rec.write_frame(&record_mix))
+                                .transpose()
+                                .err();
+                            if let Some(e) = write_err {
+                                crate::log_msg(&format!("Recording write failed: {e}"));
+                                stop_recording(
+                                    &recorder_mixer,
+                                    &is_recording_mixer,
+                                    &multitrack_mixer,
+                                    &is_multitrack_mixer,
+                                    &ui_tx_mixer,
+                                ).await;
+                                let _ = ui_tx_mixer
+                                    .send(UiEvent::Error(format!(
+                                        "Recording stopped — write failed: {e}"
+                                    )))
+                                    .await;
                             }
-                        } else if is_recording_mixer.load(Ordering::Relaxed)
-                            && let Some(rec) = recorder_mixer.lock().unwrap().as_mut()
-                            && let Err(e) = rec.write_frame(&record_mix)
-                        {
-                            crate::log_msg(&format!("Recording write failed: {e}"));
                         }
 
                         let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) {
@@ -1699,6 +1779,9 @@ async fn run_core_loop(
                 let mut room_events = match room_state.subscribe_events().await {
                     Ok(rx) => rx,
                     Err(e) => {
+                        if was_in_room {
+                            let _ = ui_tx.send(UiEvent::RoomReset).await;
+                        }
                         let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe events: {}", e))).await;
                         continue;
                     }
@@ -1713,16 +1796,16 @@ async fn run_core_loop(
                 let multitrack_events = multitrack.clone();
                 let is_multitrack_events = is_multitrack.clone();
                 let known_peers_events = known_peers.clone();
-                // The ticket of the room this event loop serves, so peer add/remove
-                // updates the right per-ticket bucket in `known_peers` (A8 archive).
-                let ticket_events = ticket_str.clone();
+                // The topic of the room this event loop serves, so peer add/remove
+                // updates the right per-topic bucket in `known_peers` (A8 archive).
+                let room_topic = topic_id;
                 let (recovery_coordinator, recovery_task) =
                     RecoveryCoordinator::spawn(room_state.clone());
                 let recovery_context = RecoveryContext {
                     coordinator: recovery_coordinator,
                     room_state: room_state.clone(),
                     known_peers: known_peers.clone(),
-                    ticket: ticket_str.clone(),
+                    topic_id,
                 };
                 let recovery_events = recovery_context.clone();
                 // Friends store + ui sender, so a connected peer who is a friend has
@@ -1759,12 +1842,12 @@ async fn run_core_loop(
                                     )
                                     .await;
                                 }
-                                // Retain this peer under this room's ticket as a
+                                // Retain this peer under this room's topic as a
                                 // future rejoin bootstrap target (A8).
                                 known_peers_events
                                     .lock()
                                     .unwrap()
-                                    .entry(ticket_events.clone())
+                                    .entry(room_topic)
                                     .or_default()
                                     .insert(peer_id, state.addr.clone());
                                 // If a multitrack recording is live, give this peer
@@ -1820,7 +1903,7 @@ async fn run_core_loop(
                                 known_peers_events
                                     .lock()
                                     .unwrap()
-                                    .entry(ticket_events.clone())
+                                    .entry(room_topic)
                                     .or_default()
                                     .insert(peer_id, state.addr.clone());
                                 let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await;
@@ -1880,6 +1963,9 @@ async fn run_core_loop(
                 let mut conn_events = match transport.subscribe_conn_events().await {
                     Ok(rx) => rx,
                     Err(e) => {
+                        if was_in_room {
+                            let _ = ui_tx.send(UiEvent::RoomReset).await;
+                        }
                         let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe conn events: {}", e))).await;
                         continue;
                     }
@@ -2255,11 +2341,15 @@ async fn run_core_loop(
 
             CoreCommand::SetGameOverride(override_) => {
                 // Applied on the detector's next poll, immediately (bypasses debounce).
-                game_detector.set_override(override_);
+                if let Some(detector) = &game_detector {
+                    detector.set_override(override_);
+                }
             }
 
             CoreCommand::SetGameProcessMap(map) => {
-                game_detector.set_process_map(map);
+                if let Some(detector) = &game_detector {
+                    detector.set_process_map(map);
+                }
             }
 
             CoreCommand::SetRecordingMode(mode) => {
@@ -2283,11 +2373,13 @@ async fn run_core_loop(
                                     .unwrap_or(0);
                                 let result: Result<String, String> = if recording_mode.is_multitrack() {
                                     // Multitrack/Both: a per-session directory of stems.
-                                    let stamp = crate::audio::recorder::timestamp_filename(now);
-                                    let session_dir = base.join(stamp.trim_end_matches(".wav"));
-                                    std::fs::create_dir_all(&session_dir)
+                                    std::fs::create_dir_all(&base)
                                         .map_err(|e| e.to_string())
                                         .and_then(|_| {
+                                            crate::audio::multitrack::create_session_dir(&base, now)
+                                                .map_err(|e| e.to_string())
+                                        })
+                                        .and_then(|session_dir| {
                                             MultitrackRecorder::create(
                                                 &session_dir,
                                                 FRAME_SAMPLES,
@@ -2484,14 +2576,66 @@ async fn run_core_loop(
 mod tests {
     use super::{
         apply_volume, audio_datagram_len_ok, frame_level, mix_frames, mix_stereo_frames,
-        stereo_to_mono, MicLevelMeter, MAX_OPUS_PAYLOAD, MIC_LEVEL_REPORT_SAMPLES,
+        next_game_change, stereo_to_mono, KnownPeers, MicLevelMeter, PeerSpeakTicket,
+        MAX_OPUS_PAYLOAD, MIC_LEVEL_REPORT_SAMPLES,
     };
 
+    #[test]
+    fn retained_peers_are_shared_by_topic_across_restamped_tickets() {
+        let topic_id = [23u8; 32];
+        let original_host = iroh::SecretKey::generate().public();
+        let member_host = iroh::SecretKey::generate().public();
+        let retained_peer = iroh::SecretKey::generate().public();
+        let original = PeerSpeakTicket {
+            host_addr: iroh::EndpointAddr::from(original_host),
+            topic_id,
+            name: "Room".to_string(),
+        }.to_string();
+        let restamped = PeerSpeakTicket {
+            host_addr: iroh::EndpointAddr::from(member_host),
+            topic_id,
+            name: "Room".to_string(),
+        }.to_string();
+        assert_ne!(original, restamped);
+
+        let original_topic = PeerSpeakTicket::topic_of(&original).unwrap();
+        let restamped_topic = PeerSpeakTicket::topic_of(&restamped).unwrap();
+        assert_eq!(original_topic, restamped_topic);
+
+        let retained_addr = iroh::EndpointAddr::from(retained_peer);
+        let known_peers: KnownPeers =
+            std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
+        known_peers
+            .lock()
+            .unwrap()
+            .entry(original_topic)
+            .or_default()
+            .insert(retained_peer, retained_addr.clone());
+
+        let found = known_peers
+            .lock()
+            .unwrap()
+            .get(&restamped_topic)
+            .and_then(|peers| peers.get(&retained_peer))
+            .cloned();
+        assert_eq!(found, Some(retained_addr));
+    }
+
     /// A frame of constant amplitude with the given sample count.
     fn frame(amp: i16, len: usize) -> Vec<i16> {
         vec![amp; len]
     }
 
+    #[tokio::test]
+    async fn closed_game_watch_is_fused_after_one_ready_event() {
+        let (tx, rx) = tokio::sync::watch::channel(None);
+        let mut rx = Some(rx);
+        drop(tx);
+
+        assert_eq!(next_game_change(&mut rx).await, None);
+        assert!(rx.is_none(), "closed receiver must disable its select source");
+    }
+
     #[test]
     fn mic_meter_reports_only_after_enough_samples() {
         let mut m = MicLevelMeter::new();

src/game/detector.rs

@@ -15,8 +15,10 @@ use super::{
 use super::scan;
 use super::steam::SteamProbe;
 use std::collections::BTreeMap;
+use std::io;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
+use std::thread::JoinHandle;
 use std::time::Duration;
 use tokio::sync::watch;
 
@@ -62,13 +64,17 @@ pub struct GameDetector {
     inputs: Arc<DetectorInputs>,
     rx: watch::Receiver<Option<DetectedGame>>,
     stop: Arc<AtomicBool>,
+    worker: Option<JoinHandle<()>>,
 }
 
 impl GameDetector {
     /// Spawn the detector worker. `process_map` seeds the non-Steam mappings;
     /// `override_` seeds the manual override (usually `Auto`). The worker runs
     /// until [`stop`](Self::stop) or the returned `GameDetector` is dropped.
-    pub fn spawn(override_: ManualOverride, process_map: BTreeMap<String, String>) -> Self {
+    pub fn spawn(
+        override_: ManualOverride,
+        process_map: BTreeMap<String, String>,
+    ) -> io::Result<Self> {
         let inputs = Arc::new(DetectorInputs {
             override_: Mutex::new(override_),
             process_map: Mutex::new(process_map),
@@ -78,12 +84,16 @@ impl GameDetector {
 
         let worker_inputs = inputs.clone();
         let worker_stop = stop.clone();
-        std::thread::Builder::new()
+        let worker = std::thread::Builder::new()
             .name("game-detector".to_string())
-            .spawn(move || worker_loop(worker_inputs, tx, worker_stop))
-            .ok();
+            .spawn(move || worker_loop(worker_inputs, tx, worker_stop))?;
 
-        Self { inputs, rx, stop }
+        Ok(Self {
+            inputs,
+            rx,
+            stop,
+            worker: Some(worker),
+        })
     }
 
     /// A clone of the watch receiver for detected-game changes. The current value
@@ -112,6 +122,9 @@ impl GameDetector {
 impl Drop for GameDetector {
     fn drop(&mut self) {
         self.stop();
+        if let Some(worker) = self.worker.take() {
+            let _ = worker.join();
+        }
     }
 }
 
@@ -223,7 +236,7 @@ mod tests {
     fn spawn_and_stop_is_clean() {
         // Smoke test the lifecycle: spawning and stopping must not panic, and the
         // initial published value is None.
-        let det = GameDetector::spawn(ManualOverride::Auto, BTreeMap::new());
+        let det = GameDetector::spawn(ManualOverride::Auto, BTreeMap::new()).unwrap();
         assert_eq!(*det.subscribe().borrow(), None);
         det.set_override(ManualOverride::ForceNone);
         det.set_process_map(map(&[("x", "X")]));

src/game/steam.rs

@@ -18,6 +18,29 @@ use std::time::SystemTime;
 /// (a manifest is a few KB); the cap stops a corrupt/hostile giant file from being
 /// slurped into memory before the parser's own depth guard kicks in.
 const MAX_STEAM_FILE_BYTES: u64 = 4 * 1024 * 1024;
+/// SteamPath is a local filesystem path. Four KiB is deliberately generous and
+/// prevents a corrupt registry length from driving an enormous allocation.
+#[cfg(any(windows, test))]
+const MAX_STEAM_PATH_BYTES: u32 = 4 * 1024;
+
+#[cfg(any(windows, test))]
+fn validate_reg_len(len: u32) -> Option<usize> {
+    (len != 0 && len.is_multiple_of(2) && len <= MAX_STEAM_PATH_BYTES)
+        .then_some(len as usize / 2)
+}
+
+#[cfg(any(windows, test))]
+fn decode_reg_sz(mut buf: Vec<u16>, returned_bytes: u32) -> Option<String> {
+    let units = validate_reg_len(returned_bytes)?;
+    if units > buf.len() {
+        return None;
+    }
+    buf.truncate(units);
+    while buf.last() == Some(&0) {
+        buf.pop();
+    }
+    Some(String::from_utf16_lossy(&buf))
+}
 
 /// Parse the live `RunningAppID` out of a Steam `registry.vdf` (the Linux/macOS
 /// client's emulated-registry text file). Returns the appid only when present and
@@ -333,6 +356,7 @@ mod win {
     //! Windows registry reads via direct Win32 FFI (windows-sys), no `winreg`
     //! crate. Steam stores both the live `RunningAppID` and its install path under
     //! `HKCU\Software\Valve\Steam`.
+    use super::{decode_reg_sz, validate_reg_len};
     use std::path::PathBuf;
     use windows_sys::Win32::Foundation::ERROR_SUCCESS;
     use windows_sys::Win32::System::Registry::{
@@ -401,12 +425,17 @@ mod win {
                 &mut len,
             )
         };
-        if rc != ERROR_SUCCESS || kind != REG_SZ || len == 0 {
+        if rc != ERROR_SUCCESS || kind != REG_SZ {
             // SAFETY: valid handle.
             unsafe { RegCloseKey(hkey) };
             return None;
         }
-        let mut buf = vec![0u16; (len as usize).div_ceil(2)];
+        let Some(units) = validate_reg_len(len) else {
+            // SAFETY: valid handle.
+            unsafe { RegCloseKey(hkey) };
+            return None;
+        };
+        let mut buf = vec![0u16; units];
         let mut len2 = len;
         // SAFETY: buffer sized to the queried byte length.
         let rc = unsafe {
@@ -421,14 +450,10 @@ mod win {
         };
         // SAFETY: valid handle.
         unsafe { RegCloseKey(hkey) };
-        if rc != ERROR_SUCCESS {
+        if rc != ERROR_SUCCESS || kind != REG_SZ || len2 > len {
             return None;
         }
-        // Trim the trailing NUL(s).
-        while buf.last() == Some(&0) {
-            buf.pop();
-        }
-        Some(PathBuf::from(String::from_utf16_lossy(&buf)))
+        Some(PathBuf::from(decode_reg_sz(buf, len2)?))
     }
 }
 
@@ -436,6 +461,17 @@ mod win {
 mod tests {
     use super::*;
 
+    #[test]
+    fn registry_string_lengths_are_bounded_and_trimmed() {
+        assert_eq!(validate_reg_len(5), None, "odd byte lengths are invalid UTF-16");
+        assert_eq!(validate_reg_len(MAX_STEAM_PATH_BYTES + 2), None);
+        assert_eq!(validate_reg_len(8), Some(4));
+
+        let raw = "C:\\Steam\0ignored".encode_utf16().collect::<Vec<_>>();
+        let returned_bytes = ("C:\\Steam\0".encode_utf16().count() * 2) as u32;
+        assert_eq!(decode_reg_sz(raw, returned_bytes).as_deref(), Some("C:\\Steam"));
+    }
+
     #[test]
     fn running_app_id_reads_nonzero_and_rejects_zero() {
         let running = r#""Registry" { "HKCU" { "Software" { "Valve" { "Steam" {

src/network/gossip.rs

@@ -133,12 +133,13 @@ fn admit_state_mutation(
 
 fn peer_state_for_log(state: &PeerState) -> String {
     format!(
-        "name={:?}, muted={}, addr_id={}, addrs={}, sharing={}",
+        "name={:?}, muted={}, addr_id={}, addrs={}, sharing={}, game={:?}",
         state.name,
         state.is_muted,
         crate::short_id(&state.addr.id.to_string()),
         state.addr.addrs.len(),
-        state.sharing.is_some()
+        state.sharing.is_some(),
+        state.game
     )
 }
 
@@ -717,6 +718,16 @@ mod tests {
         EndpointAddr::from(id)
     }
 
+    #[test]
+    fn peer_state_log_includes_game() {
+        let mut state = sample_peer_state_for(fresh_id());
+        state.game = Some("Half-Life 2".to_string());
+        assert!(peer_state_for_log(&state).contains("game=Some(\"Half-Life 2\")"));
+
+        state.game = None;
+        assert!(peer_state_for_log(&state).contains("game=None"));
+    }
+
     #[test]
     fn bootstrap_client_dials_host() {
         // A non-host (client) with no retained peers dials just the ticket host.