fix(terminal): raw-drain while child exits

Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@buzz.block.builderlab.xyz>
Signed-off-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@buzz.block.builderlab.xyz>

desktop/src-tauri/crates/buzz-terminal/src/lifecycle.rs

@@ -232,7 +232,13 @@ pub const MAX_LIVE_SESSIONS: usize = 20;
 /// drained. Join the reader first and every tab close pays that, on the arm
 /// where the user is already waiting.
 pub trait DrainingReader {
-    /// Stops consuming and releases the reader. Called *after* reap.
+    /// Detach parser work and enter raw-drain mode before child termination.
+    fn begin_closing(&self);
+
+    /// Wake a reader that remains blocked after the child has been reaped.
+    fn stop(&self);
+
+    /// Releases the reader thread. Called only after [`DrainingReader::stop`].
     fn join(self: Box<Self>);
 }
 
@@ -249,11 +255,13 @@ pub fn shutdown_draining(
     child: &mut Box<dyn Child + Send + Sync>,
     reader: Box<dyn DrainingReader>,
 ) -> io::Result<Shutdown> {
+    reader.begin_closing();
     // Terminate and reap with the reader still running, so the child never
     // blocks in a tty write while we are waiting on it.
     let outcome = shutdown(child);
-    // Unconditional: the reader must be released whether or not the shutdown
-    // reported an error, or a failed close leaks the thread and its fd.
+    // Unconditional: wake and release the reader whether or not shutdown
+    // reported an error. EOF may already have ended it; stop is idempotent.
+    reader.stop();
     reader.join();
     outcome
 }

desktop/src-tauri/crates/buzz-terminal/src/lifecycle_tests.rs

@@ -430,7 +430,8 @@ fn reader_drains_through_termination_and_reap() {
 
     let after_close = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
     let closing = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
-    let reader = RecordingReader::spawn(&pair, after_close.clone(), closing.clone());
+    let order = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
+    let reader = RecordingReader::spawn(&pair, after_close.clone(), closing.clone(), order.clone());
 
     // Let the child get well ahead of the reader before we touch anything.
     assert!(
@@ -461,6 +462,11 @@ fn reader_drains_through_termination_and_reap() {
          period: the child was blocked writing to an undrained master rather \
          than exiting on SIGTERM"
     );
+    assert_eq!(
+        *order.lock().unwrap(),
+        ["begin_closing", "stop", "join"],
+        "reader close must begin before termination and stop/join only after reap"
+    );
 }
 
 /// Spawns a child that floods the PTY without pause.
@@ -478,6 +484,7 @@ fn spawn_noisy(pair: &PtyPair) -> Box<dyn Child + Send + Sync> {
 struct RecordingReader {
     total: std::sync::Arc<std::sync::atomic::AtomicU64>,
     handle: std::thread::JoinHandle<()>,
+    order: std::sync::Arc<std::sync::Mutex<Vec<&'static str>>>,
 }
 
 impl RecordingReader {
@@ -485,6 +492,7 @@ impl RecordingReader {
         pair: &PtyPair,
         after_close: std::sync::Arc<std::sync::atomic::AtomicU64>,
         closing: std::sync::Arc<std::sync::atomic::AtomicBool>,
+        order: std::sync::Arc<std::sync::Mutex<Vec<&'static str>>>,
     ) -> Self {
         let mut reader = pair.master.try_clone_reader().expect("reader");
         let total = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
@@ -502,7 +510,11 @@ impl RecordingReader {
                 }
             }
         });
-        Self { total, handle }
+        Self {
+            total,
+            handle,
+            order,
+        }
     }
 
     fn total_bytes(&self) -> u64 {
@@ -511,7 +523,16 @@ impl RecordingReader {
 }
 
 impl DrainingReader for RecordingReader {
+    fn begin_closing(&self) {
+        self.order.lock().unwrap().push("begin_closing");
+    }
+
+    fn stop(&self) {
+        self.order.lock().unwrap().push("stop");
+    }
+
     fn join(self: Box<Self>) {
+        self.order.lock().unwrap().push("join");
         // Bounded, and that is the whole point. The read loop ends when the
         // master reports EOF, which only happens once the reaped child has
         // released the slave -- so joining *before* termination blocks

desktop/src-tauri/crates/buzz-terminal/src/shared.rs

@@ -14,7 +14,7 @@
 //! metered separately: pooling them would let the reader's millions of fast
 //! acquires dilute the renderer's tail into a false pass.
 
-use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::time::Instant;
 
 use alacritty_terminal::sync::FairMutex;
@@ -126,6 +126,7 @@ pub struct SharedTerminal {
     term: FairMutex<Terminal>,
     reader: AcquireMeter,
     renderer: AcquireMeter,
+    closing: AtomicBool,
 }
 
 impl SharedTerminal {
@@ -134,6 +135,7 @@ impl SharedTerminal {
             term: FairMutex::new(term),
             reader: AcquireMeter::default(),
             renderer: AcquireMeter::default(),
+            closing: AtomicBool::new(false),
         }
     }
 
@@ -156,13 +158,23 @@ impl SharedTerminal {
     /// the whole buffer under one acquisition is what an unbounded hold *is*,
     /// so it is not offered here.
     pub fn feed(&self, bytes: &[u8]) -> bool {
-        self.acquire(&self.reader).feed(bytes)
+        let mut term = self.acquire(&self.reader);
+        if self.closing.load(Ordering::Acquire) {
+            false
+        } else {
+            term.feed(bytes)
+        }
     }
 
     /// Parse more of the pending tail under a fresh acquisition. Reader
     /// plane. Returns whether any remains.
     pub fn drain(&self) -> bool {
-        self.acquire(&self.reader).drain()
+        let mut term = self.acquire(&self.reader);
+        if self.closing.load(Ordering::Acquire) {
+            false
+        } else {
+            term.drain()
+        }
     }
 
     /// Feed and pump to completion, re-acquiring between slices.
@@ -173,6 +185,17 @@ impl SharedTerminal {
         }
     }
 
+    /// Atomically enter close mode and discard parser work. Subsequent PTY
+    /// bytes are raw-drained by the embedder and never reach callbacks.
+    pub fn begin_closing(&self) -> usize {
+        self.closing.store(true, Ordering::Release);
+        self.acquire(&self.reader).abandon_tail()
+    }
+
+    pub fn is_closing(&self) -> bool {
+        self.closing.load(Ordering::Acquire)
+    }
+
     /// Sample damage and encode a frame. Renderer plane.
     ///
     /// The lock covers the copy only; `encode` -- hashing, span grouping,
@@ -236,3 +259,31 @@ impl SharedTerminal {
         guard
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{Fences, Size};
+
+    #[test]
+    fn closing_abandons_tail_and_permanently_refuses_parser_callbacks() {
+        let (terminal, _actions) = Terminal::new(Size::default(), Fences::ALL);
+        let shared = SharedTerminal::new(terminal);
+        let payload = b"\x1b#8".repeat(10_000);
+
+        assert!(shared.feed(&payload), "fixture must create parser tail");
+        let before = shared.lock().stats();
+        let abandoned = shared.begin_closing();
+        assert!(abandoned > 0, "close must abandon without draining first");
+        assert!(shared.is_closing());
+
+        assert!(!shared.feed(b"parser callback after close"));
+        assert!(!shared.drain());
+        let after = shared.lock().stats();
+        assert_eq!(after.completed_units, before.completed_units);
+        assert_eq!(
+            after.abandoned_bytes,
+            before.abandoned_bytes + abandoned as u64
+        );
+    }
+}

desktop/src-tauri/src/terminal_runtime.rs

@@ -1,6 +1,7 @@
 //! Rust-owned PTY sessions and the typed Tauri transport for Buzz Substrate.
 
 use std::io::{Read, Write};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::thread::JoinHandle;
 
@@ -202,10 +203,23 @@ fn wire_publication(publication: Publication) -> Result<FrameMessage> {
     })
 }
 
-struct ReaderThread(Option<JoinHandle<()>>);
+struct ReaderThread {
+    handle: Option<JoinHandle<()>>,
+    terminal: Arc<SharedTerminal>,
+    stopping: Arc<AtomicBool>,
+}
+
 impl buzz_terminal::lifecycle::DrainingReader for ReaderThread {
+    fn begin_closing(&self) {
+        self.terminal.begin_closing();
+    }
+
+    fn stop(&self) {
+        self.stopping.store(true, Ordering::Release);
+    }
+
     fn join(mut self: Box<Self>) {
-        if let Some(handle) = self.0.take() {
+        if let Some(handle) = self.handle.take() {
             let _ = handle.join();
         }
     }
@@ -253,7 +267,12 @@ impl Session {
         if let Ok(mut channel) = self.channel.lock() {
             *channel = None;
         }
-        // The slave closes on child reap; the reader continues draining until then.
+        // Publication is detached before the reader enters close mode; the
+        // lifecycle helper then abandons parser work and keeps raw-draining
+        // through child termination and reap.
+        if let Ok(mut publisher) = self.publisher.lock() {
+            publisher.close();
+        }
         if let Some(reader) = self.reader.take() {
             #[cfg(unix)]
             {
@@ -261,12 +280,13 @@ impl Session {
             }
             #[cfg(not(unix))]
             {
+                reader.begin_closing();
                 let _ = self.child.kill();
                 let _ = self.child.wait();
+                reader.stop();
                 reader.join();
             }
         }
-        self.master.take();
     }
 }
 
@@ -465,6 +485,8 @@ pub(crate) fn terminal_attach(
     let reader_terminal = Arc::clone(&terminal);
     let reader_publisher = Arc::clone(&publisher);
     let reader_channel = Arc::clone(&channel);
+    let reader_stopping = Arc::new(AtomicBool::new(false));
+    let thread_stopping = Arc::clone(&reader_stopping);
     let reader_handle = std::thread::spawn(move || {
         let mut buffer = [0u8; 16 * 1024];
         let mut encoder = buzz_terminal::damage::Encoder::new();
@@ -474,7 +496,19 @@ pub(crate) fn terminal_attach(
                 Ok(0) | Err(_) => break,
                 Ok(count) => count,
             };
-            reader_terminal.feed(&buffer[..count]);
+            if thread_stopping.load(Ordering::Acquire) {
+                break;
+            }
+            if reader_terminal.is_closing() {
+                continue;
+            }
+            let mut more = reader_terminal.feed(&buffer[..count]);
+            while more && !reader_terminal.is_closing() {
+                more = reader_terminal.drain();
+            }
+            if reader_terminal.is_closing() {
+                continue;
+            }
             let needs_snapshot = reader_publisher
                 .lock()
                 .map(|publisher| publisher.requires_snapshot())
@@ -527,12 +561,16 @@ pub(crate) fn terminal_attach(
         .map_err(|_| "terminal snapshot rejected".to_string())?;
     let session = Session {
         id,
-        terminal,
+        terminal: Arc::clone(&terminal),
         master: Some(pair.master),
         writer,
         pty_size: current_pty_size,
         child,
-        reader: Some(Box::new(ReaderThread(Some(reader_handle)))),
+        reader: Some(Box::new(ReaderThread {
+            handle: Some(reader_handle),
+            terminal: Arc::clone(&terminal),
+            stopping: reader_stopping,
+        })),
         publisher,
         channel,
     };

desktop/src-tauri/src/terminal_transport.rs

@@ -186,6 +186,11 @@ impl FramePublisher {
         }
     }
 
+    /// Permanently detach publication as the session begins closing.
+    pub(crate) fn close(&mut self) {
+        self.subscription = None;
+    }
+
     fn require_current_snapshot(&self, frame: &Frame) -> Result<(), OfferError> {
         if frame.viewport == self.applied && frame.full {
             Ok(())