feat(builtins): stream grep and tr through pipelines

crates/bashkit/src/builtins/cat.rs

@@ -10,7 +10,7 @@ use async_trait::async_trait;
 use std::ffi::OsString;
 use std::path::Path;
 
-use super::{Builtin, Context};
+use super::{Builtin, Context, STREAM_CHUNK_BYTES};
 use crate::error::Result;
 use crate::fs::vfs_join;
 use crate::interpreter::ExecResult;
@@ -121,9 +121,6 @@ impl Builtin for Cat {
     }
 }
 
-/// Bytes per write when streaming into a pipeline (one pipe's capacity).
-const STREAM_CHUNK_BYTES: usize = 4 * 1024;
-
 /// Plain `cat` as a pipeline stage: copy each operand into the stage pipe
 /// as it is read, stdin chunk by chunk, so `loop | cat | head -1` and
 /// `cat big | head -1` stop early. A closed reader ends with 141 (SIGPIPE).

crates/bashkit/src/builtins/cuttr.rs

@@ -344,38 +344,88 @@ impl Builtin for Tr {
             return Ok(Self::err("missing operand", 1));
         }
 
-        let mut set1 = match expand_char_set(&non_flag_args[0]) {
-            Ok(set) => set,
-            Err(msg) => return Ok(Self::err(&msg, 1)),
-        };
+        if let Err(msg) = expand_char_set(&non_flag_args[0]) {
+            return Ok(Self::err(&msg, 1));
+        }
         if !delete && !squeeze && non_flag_args.len() < 2 {
             return Ok(Self::err("missing operand after SET1", 1));
         }
-        let stdin = ctx.stdin.cloned().unwrap_or_default();
-        let byte_mode =
-            ctx.env.get("LC_ALL").is_some_and(|locale| locale == "C") || stdin.text().is_err();
-        if byte_mode && set1.iter().all(|c| (*c as u32) <= u8::MAX as u32) {
-            let set2 = if non_flag_args.len() >= 2 {
-                match expand_char_set(&non_flag_args[1]) {
-                    Ok(set) => Some(set),
+        let spec = TrSpec {
+            args: &non_flag_args,
+            delete,
+            squeeze,
+            complement,
+            c_locale: ctx.env.get("LC_ALL").is_some_and(|locale| locale == "C"),
+        };
+
+        // A pipeline stage translates its input as it arrives, so
+        // `yes | tr y n | head -1` ends with SIGPIPE. Squeezing looks across
+        // piece boundaries, so `-s` reads everything first.
+        if !squeeze && let Some(out) = ctx.stdout_stream() {
+            let mut input = super::InputChunks::stdin(&ctx);
+            let cut = |data: &[u8]| {
+                if spec.c_locale {
+                    data.len()
+                } else {
+                    super::cut_utf8(data)
+                }
+            };
+            while let Some(piece) = input.next(cut).await {
+                ctx.consume_budget_work(1)?;
+                let output = match spec.apply(&piece.into()) {
+                    Ok(output) => output,
                     Err(msg) => return Ok(Self::err(&msg, 1)),
+                };
+                if !output.is_empty() && !out.write(&output).await {
+                    return Ok(ExecResult::with_code(String::new(), 141));
                 }
+            }
+            return Ok(ExecResult::ok(String::new()));
+        }
+
+        let stdin = ctx.stdin_to_end().await.unwrap_or_default();
+        match spec.apply(&stdin) {
+            Ok(output) => Ok(ExecResult::ok_bytes(output)),
+            Err(msg) => Ok(Self::err(&msg, 1)),
+        }
+    }
+}
+
+/// Parsed `tr` operands and flags.
+struct TrSpec<'a> {
+    args: &'a [String],
+    delete: bool,
+    squeeze: bool,
+    complement: bool,
+    /// `LC_ALL=C`: work on bytes, not characters.
+    c_locale: bool,
+}
+
+impl TrSpec<'_> {
+    /// Translate one piece of input; `Err` is a SET diagnostic.
+    fn apply(&self, stdin: &crate::StreamData) -> std::result::Result<Vec<u8>, String> {
+        let (non_flag_args, delete, squeeze, complement) =
+            (self.args, self.delete, self.squeeze, self.complement);
+        let mut set1 = expand_char_set(&non_flag_args[0])?;
+        let byte_mode = self.c_locale || stdin.text().is_err();
+        if byte_mode && set1.iter().all(|c| (*c as u32) <= u8::MAX as u32) {
+            let set2 = if non_flag_args.len() >= 2 {
+                Some(expand_char_set(&non_flag_args[1])?)
             } else {
                 None
             };
             if set2
                 .as_ref()
                 .is_none_or(|set| set.iter().all(|c| (*c as u32) <= u8::MAX as u32))
             {
-                let output = translate_bytes(
+                return Ok(translate_bytes(
                     stdin.as_bytes(),
                     &set1,
                     set2.as_deref(),
                     delete,
                     squeeze,
                     complement,
-                );
-                return Ok(ExecResult::ok_bytes(output));
+                ));
             }
         }
         if complement {
@@ -389,15 +439,12 @@ impl Builtin for Tr {
                 .collect();
         }
 
-        let stdin = &*stdin;
+        let stdin: &str = stdin;
 
         let result = if delete && squeeze {
             // -ds: delete SET1 chars, then squeeze SET2 chars
             let set2 = if non_flag_args.len() >= 2 {
-                match expand_char_set(&non_flag_args[1]) {
-                    Ok(set) => set,
-                    Err(msg) => return Ok(Self::err(&msg, 1)),
-                }
+                expand_char_set(&non_flag_args[1])?
             } else {
                 set1.clone()
             };
@@ -412,14 +459,7 @@ impl Builtin for Tr {
             // -s with only SET1: squeeze characters in SET1
             squeeze_chars(stdin, &set1)
         } else {
-            if non_flag_args.len() < 2 {
-                return Ok(Self::err("missing operand after SET1", 1));
-            }
-
-            let set2 = match expand_char_set(&non_flag_args[1]) {
-                Ok(set) => set,
-                Err(msg) => return Ok(Self::err(&msg, 1)),
-            };
+            let set2 = expand_char_set(&non_flag_args[1])?;
 
             let translated: String = stdin
                 .chars()
@@ -439,7 +479,7 @@ impl Builtin for Tr {
             }
         };
 
-        Ok(ExecResult::ok(result))
+        Ok(result.into_bytes())
     }
 }
 

crates/bashkit/src/builtins/grep.rs

@@ -620,6 +620,23 @@ impl Builtin for Grep {
 
         let matcher = opts.build_matcher()?;
 
+        // A pipeline stage that prints matching lines streams them: lines
+        // go out as they are found, and `loop | grep y | head -1` ends with
+        // SIGPIPE. Options that summarize (-c, -l, -L, -q), look around (-A,
+        // -B, -C), offset (-b) or recurse keep the buffered path.
+        if !opts.recursive
+            && !opts.count_only
+            && !opts.files_with_matches
+            && !opts.files_without_matches
+            && !opts.quiet
+            && !opts.byte_offset
+            && opts.before_context == 0
+            && opts.after_context == 0
+            && let Some(out) = ctx.stdout_stream()
+        {
+            return stream_grep(&ctx, &opts, &matcher, &out).await;
+        }
+
         let mut output = String::new();
         // Diagnostics are a separate stream: a `grep: FILE: ...` line written
         // into `output` lands in the data a pipeline consumes, so
@@ -650,7 +667,11 @@ impl Builtin for Grep {
         let inputs: Vec<(String, String)> = if opts.files.is_empty() {
             // Read from stdin
             // No NUL filtering here either — see `process_content`.
-            let stdin_content = ctx.stdin.map(ToString::to_string).unwrap_or_default();
+            let stdin_content = ctx
+                .stdin_to_end()
+                .await
+                .map(|s| s.to_string())
+                .unwrap_or_default();
             vec![(stdin_name.to_string(), stdin_content)]
         } else if opts.recursive {
             // Try indexed search via SearchCapable if available. Skip it for -P:
@@ -1075,6 +1096,139 @@ impl Builtin for Grep {
     }
 }
 
+/// `grep` as a pipeline stage writing straight into the stage pipe: each
+/// input is read in record-sized pieces and matching lines are written as
+/// they are found. Same output as the buffered path for the options it
+/// takes; a reader that went away ends it with 141 (SIGPIPE), and `-m`
+/// stops reading as soon as the budget is spent.
+async fn stream_grep(
+    ctx: &Context<'_>,
+    opts: &GrepOptions,
+    matcher: &Matcher,
+    out: &super::StdoutStream,
+) -> Result<ExecResult> {
+    let term = if opts.null_terminated { b'\0' } else { b'\n' };
+    let fname_sep = if opts.null_filename { '\0' } else { ':' };
+    let show_filename = !opts.no_filename && (opts.show_filename || opts.files.len() > 1);
+    let mut errors = String::new();
+    let mut read_failed = false;
+    let mut any_match = false;
+    let sources: Vec<Option<&String>> = if opts.files.is_empty() {
+        vec![None]
+    } else {
+        opts.files.iter().map(Some).collect()
+    };
+    for source in sources {
+        let (name, mut input) = match source {
+            None => ("(standard input)", super::InputChunks::stdin(ctx)),
+            Some(file) => {
+                let path = if file.starts_with('/') {
+                    std::path::PathBuf::from(file)
+                } else {
+                    vfs_join(ctx.cwd, file)
+                };
+                match ctx.fs.read_file(&path).await {
+                    Ok(bytes) => (file.as_str(), super::InputChunks::bytes(bytes)),
+                    Err(e) => {
+                        read_failed = true;
+                        if !opts.suppress_errors {
+                            let reason = crate::error::io_error_reason(&e);
+                            errors.push_str(&format!("grep: {file}: {reason}\n"));
+                        }
+                        continue;
+                    }
+                }
+            }
+        };
+        let mut line_no = 0usize;
+        let mut match_count = 0usize;
+        let mut binary = false;
+        let budget_spent = |count: usize| opts.max_count.is_some_and(|max| count >= max);
+        'file: while let Some(piece) = input.next(super::cut_records(term)).await {
+            ctx.consume_budget_work(1)?;
+            // Binary input (a NUL without -a/-z): matches are reported once
+            // on stderr instead of printed; -I makes it a non-match.
+            if !binary && !opts.binary_as_text && !opts.null_terminated && piece.contains(&0) {
+                binary = true;
+                if opts.skip_binary {
+                    break 'file;
+                }
+            }
+            let mut output = Vec::new();
+            let mut stop = false;
+            for record in piece.split_inclusive(|&b| b == term) {
+                let record = record.strip_suffix(&[term]).unwrap_or(record);
+                let text = String::from_utf8_lossy(record);
+                // Same line text as `str::lines` in the buffered path.
+                let line = if term == b'\n' {
+                    text.strip_suffix('\r').unwrap_or(&text)
+                } else {
+                    &text
+                };
+                line_no += 1;
+                let prefix = |output: &mut Vec<u8>| {
+                    if show_filename {
+                        output.extend_from_slice(name.as_bytes());
+                        output.push(fname_sep as u8);
+                    }
+                    if opts.line_numbers {
+                        output.extend_from_slice(format!("{line_no}:").as_bytes());
+                    }
+                };
+                if opts.only_matching && !opts.invert_match {
+                    matcher.for_each_range(line, |(start, end)| {
+                        if budget_spent(match_count) {
+                            return false;
+                        }
+                        any_match = true;
+                        match_count += 1;
+                        if binary {
+                            stop = true;
+                            return false;
+                        }
+                        prefix(&mut output);
+                        output.extend_from_slice(&line.as_bytes()[start..end]);
+                        output.push(term);
+                        true
+                    });
+                } else if matcher.is_match(line) != opts.invert_match {
+                    any_match = true;
+                    match_count += 1;
+                    if binary {
+                        stop = true;
+                    } else {
+                        prefix(&mut output);
+                        output.extend_from_slice(line.as_bytes());
+                        output.push(term);
+                    }
+                }
+                if stop || budget_spent(match_count) {
+                    stop = true;
+                    break;
+                }
+            }
+            if !output.is_empty() && !out.write(&output).await {
+                return Ok(ExecResult::err(errors, 141));
+            }
+            if stop {
+                if binary && match_count > 0 {
+                    // GNU grep >= 3.5: a diagnostic, never pipeline data.
+                    errors.push_str(&format!("grep: {name}: binary file matches\n"));
+                }
+                break 'file;
+            }
+        }
+    }
+    let code = if read_failed {
+        2
+    } else if any_match {
+        0
+    } else {
+        1
+    };
+    Ok(ExecResult::err(errors, code))
+}
+
 /// Name a file found under a recursive-grep operand the way GNU grep does:
 /// the operand as typed, then the path below it (`d/` + `b` -> `d/b`,
 /// `.` + `b` -> `./b`). An empty operand is the implicit `-r` root and