core: one screen-share player per share — re-watch replaces, not stacks

Every click of Watch (`CoreCommand::ViewShare`) spawned a fresh pixelpass
viewer + mpv and pushed it onto an untracked Vec. A field test hit the
consequence: the first click gave a frozen player (the host's capture was
stalling), so the viewer clicked again to retry — and got a SECOND mpv,
doubling the shared audio.

Track viewers paired with their share ticket. On ViewShare, reap players
whose window already closed (try_wait), then if a live player for the same
ticket exists, kill it before spawning the replacement. Re-watching a
share now swaps its player instead of stacking a second one. Pure
`replace_viewer_index` seam + test.

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

src/core/mod.rs

@@ -662,9 +662,11 @@ struct ActiveSession {
     /// Our pixelpass screen-share host child while sharing (`kill_on_drop`, so it
     /// also dies if the session is dropped without an explicit stop).
     screenshare_host: Option<tokio::process::Child>,
-    /// pixelpass viewer children we spawned to watch peers' shares; killed on
-    /// session teardown (each also self-exits when its player window closes).
-    screenshare_viewers: Vec<tokio::process::Child>,
+    /// pixelpass viewer children we spawned to watch peers' shares, each paired
+    /// with the share ticket it's viewing so a re-watch of the same share can
+    /// replace (not stack) its player. Killed on session teardown (each also
+    /// self-exits when its player window closes).
+    screenshare_viewers: Vec<(String, tokio::process::Child)>,
 }
 
 impl ActiveSession {
@@ -676,7 +678,7 @@ impl ActiveSession {
         if let Some(mut host) = self.screenshare_host.take() {
             let _ = host.kill().await;
         }
-        for mut viewer in self.screenshare_viewers.drain(..) {
+        for (_, mut viewer) in self.screenshare_viewers.drain(..) {
             let _ = viewer.kill().await;
         }
         self.datagram_task.abort();
@@ -2497,7 +2499,7 @@ async fn run_core_loop(
                     #[cfg(target_os = "linux")]
                     echo_cancel: echo_cancel_guard,
                     screenshare_host: None,
-                    screenshare_viewers: Vec::new(),
+                    screenshare_viewers: Vec::<(String, tokio::process::Child)>::new(),
                 };
 
                 let self_id = endpoint.id().to_string();
@@ -3221,11 +3223,27 @@ async fn run_core_loop(
                         continue;
                     }
                 };
+                if let Some(session) = &mut active_session {
+                    // Drop viewers whose player window has already closed so the
+                    // list only tracks live players.
+                    session
+                        .screenshare_viewers
+                        .retain_mut(|(_, child)| !matches!(child.try_wait(), Ok(Some(_))));
+                    // One player per share: a second Watch click on a share we're
+                    // already viewing is a retry (usually because the first window
+                    // froze), so replace the existing player rather than stacking a
+                    // second mpv — two players would double the shared audio.
+                    if let Some(pos) = replace_viewer_index(&session.screenshare_viewers, &ticket) {
+                        let (_, mut old) = session.screenshare_viewers.remove(pos);
+                        let _ = old.kill().await;
+                        crate::log_msg("Screen share viewer replaced (re-watch)");
+                    }
+                }
                 match crate::screenshare::spawn_viewer(&bin, &ticket).await {
                     Ok(child) => {
                         crate::log_msg("Screen share viewer started");
                         if let Some(session) = &mut active_session {
-                            session.screenshare_viewers.push(child);
+                            session.screenshare_viewers.push((ticket, child));
                         }
                     }
                     Err(e) => {
@@ -3241,14 +3259,23 @@ async fn run_core_loop(
     Ok(())
 }
 
+/// Index of an existing viewer for `ticket` in the live-viewers list, if any.
+/// A re-watch of the same share replaces that player instead of stacking a
+/// second one — two players decoding the same stream would double the shared
+/// audio. Generic over the child value so the dedup rule is unit-testable
+/// without spawning real player processes.
+fn replace_viewer_index<T>(viewers: &[(String, T)], ticket: &str) -> Option<usize> {
+    viewers.iter().position(|(t, _)| t == ticket)
+}
+
 #[cfg(test)]
 mod tests {
     use super::{
         KnownPeers, MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, MIC_LEVEL_REPORT_SAMPLES, MicLevelMeter,
         PLAYBACK_HANDOFF_QUEUE_FRAMES, PeerSpeakTicket, admit_retained, apply_peer_volume,
         apply_volume, audio_datagram_len_ok, coalesce_insert, coalesce_pop, frame_level,
-        mix_frames, mix_stereo_frames, next_game_change, send_playback_frame, should_auto_fetch,
-        stereo_to_mono,
+        mix_frames, mix_stereo_frames, next_game_change, replace_viewer_index, send_playback_frame,
+        should_auto_fetch, stereo_to_mono,
     };
     use crate::core::messages::{CoalesceKey, CoreCommand, coalesce_key};
     use std::collections::{HashMap, HashSet};
@@ -3259,6 +3286,19 @@ mod tests {
         iroh::SecretKey::generate().public()
     }
 
+    #[test]
+    fn re_watch_replaces_existing_viewer_for_same_ticket() {
+        // The value type stands in for a viewer Child; only the ticket matters.
+        let viewers = vec![("ticket-A".to_string(), 0u8), ("ticket-B".to_string(), 1u8)];
+        // Re-watching an already-open share finds the existing player to replace.
+        assert_eq!(replace_viewer_index(&viewers, "ticket-A"), Some(0));
+        assert_eq!(replace_viewer_index(&viewers, "ticket-B"), Some(1));
+        // A different (new) share has nothing to replace — it opens fresh.
+        assert_eq!(replace_viewer_index(&viewers, "ticket-C"), None);
+        // Empty list: first watch of anything opens fresh.
+        assert_eq!(replace_viewer_index::<u8>(&[], "ticket-A"), None);
+    }
+
     #[test]
     fn admit_retained_rejects_only_new_ids_at_the_cap() {
         // Below the cap, a brand-new identity is retained.