screenshare/core: a host that dies is no longer advertised as sharing (S2)

The defect: pixelpass's stdout EOF was silently discarded, the notice
channel existed only for app-audio shares, and nothing cleared the host
from the teardown slot or the ticket from presence — so a crashed host
stayed advertised in the room and the UI kept saying "sharing".

Every share now gets a notice channel. The drain task synthesizes a
terminal HostNotice::Eof when the stream ends (EOF or read error — a
crash can abort across `extern "C"` before any JSON line is written, so
the stream ending is the only reliable death signal). The core's
forwarder turns that into a ScreenShareHostFault scoped to the spawn's
generation; a stale fault (already stopped, or a newer share running) is
dropped. The handler reaps the child through the existing confirmed-reap
path, pulls the ticket off presence, and emits ScreenShareStopped BEFORE
the error, so the UI never shows "sharing" next to the explanation.

The ticket and its generation live in one ActiveShare value on purpose:
they must appear and vanish together, or the staleness gate drifts.

Both drain gates are mutation-verified: swallowing the Eof fails both
tests; skipping it only on the read-error path fails exactly the
error-path test.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/core/messages.rs

@@ -137,6 +137,16 @@ pub enum CoreCommand {
     /// Stop sharing our screen: kill the pixelpass host and clear the presence
     /// ticket. No-op when not sharing.
     StopScreenShare,
+    /// **Core-internal.** The running pixelpass host's stdout ended — the
+    /// process died (or its event stream broke), so the share identified by
+    /// `generation` is over: reap the child, pull the ticket off presence, and
+    /// tell the user. Synthesized by the core's own notice-forwarder task; the
+    /// UI never sends it. `generation` scopes the fault to one specific host
+    /// spawn, so a stale fault (the user already stopped, or started a new
+    /// share) is ignored rather than tearing down the wrong share.
+    ScreenShareHostFault {
+        generation: u64,
+    },
     /// Watch a peer's screen share: spawn a pixelpass viewer for `ticket` and
     /// open it in a local player.
     ViewShare {
@@ -271,6 +281,7 @@ pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass {
             quality: _,
         }
         | CoreCommand::StopScreenShare
+        | CoreCommand::ScreenShareHostFault { generation: _ }
         | CoreCommand::ViewShare {
             ticket: _,
             settings: _,
@@ -363,6 +374,7 @@ pub fn coalesce_key(cmd: &CoreCommand) -> Option<CoalesceKey> {
             quality: _,
         }
         | CoreCommand::StopScreenShare
+        | CoreCommand::ScreenShareHostFault { generation: _ }
         | CoreCommand::ViewShare {
             ticket: _,
             settings: _,

src/core/mod.rs

@@ -1397,10 +1397,25 @@ async fn run_core_loop(
     // later opt-in can immediately publish whatever is currently running.
     let mut current_game: Option<crate::game::DetectedGame> = None;
     let mut network_mode = NetworkMode::default();
-    // Pixelpass binary override (config), and the ticket of our own active screen
+    // Pixelpass binary override (config), and our own active screen share: the
-    // share (rides our presence so the room — incl. late joiners — can watch).
+    // ticket rides our presence so the room — incl. late joiners — can watch,
+    // and the generation ties host-fault notices to this specific host spawn
+    // (see `ScreenShareHostFault`). One variable on purpose: the ticket and the
+    // generation must appear and vanish together, or a stale fault could tear
+    // down a share it doesn't belong to.
     let mut pixelpass_override: Option<String> = None;
-    let mut current_sharing: Option<String> = None;
+    struct ActiveShare {
+        generation: u64,
+        ticket: String,
+    }
+    let mut current_sharing: Option<ActiveShare> = None;
+    // Monotonic per-spawn counter feeding `ActiveShare::generation`.
+    let mut share_generations: u64 = 0;
+    // Host faults re-enter the loop here (the notice-forwarder task can't touch
+    // loop state). The loop keeps `host_fault_tx` to clone into each share's
+    // forwarder, so this channel never closes — the select arm's `Some` pattern
+    // is total in practice and a closed-channel branch would be unreachable.
+    let (host_fault_tx, mut host_fault_rx) = mpsc::unbounded_channel::<u64>();
 
     let mut active_session: Option<ActiveSession> = None;
     // Standalone capture-only mic meter, live only when no session exists.
@@ -1549,6 +1564,13 @@ async fn run_core_loop(
                 // reachable it is already covered — nothing to add here.
                 None => break,
             },
+            // A share's notice-forwarder task reported the host's stdout ended.
+            // The `Some` pattern is total: this loop owns `host_fault_tx` (see
+            // its declaration), so the channel cannot close — no `None` arm is
+            // written because one would be unreachable by construction.
+            Some(generation) = host_fault_rx.recv() => {
+                CoreCommand::ScreenShareHostFault { generation }
+            }
             game_change = next_game_change(&mut game_rx) => {
                 // The detector worker published a new debounced game (or `None`).
                 let Some(detected) = game_change else {
@@ -1567,7 +1589,7 @@ async fn run_core_loop(
                         let self_state = presence.to_state(
                             is_muted.load(Ordering::Relaxed),
                             net.endpoint.addr(),
-                            current_sharing.clone(),
+                            current_sharing.as_ref().map(|s| s.ticket.clone()),
                         );
                         let _ = session.room_state.update_self_state(self_state).await;
                     }
@@ -2828,8 +2850,11 @@ async fn run_core_loop(
                 is_muted.store(new_state, Ordering::Relaxed);
 
                 if let Some(session) = &active_session {
-                    let self_state =
+                    let self_state = presence.to_state(
-                        presence.to_state(new_state, net.endpoint.addr(), current_sharing.clone());
+                        new_state,
+                        net.endpoint.addr(),
+                        current_sharing.as_ref().map(|s| s.ticket.clone()),
+                    );
                     let _ = session.room_state.update_self_state(self_state).await;
                 }
             }
@@ -2842,7 +2867,7 @@ async fn run_core_loop(
                     let self_state = presence.to_state(
                         is_muted.load(Ordering::Relaxed),
                         net.endpoint.addr(),
-                        current_sharing.clone(),
+                        current_sharing.as_ref().map(|s| s.ticket.clone()),
                     );
                     let _ = session.room_state.update_self_state(self_state).await;
                 }
@@ -3155,7 +3180,7 @@ async fn run_core_loop(
                     let self_state = presence.to_state(
                         is_muted.load(Ordering::Relaxed),
                         net.endpoint.addr(),
-                        current_sharing.clone(),
+                        current_sharing.as_ref().map(|s| s.ticket.clone()),
                     );
                     let _ = session.room_state.update_self_state(self_state).await;
                 }
@@ -3361,7 +3386,7 @@ async fn run_core_loop(
                     let self_state = presence.to_state(
                         is_muted.load(Ordering::Relaxed),
                         net.endpoint.addr(),
-                        current_sharing.clone(),
+                        current_sharing.as_ref().map(|s| s.ticket.clone()),
                     );
                     let _ = session.room_state.update_self_state(self_state).await;
                 }
@@ -3435,17 +3460,24 @@ async fn run_core_loop(
                         continue;
                     }
                 };
-                // Forward pixelpass `app_audio` events (only emitted when an app
+                // Every share gets a notice forwarder — not just app-audio ones.
-                // is selected) to the UI so it can warn when the chosen app's
+                // pixelpass `app_audio` events (only emitted when an app is
-                // audio drops. The channel closes when the host dies (drain hits
+                // selected) become UI warnings, and the drain's terminal `Eof`
-                // EOF), ending the forwarder task on its own.
+                // becomes a host fault scoped to this spawn's generation, so a
-                let notices = audio_app.as_deref().map(|_| {
+                // host that dies is torn down instead of staying advertised in
-                    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<
+                // presence forever. On a failed spawn the sender is dropped
-                        crate::screenshare::PixelpassEvent,
+                // before the drain ever runs, so the forwarder just ends and no
-                    >();
+                // fault is sent (the spawn error carries the news instead).
+                share_generations += 1;
+                let generation = share_generations;
+                let (notices_tx, mut notices_rx) =
+                    tokio::sync::mpsc::unbounded_channel::<crate::screenshare::HostNotice>();
                 let ui_tx_notices = ui_tx.clone();
+                let fault_tx = host_fault_tx.clone();
                 tokio::spawn(async move {
-                        while let Some(ev) = rx.recv().await {
+                    while let Some(notice) = notices_rx.recv().await {
+                        match notice {
+                            crate::screenshare::HostNotice::Event(ev) => {
                                 let active = match ev {
                                     crate::screenshare::PixelpassEvent::AppAudioRouted => true,
                                     crate::screenshare::PixelpassEvent::AppAudioLost => false,
@@ -3459,22 +3491,31 @@ async fn run_core_loop(
                                     break;
                                 }
                             }
-                    });
+                            // Terminal by contract: nothing follows on the
-                    tx
+                            // channel, so the task ends here.
+                            crate::screenshare::HostNotice::Eof => {
+                                let _ = fault_tx.send(generation);
+                                break;
+                            }
+                        }
+                    }
                 });
                 match crate::screenshare::spawn_host(
                     &bin,
                     audio_app.as_deref(),
                     &settings,
                     quality,
-                    notices,
+                    notices_tx,
                 )
                 .await
                 {
                     Ok((child, ticket)) => {
                         crate::log_msg("Screen share host started");
                         session.teardown.set_host(child);
-                        current_sharing = Some(ticket.clone());
+                        current_sharing = Some(ActiveShare {
+                            generation,
+                            ticket: ticket.clone(),
+                        });
                         let self_state = presence.to_state(
                             is_muted.load(Ordering::Relaxed),
                             net.endpoint.addr(),
@@ -3523,6 +3564,58 @@ async fn run_core_loop(
                 let _ = ui_tx.send(UiEvent::ScreenShareStopped).await;
             }
 
+            CoreCommand::ScreenShareHostFault { generation } => {
+                // Stale unless it names the share we are advertising RIGHT NOW.
+                // Every deliberate end of a share (StopScreenShare, Leave, a
+                // fresh Join) clears `current_sharing` before or while reaping
+                // the child, and the reaped child's stdout EOF then arrives
+                // here late — dropping it is the correct handling, not an edge
+                // case. A mismatched generation likewise: that fault belongs to
+                // an older spawn than the share now running.
+                let stale = current_sharing.as_ref().map(|s| s.generation) != Some(generation);
+                if stale {
+                    continue;
+                }
+                crate::log_msg(
+                    "Screen share host died (stdout EOF with the share still advertised)",
+                );
+                current_sharing = None;
+                let mut unconfirmed = false;
+                if let Some(session) = &mut active_session {
+                    // The child is usually already dead, so this confirms the
+                    // reap immediately; if it merely closed stdout and lives
+                    // on, this is the SIGINT → grace → SIGKILL path. Either
+                    // way the dead-or-dying child leaves the teardown slot, so
+                    // `is_sharing` stops lying.
+                    unconfirmed = matches!(
+                        session.teardown.stop_host().await,
+                        Some(teardown::StopOutcome::Unconfirmed)
+                    );
+                    // Pull the ticket off presence so nobody clicks Watch on a
+                    // dead share.
+                    let self_state = presence.to_state(
+                        is_muted.load(Ordering::Relaxed),
+                        net.endpoint.addr(),
+                        None,
+                    );
+                    let _ = session.room_state.update_self_state(self_state).await;
+                }
+                // Stopped first — it clears the UI's sharing state — and only
+                // then the error explaining why, so the user is never left
+                // looking at a "sharing" UI with an error beside it.
+                let _ = ui_tx.send(UiEvent::ScreenShareStopped).await;
+                let detail = if unconfirmed {
+                    " Its process also couldn't be confirmed dead — check for a stray pixelpass."
+                } else {
+                    ""
+                };
+                let _ = ui_tx
+                    .send(UiEvent::Error(format!(
+                        "Screen share ended unexpectedly — pixelpass exited.{detail}"
+                    )))
+                    .await;
+            }
+
             CoreCommand::ViewShare { ticket, settings } => {
                 let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) {
                     Some(b) => b,

src/screenshare/mod.rs

@@ -90,6 +90,21 @@ pub enum PixelpassEvent {
     Other,
 }
 
+/// What the host's stdout drain forwards to the core over the notice channel.
+///
+/// `Eof` is **synthesized here**, not parsed: pixelpass has no "I died" event,
+/// and a crash can abort across `extern "C"` before any JSON line is written,
+/// so the stream ending is the only reliable death signal. A read *error*
+/// counts too — either way the event stream is gone and the host must be
+/// treated as over.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum HostNotice {
+    /// A parsed pixelpass event line.
+    Event(PixelpassEvent),
+    /// The host's stdout ended (EOF or read error). Terminal: nothing follows.
+    Eof,
+}
+
 /// Parse a single stdout line from pixelpass `--output json`. Pure: no I/O.
 pub fn parse_pixelpass_event(line: &str) -> Option<PixelpassEvent> {
     let line = line.trim();
@@ -370,7 +385,10 @@ pub fn is_available(config_override: Option<&str>) -> bool {
 /// `audio_app` is `Some`, pixelpass captures only that app's audio instead of the
 /// whole desktop sink, which avoids the call-loopback echo (A23). The child keeps
 /// running (streaming to viewers) until killed or dropped; remaining stdout is
-/// drained in a background task so a full pipe can't stall the host. We do
+/// drained in a background task so a full pipe can't stall the host. The drain
+/// forwards every parsed event over `notices` and — the part no share may opt
+/// out of — a terminal [`HostNotice::Eof`] when the stream ends, which is the
+/// caller's only reliable signal that the host died. We do
 /// not pass encode/viewer overrides unless the local settings explicitly ask for
 /// them, so pixelpass keeps its own defaults in the common case.
 pub async fn spawn_host(
@@ -378,7 +396,7 @@ pub async fn spawn_host(
     audio_app: Option<&str>,
     settings: &ScreenShareSettings,
     quality: ShareQuality,
-    notices: Option<tokio::sync::mpsc::UnboundedSender<PixelpassEvent>>,
+    notices: tokio::sync::mpsc::UnboundedSender<HostNotice>,
 ) -> std::io::Result<(Child, String)> {
     let args = host_args(audio_app, settings, quality);
     // Log the exact argv we hand pixelpass so a field log can confirm which
@@ -433,7 +451,7 @@ pub async fn spawn_host(
     if let Some(stderr) = stderr {
         drain_stderr_in_background(stderr);
     }
-    drain_in_background(lines, "host", notices);
+    drain_in_background(lines, "host", Some(notices));
     Ok((child, ticket))
 }
 
@@ -572,13 +590,15 @@ where
 
 /// Keep reading the child's stdout to EOF in the background so a full pipe can't
 /// stall it; log notable events for diagnostics. When `notices` is `Some`, each
-/// parsed event is also forwarded to the caller (the core, which translates the
+/// parsed event is also forwarded to the caller (the core), and when the stream
-/// `app_audio` ones into a UI warning); a send failure (receiver dropped) just
+/// ends — EOF or read error, i.e. the child exited or its event stream broke —
-/// stops forwarding, draining continues. The task ends on EOF (child exited).
+/// a final [`HostNotice::Eof`] is sent so the caller learns the child is gone
+/// (a host that dies must not stay advertised as sharing). A send failure
+/// (receiver dropped) just stops forwarding, draining continues.
 fn drain_in_background<R>(
     mut lines: tokio::io::Lines<BufReader<R>>,
     role: &'static str,
-    notices: Option<tokio::sync::mpsc::UnboundedSender<PixelpassEvent>>,
+    notices: Option<tokio::sync::mpsc::UnboundedSender<HostNotice>>,
 ) where
     R: tokio::io::AsyncRead + Unpin + Send + 'static,
 {
@@ -587,10 +607,14 @@ fn drain_in_background<R>(
             if let Some(ev) = parse_pixelpass_event(&line) {
                 crate::log_msg(&format!("pixelpass {role}: {}", event_for_log(&ev)));
                 if let Some(tx) = &notices {
-                    let _ = tx.send(ev);
+                    let _ = tx.send(HostNotice::Event(ev));
                 }
             }
         }
+        if let Some(tx) = &notices {
+            crate::log_msg(&format!("pixelpass {role}: stdout ended"));
+            let _ = tx.send(HostNotice::Eof);
+        }
     });
 }
 
@@ -1376,4 +1400,63 @@ Install hint: sudo apt install gstreamer1.0-plugins-bad
         #[cfg(not(windows))]
         assert_eq!(candidates, vec![dir.join("pixelpass")]);
     }
+
+    /// The host-fault contract, clean-exit half: events are forwarded in order
+    /// and the stream ending yields exactly one terminal [`HostNotice::Eof`],
+    /// after which the drain task drops its sender (the closed channel is what
+    /// ends the core's forwarder). A host that dies silently — EOF swallowed —
+    /// is the S2 defect: the dead share stays advertised in presence.
+    #[tokio::test]
+    async fn drain_forwards_events_then_synthesizes_eof_when_stdout_ends() {
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let (read_half, mut write_half) = tokio::io::duplex(1024);
+        drain_in_background(BufReader::new(read_half).lines(), "test", Some(tx));
+
+        use tokio::io::AsyncWriteExt;
+        write_half
+            .write_all(b"{\"event\":\"app_audio\",\"state\":\"routed\"}\nnot json\n")
+            .await
+            .unwrap();
+        drop(write_half); // child exited: stdout EOF
+
+        assert_eq!(
+            rx.recv().await,
+            Some(HostNotice::Event(PixelpassEvent::AppAudioRouted))
+        );
+        // The non-JSON line is dropped, not forwarded.
+        assert_eq!(rx.recv().await, Some(HostNotice::Eof));
+        assert_eq!(rx.recv().await, None, "task ended and dropped the sender");
+    }
+
+    /// The host-fault contract, broken-stream half: a read *error* (not a tidy
+    /// EOF) must synthesize the same terminal `Eof` — the event stream is gone
+    /// either way, and only the drain task can tell the core so.
+    #[tokio::test]
+    async fn drain_synthesizes_eof_on_a_read_error_too() {
+        struct BrokenPipe;
+        impl tokio::io::AsyncRead for BrokenPipe {
+            fn poll_read(
+                self: std::pin::Pin<&mut Self>,
+                _cx: &mut std::task::Context<'_>,
+                _buf: &mut tokio::io::ReadBuf<'_>,
+            ) -> std::task::Poll<std::io::Result<()>> {
+                std::task::Poll::Ready(Err(std::io::Error::other("stream broke")))
+            }
+        }
+
+        use tokio::io::AsyncReadExt;
+        // One good event line, then the stream breaks mid-read.
+        let reader =
+            std::io::Cursor::new(b"{\"event\":\"capture\",\"state\":\"started\"}\n".to_vec())
+                .chain(BrokenPipe);
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        drain_in_background(BufReader::new(reader).lines(), "test", Some(tx));
+
+        assert_eq!(
+            rx.recv().await,
+            Some(HostNotice::Event(PixelpassEvent::CaptureStarted))
+        );
+        assert_eq!(rx.recv().await, Some(HostNotice::Eof));
+        assert_eq!(rx.recv().await, None, "task ended and dropped the sender");
+    }
 }