fix: stream preview command instead of waiting for completion (#1176)

closes #1174

ARCHITECTURE.md

@@ -870,7 +870,7 @@ Pre-selection is applied when items first appear: `DefaultSkimSelector::should_s
 
 `Preview` (`src/tui/preview.rs`) renders a side/top/bottom pane showing expanded information about the focused item. Its stored content is one of three variants:
 
-**Plain text mode** (no `pty`): spawns `sh -c <cmd>` on Unix or `cmd /c <cmd>` on Windows. On Windows, `Command::raw_arg` is used so `cmd.exe` receives shell metacharacters exactly as written. The worker drains stdout and stderr concurrently, but retains at most `PREVIEW_MAX_BYTES` from each stream. Cancellation terminates the child process group (the process tree on Windows), so selection changes do not leave old preview commands running. Successful stdout or failed stderr is parsed with `ansi_to_tui::IntoText`, stored as `PreviewContent::Text`, and followed by `Event::PreviewReady`.
+**Plain text mode** (no `pty`): spawns `sh -c <cmd>` on Unix or `cmd /c <cmd>` on Windows. On Windows, `Command::raw_arg` is used so `cmd.exe` receives shell metacharacters exactly as written. The worker drains stdout and stderr concurrently, but retains at most `PREVIEW_MAX_BYTES` from each stream. Retained stdout is parsed with `ansi_to_tui::IntoText` and published while the command runs. Cancellation terminates the child process group (the process tree on Windows) and invalidates its output writer, so an old reader cannot replace content from a newer preview. At exit, successful stdout or failed stderr is stored as `PreviewContent::Text` and followed by `Event::PreviewReady`.
 
 **PTY mode** (`--preview-window pty`): creates a real pseudo-terminal pair via `portable_pty`. The child process sees a properly sized terminal (via `ROWS`/`COLUMNS` env and PTY dimensions). Output is parsed by a `vt100::Parser` with a scrollback buffer, stored as `PreviewContent::Terminal(Arc<RwLock<vt100::Parser>>)`. This enables interactive preview programs (e.g. `bat`, `delta`).
 
@@ -896,8 +896,8 @@ else if pty mode:
 
 else:
   start shell in a dedicated process group with piped stdout + stderr
-  thread: drain both streams with bounded retention; poll child status
+  thread: drain both streams with bounded retention; stream active stdout; poll child status
-          → cancellation kills the process group
+          → cancellation invalidates the writer and kills the process group
           → content.write() = PreviewContent::Text(…)
           → Event::PreviewReady
 ```

src/tui/preview.rs

@@ -16,6 +16,7 @@ use tui_term::widget::PseudoTerminal;
 use std::env;
 use std::io::Read;
 use std::process::{Child, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock, mpsc};
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
@@ -35,20 +36,49 @@ const VT_SCROLLBACK: usize = 100_000;
 type PlainChild = Arc<Mutex<Option<Child>>>;
 
 fn read_bounded(mut reader: impl Read) -> Vec<u8> {
+    read_bounded_with_updates(&mut reader, |_| {})
+}
+
+fn read_bounded_with_updates(mut reader: impl Read, mut update: impl FnMut(&[u8])) -> Vec<u8> {
+    const UPDATE_INTERVAL: Duration = Duration::from_millis(16);
+
     let mut output = Vec::with_capacity(PREVIEW_MAX_BYTES);
     let mut buffer = [0; 8192];
+    let mut last_update = None;
+    let mut published_len = 0;
     loop {
         match reader.read(&mut buffer) {
             Ok(0) | Err(_) => break,
             Ok(read) => {
                 let retained = PREVIEW_MAX_BYTES.saturating_sub(output.len()).min(read);
                 output.extend_from_slice(&buffer[..retained]);
+
+                let update_due = last_update.is_none_or(|last: Instant| last.elapsed() >= UPDATE_INTERVAL);
+                if retained > 0 && (update_due || output.len() == PREVIEW_MAX_BYTES) {
+                    update(&output);
+                    published_len = output.len();
+                    last_update = Some(Instant::now());
+                }
             }
         }
     }
+    if output.len() != published_len {
+        update(&output);
+    }
     output
 }
 
+fn update_plain_content(content: &RwLock<PreviewContent>, cancelled: &AtomicBool, output: &[u8]) {
+    let Ok(text) = output.to_vec().into_text() else {
+        return;
+    };
+    if let Ok(mut content) = content.write()
+        && !cancelled.load(Ordering::Acquire)
+    {
+        *content = PreviewContent::Text(text);
+    }
+}
+
 fn terminate_plain_child(child: &PlainChild) {
     let Ok(mut guard) = child.lock() else {
         return;
@@ -132,6 +162,7 @@ pub struct Preview {
     /// Channel to signal thread interruption
     interrupt_tx: Option<mpsc::Sender<()>>,
     plain_child: Option<PlainChild>,
+    plain_cancelled: Option<Arc<AtomicBool>>,
     pub theme: Arc<ColorTheme>,
     /// Border type
     pub border: BorderType,
@@ -338,6 +369,10 @@ impl Preview {
     }
     /// Kill the preview child process and interrupt the reader thread.
     pub fn kill(&mut self) {
+        if let Some(cancelled) = self.plain_cancelled.take() {
+            cancelled.store(true, Ordering::Release);
+        }
+
         if let Some(tx) = self.interrupt_tx.take() {
             let _ = tx.send(());
         }
@@ -555,6 +590,8 @@ impl Preview {
 
             let (interrupt_tx, interrupt_rx) = mpsc::channel();
             self.interrupt_tx = Some(interrupt_tx);
+            let cancelled = Arc::new(AtomicBool::new(false));
+            self.plain_cancelled = Some(cancelled.clone());
 
             let mut child = match shell_cmd.spawn() {
                 Ok(child) => child,
@@ -570,7 +607,13 @@ impl Preview {
             self.plain_child = Some(child.clone());
 
             self.thread_handle = Some(std::thread::spawn(move || {
-                let stdout_reader = std::thread::spawn(move || read_bounded(stdout));
+                let streaming_content = content.clone();
+                let streaming_cancelled = cancelled.clone();
+                let stdout_reader = std::thread::spawn(move || {
+                    read_bounded_with_updates(stdout, |output| {
+                        update_plain_content(&streaming_content, &streaming_cancelled, output);
+                    })
+                });
                 let stderr_reader = std::thread::spawn(move || read_bounded(stderr));
 
                 let status = loop {
@@ -612,13 +655,17 @@ impl Preview {
                 let Some(status) = status else {
                     return;
                 };
-                if let Ok(mut c) = content.write() {
+                if let Ok(mut c) = content.write()
+                    && !cancelled.load(Ordering::Acquire)
+                {
                     let output = if status.success() { stdout } else { stderr };
                     *c = PreviewContent::Text(output.into_text().unwrap_or_default());
                 }
 
-                trace!("sending ready ping");
+                if !cancelled.load(Ordering::Acquire) {
-                let _ = event_tx_clone.blocking_send(Event::PreviewReady);
+                    trace!("sending ready ping");
+                    let _ = event_tx_clone.blocking_send(Event::PreviewReady);
+                }
             }));
         }
         Ok(())
@@ -767,6 +814,7 @@ impl SkimWidget for Preview {
             thread_handle: None,
             interrupt_tx: None,
             plain_child: None,
+            plain_cancelled: None,
             pty: None,
             pty_child: None,
             #[cfg(feature = "image")]

src/tui/preview_tests.rs

@@ -5,7 +5,7 @@ use ratatui::layout::Size;
 #[cfg(feature = "image")]
 use ratatui_image::picker::Picker;
 
-use super::{PREVIEW_MAX_BYTES, Preview, PreviewContent, read_bounded};
+use super::{PREVIEW_MAX_BYTES, Preview, PreviewContent, read_bounded, update_plain_content};
 
 #[cfg(feature = "image")]
 fn image(width: u32, height: u32) -> DynamicImage {
@@ -86,12 +86,91 @@ fn bounded_reader_discards_output_after_limit() {
     assert_eq!(output.len(), PREVIEW_MAX_BYTES);
 }
 
+fn preview_contains(preview: &Preview, expected: &str) -> bool {
+    preview.content.read().is_ok_and(|content| match &*content {
+        PreviewContent::Text(text) => text
+            .lines
+            .iter()
+            .any(|line| line.spans.iter().any(|span| span.content.as_ref().contains(expected))),
+        _ => false,
+    })
+}
+
+#[cfg(unix)]
+#[test]
+fn plain_preview_streams_before_command_exits() {
+    use std::time::{Duration, Instant};
+
+    use ratatui::backend::TestBackend;
+
+    let mut preview = Preview::default();
+    preview.pty = None;
+    let mut tui =
+        super::super::Tui::new_with_height_and_backend(TestBackend::new(20, 5), super::super::Size::Percent(100))
+            .unwrap();
+    preview.spawn(&mut tui, "printf streamed; sleep 30").unwrap();
+
+    let started = Instant::now();
+    let streamed_in_time = loop {
+        if preview_contains(&preview, "streamed") {
+            break true;
+        }
+        if started.elapsed() >= Duration::from_secs(2) {
+            break false;
+        }
+        std::thread::sleep(Duration::from_millis(10));
+    };
+
+    preview.kill();
+    preview.thread_handle.take().unwrap().join().unwrap();
+    assert!(streamed_in_time, "preview output did not stream");
+}
+
+#[cfg(unix)]
+#[test]
+fn stale_plain_preview_cannot_replace_newer_streamed_output() {
+    use std::time::{Duration, Instant};
+
+    use ratatui::backend::TestBackend;
+
+    let mut preview = Preview::default();
+    preview.pty = None;
+    let mut tui =
+        super::super::Tui::new_with_height_and_backend(TestBackend::new(20, 5), super::super::Size::Percent(100))
+            .unwrap();
+
+    preview.spawn(&mut tui, "printf stale; sleep 30").unwrap();
+    let stale_cancelled = preview.plain_cancelled.as_ref().unwrap().clone();
+    let stale_thread = preview.thread_handle.take().unwrap();
+    preview.spawn(&mut tui, "printf current; sleep 30").unwrap();
+
+    let started = Instant::now();
+    let current_streamed = loop {
+        if preview_contains(&preview, "current") {
+            break true;
+        }
+        if started.elapsed() >= Duration::from_secs(2) {
+            break false;
+        }
+        std::thread::sleep(Duration::from_millis(10));
+    };
+    update_plain_content(&preview.content, &stale_cancelled, b"stale");
+    let stale_write_was_ignored = preview_contains(&preview, "current") && !preview_contains(&preview, "stale");
+
+    preview.kill();
+    preview.thread_handle.take().unwrap().join().unwrap();
+    stale_thread.join().unwrap();
+    assert!(current_streamed, "new preview output did not stream");
+    assert!(stale_write_was_ignored, "stale preview replaced newer output");
+}
+
 #[cfg(unix)]
 #[test]
 fn plain_preview_can_be_cancelled() {
-    use ratatui::backend::TestBackend;
     use std::time::{Duration, Instant};
 
+    use ratatui::backend::TestBackend;
+
     let mut preview = Preview::default();
     preview.pty = None;
     let mut tui =