fix: wake the command runner when output arrives instead of after a fixed nap

crates/codexbar-providers/src/command.rs

@@ -179,6 +179,10 @@ impl<P: pipe::Pipe> Capture<P> {
         })
     }
 
+    fn open_pipe(&self) -> Option<&P> {
+        (!self.closed).then_some(&self.pipe)
+    }
+
     /// At most one chunk per stream per turn, so a noisy stream cannot starve the other stream or the deadline.
     fn drain(&mut self) -> Result<bool, CommandError> {
         if self.closed {
@@ -255,7 +259,11 @@ fn collect(
             return Ok((stdout.bytes, stderr.bytes));
         }
         if !progress {
-            std::thread::sleep(CHECK_INTERVAL.min(timeout.saturating_sub(start.elapsed())));
+            pipe::wait(
+                stdout.open_pipe(),
+                stderr.open_pipe(),
+                CHECK_INTERVAL.min(timeout.saturating_sub(start.elapsed())),
+            );
         }
     }
 }
@@ -272,6 +280,59 @@ mod tests {
         );
     }
 
+    /// Every drain outcome on a real pipe, independent of how a command's writes and EOFs happen to interleave.
+    #[cfg(unix)]
+    #[test]
+    fn a_drain_reports_pending_output_eof_a_closed_stream_and_overflow() {
+        use std::io::Write as _;
+        let (reader, mut writer) = std::io::pipe().unwrap();
+        let mut capture = Capture::new(reader, 4).unwrap();
+        assert_eq!(capture.drain(), Ok(false), "nothing written yet");
+        writer.write_all(b"abc").unwrap();
+        assert_eq!(capture.drain(), Ok(true));
+        assert_eq!(capture.bytes, b"abc");
+        writer.write_all(b"de").unwrap();
+        assert_eq!(capture.drain(), Err(CommandError::Failed), "past the 4-byte limit");
+        let (reader, writer) = std::io::pipe().unwrap();
+        let mut capture = Capture::new(reader, 4).unwrap();
+        drop(writer);
+        assert_eq!(capture.drain(), Ok(true), "EOF is progress");
+        assert!(capture.closed);
+        assert_eq!(capture.drain(), Ok(false), "a closed stream is never read again");
+    }
+
+    /// Every way a collection turn ends or goes on, independent of which pipe a real command closes first.
+    #[cfg(unix)]
+    #[test]
+    fn collection_ends_at_the_stop_marker_or_once_both_streams_closed_and_the_command_exited() {
+        let mut child = std::process::Command::new("true").spawn().unwrap();
+        let (reader, _writer) = std::io::pipe().unwrap();
+        let mut stderr = Capture {
+            pipe: reader,
+            bytes: b"Error: authentication required.".to_vec(),
+            closed: false,
+            limit: OUTPUT_LIMIT,
+        };
+        let stop = |marker| StopMarker { marker, searched: 0 };
+        assert_eq!(
+            complete(&mut child, false, &stderr, &mut stop(b"authentication required")),
+            Ok(true)
+        );
+        assert_eq!(complete(&mut child, false, &stderr, &mut stop(b"")), Ok(false));
+        assert_eq!(
+            complete(&mut child, true, &stderr, &mut stop(b"")),
+            Ok(false),
+            "stderr still open"
+        );
+        stderr.closed = true;
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while complete(&mut child, true, &stderr, &mut stop(b"")) != Ok(true) {
+            assert!(Instant::now() < deadline, "the exited command ends collection");
+            std::thread::sleep(Duration::from_millis(5));
+        }
+        child.wait().unwrap();
+    }
+
     #[test]
     fn a_stop_marker_is_found_across_reads_and_case() {
         let text = b"Authentication required. Please visit the URL";

crates/codexbar-providers/src/command/pipe.rs

@@ -57,6 +57,42 @@ pub(super) fn read(pipe: &mut impl Pipe, bytes: &mut [u8]) -> io::Result<Option<
     read_result(pipe.read(&mut bytes[..count]))
 }
 
+/// Waits up to `timeout` for either open pipe to have output or close (#281). Unix wakes as soon as one does, so a
+/// writer that refills its pipe faster than a fixed nap is read at its own pace; macOS's small pipe buffers otherwise
+/// turned a 4 MiB stream into hundreds of naps. With both pipes closed, or on Windows, whose anonymous pipes can't be
+/// polled, it naps for `timeout`. A signal or poll error only ends the wait early; the caller checks again.
+#[cfg(unix)]
+pub(super) fn wait(first: Option<&impl Pipe>, second: Option<&impl Pipe>, timeout: std::time::Duration) {
+    poll_or_nap([first.map(|p| p.as_raw_fd()), second.map(|p| p.as_raw_fd())], timeout);
+}
+
+/// `wait` for raw descriptors: one implementation however many pipe types call it.
+#[cfg(unix)]
+fn poll_or_nap(fds: [Option<std::os::fd::RawFd>; 2], timeout: std::time::Duration) {
+    let mut fds: Vec<libc::pollfd> = fds
+        .into_iter()
+        .flatten()
+        .map(|fd| libc::pollfd {
+            fd,
+            events: libc::POLLIN,
+            revents: 0,
+        })
+        .collect();
+    if fds.is_empty() {
+        std::thread::sleep(timeout);
+        return;
+    }
+    // Rounded up, so a sub-millisecond remainder still waits instead of spinning.
+    let ms = timeout.as_micros().div_ceil(1000).min(libc::c_int::MAX as u128) as libc::c_int;
+    // SAFETY: `fds` is valid writable storage for its length, and each descriptor is an open pipe this thread owns.
+    let _ = unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as libc::nfds_t, ms) };
+}
+
+#[cfg(windows)]
+pub(super) fn wait(_: Option<&impl Pipe>, _: Option<&impl Pipe>, timeout: std::time::Duration) {
+    std::thread::sleep(timeout);
+}
+
 fn read_result(result: io::Result<usize>) -> io::Result<Option<usize>> {
     match result {
         Ok(count) => Ok(Some(count)),
@@ -162,6 +198,39 @@ fn group_has_live_members(id: u32, _: std::time::Instant) -> io::Result<bool> {
 mod tests {
     use super::*;
 
+    #[cfg(unix)]
+    #[test]
+    fn an_idle_wait_ends_when_output_arrives_not_after_its_timeout() {
+        use std::io::Write as _;
+        use std::time::{Duration, Instant};
+        let (reader, mut writer) = io::pipe().unwrap();
+        let none: Option<&io::PipeReader> = None;
+        let start = Instant::now();
+        wait(Some(&reader), none, Duration::from_millis(30));
+        assert!(
+            start.elapsed() >= Duration::from_millis(25),
+            "no output: the whole wait"
+        );
+        let late = std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(20));
+            writer.write_all(b"x").unwrap();
+            writer
+        });
+        let start = Instant::now();
+        wait(Some(&reader), none, Duration::from_secs(10));
+        assert!(start.elapsed() < Duration::from_secs(5), "output ends the wait");
+        drop(late.join().unwrap());
+        let start = Instant::now();
+        wait(Some(&reader), none, Duration::from_secs(10));
+        assert!(
+            start.elapsed() < Duration::from_secs(5),
+            "readable data or a closed pipe ends it at once"
+        );
+        let start = Instant::now();
+        wait(none, none, Duration::from_millis(30));
+        assert!(start.elapsed() >= Duration::from_millis(25), "no open pipe: a nap");
+    }
+
     #[test]
     fn nonblocking_reads_distinguish_pending_eof_data_and_failure() {
         assert_eq!(read_result(Ok(0)).unwrap(), Some(0));

docs/COMMAND_RUNNER.md

@@ -49,7 +49,11 @@ uses agy's local browser flow.
   the previous replacement behavior for invalid UTF-8. Nonzero exit codes return
   `CommandOutput { success: false, .. }` with the collected output.
 - The supplied timeout starts after successful process creation and covers both
-  pipe EOF and process exit. Idle checks sleep at most 25 ms. A timeout returns
+  pipe EOF and process exit. An idle check waits at most 25 ms: on Unix it
+  `poll`s the open pipes and returns as soon as one has output or closes, so a
+  writer that refills a small pipe (16 KiB on macOS) isn't read one fixed nap at
+  a time (#281); on Windows, whose anonymous pipes can't be polled, and once
+  both pipes have closed, it sleeps. A timeout returns
   `TimedOut` after cleanup; cleanup failure returns `Failed` instead of claiming
   that termination completed. Process creation itself is a native system call.
 - Verify owned tree termination for up to one additional second. Stop and reap