diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index de67bb0..eef8642 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -163,11 +163,11 @@ pub async fn open_terminal( let session_id = output.session.id; // (2) Register the xterm output channel for this session. - state.pty_bridge.register(session_id, on_output); + let gen = state.pty_bridge.register(session_id, on_output); // (3) Subscribe to the PTY's byte stream and pump it to the channel. The // stream is a blocking iterator, so it runs on a dedicated OS thread; it - // ends when the PTY hits EOF (process exit) or the bridge channel is gone. + // ends when the PTY hits EOF (process exit) or this attach is superseded. let handle = PtyHandle { session_id }; match state.pty_port.subscribe_output(&handle) { Ok(stream) => { @@ -178,8 +178,9 @@ pub async fn open_terminal( break; } } - // Stream ended (process exited): drop the channel registration. - bridge.unregister(&session_id); + // Stream ended: drop the channel only if still ours (a re-attach + // may have superseded this generation — don't tear down its channel). + bridge.unregister_if(&session_id, gen); }); } Err(e) => { @@ -283,10 +284,12 @@ pub fn reattach_terminal( .map_err(|e| ErrorDto::from(AppError::from(e)))?; // (2) Register the new output channel for this session, replacing any stale - // one from a previous attach. - state.pty_bridge.register(sid, on_output); + // one from a previous attach (and bumping the generation). + let gen = state.pty_bridge.register(sid, on_output); // (3) Subscribe afresh to the live byte stream and pump it to the channel. + // The fresh subscription supersedes the previous attach's, so its pump thread + // ends and stops double-delivering this session's bytes. match state.pty_port.subscribe_output(&handle) { Ok(stream) => { let bridge: std::sync::Arc = std::sync::Arc::clone(&state.pty_bridge); @@ -296,7 +299,7 @@ pub fn reattach_terminal( break; } } - bridge.unregister(&sid); + bridge.unregister_if(&sid, gen); }); } Err(e) => { @@ -767,11 +770,11 @@ pub async fn launch_agent( let session_id = output.session.id; // Register the xterm output channel for this session. - state.pty_bridge.register(session_id, on_output); + let gen = state.pty_bridge.register(session_id, on_output); // Subscribe to the PTY's byte stream and pump it to the channel. // The stream is a blocking iterator; it runs on a dedicated OS thread and - // ends when the PTY hits EOF or the bridge channel is gone. + // ends when the PTY hits EOF or this attach is superseded. let handle = PtyHandle { session_id }; match state.pty_port.subscribe_output(&handle) { Ok(stream) => { @@ -782,7 +785,7 @@ pub async fn launch_agent( break; } } - bridge.unregister(&session_id); + bridge.unregister_if(&session_id, gen); }); } Err(e) => { diff --git a/crates/app-tauri/src/pty.rs b/crates/app-tauri/src/pty.rs index caf5293..fa43e8d 100644 --- a/crates/app-tauri/src/pty.rs +++ b/crates/app-tauri/src/pty.rs @@ -34,7 +34,11 @@ pub type PtyChunk = Vec; /// Thread-safe; cloned `Arc` is held in [`crate::state::AppState`]. #[derive(Default)] pub struct PtyBridge { - channels: Mutex>>, + /// Per session: a monotonically-increasing **generation** plus the current + /// output channel. The generation distinguishes successive (re-)attaches so a + /// superseded pump thread can't tear down the channel of the attach that + /// replaced it (see [`PtyBridge::unregister_if`]). + channels: Mutex)>>, } impl PtyBridge { @@ -46,21 +50,39 @@ impl PtyBridge { } } - /// Registers the output channel for a session (called when a terminal is - /// opened, from a `#[tauri::command]` that receives the `Channel` argument). - pub fn register(&self, session: SessionId, channel: Channel) { + /// Registers (or replaces) the output channel for a session and returns the + /// **generation** of this registration. Each call for a session bumps the + /// generation, so the caller's pump thread can later tear down *only its own* + /// registration via [`unregister_if`](Self::unregister_if). + pub fn register(&self, session: SessionId, channel: Channel) -> u64 { if let Ok(mut map) = self.channels.lock() { - map.insert(session, channel); + let gen = map.get(&session).map_or(0, |(g, _)| g.wrapping_add(1)); + map.insert(session, (gen, channel)); + gen + } else { + 0 } } - /// Removes a session's channel (terminal closed). + /// Removes a session's channel unconditionally (terminal explicitly closed). pub fn unregister(&self, session: &SessionId) { if let Ok(mut map) = self.channels.lock() { map.remove(session); } } + /// Removes a session's channel **only if** `gen` is still the current + /// generation. A pump thread calls this when its output stream ends: if the + /// session has since been re-attached (newer generation), this is a no-op, so + /// the dying thread never unregisters the live channel that superseded it. + pub fn unregister_if(&self, session: &SessionId, gen: u64) { + if let Ok(mut map) = self.channels.lock() { + if matches!(map.get(session), Some((g, _)) if *g == gen) { + map.remove(session); + } + } + } + /// Forwards a chunk of output bytes to a session's channel. /// /// Returns `true` if the chunk was delivered, `false` if no channel is @@ -71,7 +93,7 @@ impl PtyBridge { return false; }; match map.get(session) { - Some(channel) => channel.send(chunk).is_ok(), + Some((_, channel)) => channel.send(chunk).is_ok(), None => false, } } diff --git a/crates/app-tauri/tests/pty_bridge.rs b/crates/app-tauri/tests/pty_bridge.rs index 4c6b93d..f5f77d3 100644 --- a/crates/app-tauri/tests/pty_bridge.rs +++ b/crates/app-tauri/tests/pty_bridge.rs @@ -88,3 +88,47 @@ fn register_same_session_twice_replaces_channel() { assert!(first.lock().unwrap().is_empty(), "old channel no longer used"); assert_eq!(second.lock().unwrap().as_slice(), &[vec![9]]); } + +#[test] +fn register_returns_monotonic_generation_per_session() { + let bridge = PtyBridge::new(); + let session = sid(); + let g0 = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); + let g1 = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); + assert_eq!(g0, 0); + assert_eq!(g1, 1, "re-attaching the same session bumps the generation"); +} + +/// The regression guard for on-resize output duplication: a superseded attach's +/// pump thread (older generation) must NOT tear down the channel of the +/// re-attach that replaced it. +#[test] +fn unregister_if_is_a_noop_for_a_superseded_generation() { + let bridge = PtyBridge::new(); + let session = sid(); + let old = Arc::new(Mutex::new(Vec::new())); + let new = Arc::new(Mutex::new(Vec::new())); + + let old_gen = bridge.register(session, capturing_channel(Arc::clone(&old))); + let _new_gen = bridge.register(session, capturing_channel(Arc::clone(&new))); + + // The stale (old) pump thread ends and tries to clean up with its own gen. + bridge.unregister_if(&session, old_gen); + + // The current channel survives and still delivers — no duplication, no drop. + assert_eq!(bridge.active_sessions(), 1, "live re-attach must not be removed"); + assert!(bridge.send_output(&session, vec![7])); + assert_eq!(new.lock().unwrap().as_slice(), &[vec![7]]); +} + +#[test] +fn unregister_if_removes_when_generation_is_current() { + let bridge = PtyBridge::new(); + let session = sid(); + let gen = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new())))); + + // The current attach's pump thread ends (PTY EOF): it owns the live channel. + bridge.unregister_if(&session, gen); + assert_eq!(bridge.active_sessions(), 0); + assert!(!bridge.send_output(&session, vec![1])); +} diff --git a/crates/infrastructure/src/pty/mod.rs b/crates/infrastructure/src/pty/mod.rs index 3daa316..a58d6e4 100644 --- a/crates/infrastructure/src/pty/mod.rs +++ b/crates/infrastructure/src/pty/mod.rs @@ -57,12 +57,14 @@ const READ_BUF: usize = 8 * 1024; const SCROLLBACK_CAP: usize = 100 * 1024; /// The shared output hub of one PTY: a bounded scrollback ring buffer plus the -/// set of currently-subscribed receivers. The reader thread feeds both; views -/// subscribe and unsubscribe freely over the PTY's lifetime (re-attach support). +/// **single** currently-subscribed receiver. The reader thread feeds both; a +/// view subscribes on (re-)attach, and each new subscription supersedes the +/// previous one ([`Broadcast::subscribe`]) so a PTY is never fanned to two live +/// consumers at once. /// -/// Each subscriber is a [`Sender`]; a send failing (receiver dropped because the -/// view detached) prunes that subscriber on the next chunk. This is the -/// fan-out/broadcast that replaces the old single-take `output_rx`. +/// A subscriber is a [`Sender`]; a send failing (receiver dropped because the +/// view detached) prunes it on the next chunk. The `Vec` is retained (rather +/// than an `Option`) only to keep the prune-on-send logic uniform. #[derive(Default)] struct Broadcast { /// Bounded ring buffer of the most recent output bytes. @@ -88,11 +90,18 @@ impl Broadcast { /// Registers a new subscriber, returning its receiver. /// + /// **Single live consumer.** A PTY session is shown by exactly one view at a + /// time; a re-attach supersedes the previous view. So a new subscription + /// **drops every prior subscriber**, which ends the previous attach's pump + /// thread (its receiver closes) instead of leaving it alive to fan the *same* + /// bytes a second time — the root cause of on-resize output duplication. + /// /// If the PTY already hit EOF, the returned stream is immediately closed /// (its sender is dropped) so a late re-attach to a finished session doesn't /// block forever waiting for output that will never come. fn subscribe(&mut self) -> Receiver> { let (tx, rx) = mpsc::channel(); + self.subscribers.clear(); if !self.eof { self.subscribers.push(tx); } @@ -327,3 +336,43 @@ impl PtyPort for PortablePtyAdapter { fn exit_code(status: &portable_pty::ExitStatus) -> Option { Some(status.exit_code() as i32) } + +#[cfg(test)] +mod tests { + use super::Broadcast; + + /// Regression: a re-attach (new subscribe) must end the previous subscriber's + /// stream so only ONE consumer is fed. Two live subscribers would each pump + /// the same bytes to the bridge → the on-resize output duplication. + #[test] + fn subscribe_supersedes_the_previous_subscriber() { + let mut hub = Broadcast::default(); + let first = hub.subscribe(); + let second = hub.subscribe(); + + hub.push(b"hello"); + + // The superseded receiver's stream is closed (sender dropped) and gets + // nothing; only the current subscriber receives the chunk. + assert!(first.recv().is_err(), "old subscriber must be disconnected"); + assert_eq!(second.recv().unwrap(), b"hello".to_vec()); + } + + #[test] + fn subscribe_after_eof_yields_a_closed_stream() { + let mut hub = Broadcast::default(); + hub.close_subscribers(); + let rx = hub.subscribe(); + assert!(rx.recv().is_err(), "no output will ever come after EOF"); + } + + #[test] + fn snapshot_retains_scrollback_across_resubscribe() { + let mut hub = Broadcast::default(); + let _first = hub.subscribe(); + hub.push(b"abc"); + // A re-attach drops the old subscriber but the scrollback is preserved. + let _second = hub.subscribe(); + assert_eq!(hub.snapshot(), b"abc".to_vec()); + } +}