feat(interpreter): stream cat through pipelines

crates/bashkit/src/builtins/cat.rs

@@ -63,12 +63,26 @@ impl Builtin for Cat {
             .map(|vs| vs.map(|v| v.to_string_lossy().into_owned()).collect())
             .unwrap_or_default();
 
+        let plain = !(show_ends
+            || show_tabs
+            || show_nonprinting
+            || number_all
+            || number_nonblank
+            || squeeze);
+        if plain && let Some(stream) = ctx.stdout_stream() {
+            return stream_plain(&ctx, &files, &stream).await;
+        }
+
         let mut raw = Vec::new();
+        let mut stderr = String::new();
         for file in &files {
             if file == "-" {
                 if let Some(stdin) = ctx.stdin {
                     raw.extend_from_slice(stdin.as_bytes());
                 }
+                if let Some(input) = ctx.stdin_stream() {
+                    raw.extend_from_slice(&input.read_to_end().await);
+                }
             } else {
                 let path = if Path::new(file).is_absolute() {
                     file.clone()
@@ -77,30 +91,95 @@ impl Builtin for Cat {
                 };
                 match ctx.fs.read_file(Path::new(&path)).await {
                     Ok(bytes) => raw.extend_from_slice(&bytes),
+                    // GNU cat reports the operand and goes on with the rest.
                     Err(e) => {
                         let reason = crate::error::io_error_reason(&e);
-                        return Ok(ExecResult::err(format!("cat: {file}: {reason}\n"), 1));
+                        stderr.push_str(&format!("cat: {file}: {reason}\n"));
                     }
                 }
             }
         }
 
-        if !(show_ends || show_tabs || show_nonprinting || number_all || number_nonblank || squeeze)
-        {
-            return Ok(ExecResult::ok_bytes(raw));
+        let mut result = if plain {
+            ExecResult::ok_bytes(raw)
+        } else {
+            ExecResult::ok_bytes(render(
+                &raw,
+                show_ends,
+                show_tabs,
+                show_nonprinting,
+                number_all,
+                number_nonblank,
+                squeeze,
+            ))
+        };
+        if !stderr.is_empty() {
+            result.stderr = stderr.into();
+            result.exit_code = 1;
         }
+        Ok(result)
+    }
+}
+
+/// Bytes per write when streaming into a pipeline (one pipe's capacity).
+const STREAM_CHUNK_BYTES: usize = 4 * 1024;
 
-        let output = render(
-            &raw,
-            show_ends,
-            show_tabs,
-            show_nonprinting,
-            number_all,
-            number_nonblank,
-            squeeze,
-        );
-        Ok(ExecResult::ok(output))
+/// 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).
+async fn stream_plain(
+    ctx: &Context<'_>,
+    files: &[String],
+    stream: &super::StdoutStream,
+) -> Result<ExecResult> {
+    async fn send(ctx: &Context<'_>, stream: &super::StdoutStream, data: &[u8]) -> Result<bool> {
+        for chunk in data.chunks(STREAM_CHUNK_BYTES) {
+            ctx.consume_budget_work(1)?;
+            if !stream.write(chunk).await {
+                return Ok(false);
+            }
+        }
+        Ok(true)
+    }
+    let mut stderr = String::new();
+    let mut code = 0;
+    for file in files {
+        let sent = if file == "-" {
+            let mut ok = match ctx.stdin {
+                Some(stdin) => send(ctx, stream, stdin.as_bytes()).await?,
+                None => true,
+            };
+            if let Some(input) = ctx.stdin_stream() {
+                while ok {
+                    let chunk = input.read().await;
+                    if chunk.is_empty() {
+                        break;
+                    }
+                    ok = send(ctx, stream, &chunk).await?;
+                }
+            }
+            ok
+        } else {
+            let path = if Path::new(file).is_absolute() {
+                file.clone()
+            } else {
+                vfs_join(ctx.cwd, file).to_string_lossy().into_owned()
+            };
+            match ctx.fs.read_file(Path::new(&path)).await {
+                Ok(bytes) => send(ctx, stream, &bytes).await?,
+                Err(e) => {
+                    let reason = crate::error::io_error_reason(&e);
+                    stderr.push_str(&format!("cat: {file}: {reason}\n"));
+                    code = 1;
+                    true
+                }
+            }
+        };
+        if !sent {
+            return Ok(ExecResult::err(stderr, 141));
+        }
     }
+    Ok(ExecResult::err(stderr, code))
 }
 
 /// Apply cat's display transforms in a single pass.

crates/bashkit/src/builtins/mod.rs

@@ -952,6 +952,21 @@ pub struct Context<'a> {
     pub(crate) shell: Option<ShellRef<'a>>,
 }
 
+/// Stdin pipe of a streaming filter stage, see [`Context::stdin_stream`].
+pub(crate) struct StdinStream(std::sync::Arc<crate::interpreter::pipe::Pipe>);
+
+impl StdinStream {
+    /// Next chunk of input; empty at end of input.
+    pub(crate) async fn read(&self) -> Vec<u8> {
+        self.0.read_some().await
+    }
+
+    /// Everything left, for filters that need their whole input.
+    pub(crate) async fn read_to_end(&self) -> Vec<u8> {
+        self.0.read_to_end().await
+    }
+}
+
 /// Stdout pipe of a streaming pipeline stage, see [`Context::stdout_stream`].
 pub(crate) struct StdoutStream(std::sync::Arc<crate::interpreter::pipe::Pipe>);
 
@@ -1056,6 +1071,16 @@ impl<'a> Context<'a> {
             .map(StdoutStream)
     }
 
+    /// Streaming stdin when this builtin is a filter stage of a concurrent
+    /// pipeline (`loop | cat | head -1`). `ctx.stdin` then holds only data
+    /// already buffered before the stage started; read the rest here.
+    pub(crate) fn stdin_stream(&self) -> Option<StdinStream> {
+        self.shell
+            .as_ref()
+            .and_then(|shell| shell.stdin_pipe.clone())
+            .map(StdinStream)
+    }
+
     /// Remaining wall-clock budget of the current `exec*` call, if limited.
     pub fn remaining_deadline(&self) -> Option<std::time::Duration> {
         self.execution_extension::<ExecutionDeadline>()?

crates/bashkit/src/interpreter/mod.rs

@@ -368,6 +368,8 @@ pub(crate) struct ShellRef<'a> {
     pub(crate) execution_extensions: Arc<builtins::ExecutionExtensions>,
     /// Stdout of a streaming pipeline stage (see `Context::stdout_stream`).
     pub(crate) stdout_pipe: Option<Arc<pipe::Pipe>>,
+    /// Stdin of a streaming filter stage (see `Context::stdin_stream`).
+    pub(crate) stdin_pipe: Option<Arc<pipe::Pipe>>,
 }
 
 // Interpreter-dispatched "special" builtins, listed here so the public
@@ -1421,6 +1423,8 @@ pub struct Interpreter {
     stream_stdout_command: Option<usize>,
     /// Pipe handed to the next builtin's `Context` (taken at dispatch).
     builtin_stdout_pipe: Option<Arc<pipe::Pipe>>,
+    /// Input pipe handed to a streaming filter's `Context` (taken at dispatch).
+    builtin_stdin_pipe: Option<Arc<pipe::Pipe>>,
     /// Position within the current argument while `getopts` walks a clustered
     /// short-option group (e.g. `-abc`). Interpreter-internal working state for
     /// `execute_getopts`; `0` means "at the start of the next option group".
@@ -1997,6 +2001,7 @@ impl Interpreter {
             pipe_out: None,
             stream_stdout_command: None,
             builtin_stdout_pipe: None,
+            builtin_stdin_pipe: None,
             getopts_char_idx: 0,
             last_bg_pid: None,
             output_callback: None,
@@ -2154,6 +2159,7 @@ impl Interpreter {
             pipe_out: None,
             stream_stdout_command: None,
             builtin_stdout_pipe: None,
+            builtin_stdin_pipe: None,
             getopts_char_idx: self.getopts_char_idx,
             last_bg_pid: self.last_bg_pid.clone(),
             output_callback: None,
@@ -5856,10 +5862,16 @@ impl Interpreter {
         simple.redirects.is_empty()
             && matches!(
                 simple.name.parts.as_slice(),
-                [WordPart::Literal(name)] if matches!(name.as_str(), "yes" | "seq")
+                [WordPart::Literal(name)] if matches!(name.as_str(), "yes" | "seq" | "cat")
             )
     }
 
+    /// Streaming stages that also read their stdin pipe incrementally
+    /// (`Context::stdin_stream`), so `loop | cat | head -1` stops early.
+    fn streams_stdin(name: &str) -> bool {
+        name == "cat"
+    }
+
     /// Run `commands` (the tail of a pipeline) concurrently: every stage but
     /// the last on a forked shell writing into a bounded [`pipe::Pipe`], the
     /// last one here (a subshell unless `lastpipe`) so its output streams.
@@ -7001,6 +7013,14 @@ impl Interpreter {
             } else {
                 None
             };
+            // A streaming filter (`cat`) reads its input pipe as it goes
+            // instead of having it collected up front.
+            self.builtin_stdin_pipe =
+                if self.builtin_stdout_pipe.is_some() && Self::streams_stdin(name) {
+                    self.pipe_in.take()
+                } else {
+                    None
+                };
             // Track $_ (last argument of previous command, from already-expanded args)
             if let Some(last) = args.last() {
                 self.insert_variable_checked("_".to_string(), last.clone());
@@ -7405,6 +7425,7 @@ impl Interpreter {
                     jobs: &self.jobs,
                     execution_extensions,
                     stdout_pipe: None,
+                    stdin_pipe: None,
                 };
                 let plan_ctx = builtins::Context {
                     args,
@@ -7471,6 +7492,7 @@ impl Interpreter {
                 jobs: &self.jobs,
                 execution_extensions,
                 stdout_pipe: self.builtin_stdout_pipe.take(),
+                stdin_pipe: self.builtin_stdin_pipe.take(),
             };
             let ctx = builtins::Context {
                 args,

crates/bashkit/src/interpreter/pipe.rs

@@ -164,7 +164,6 @@ impl Pipe {
     }
 
     /// Read everything until the writer closes.
-    #[cfg(test)]
     pub(crate) async fn read_to_end(&self) -> Vec<u8> {
         let mut out = Vec::new();
         loop {

crates/bashkit/tests/integration/limitations_evidence_tests.rs

@@ -172,18 +172,34 @@ async fn l_term_002_cat_does_not_wait_for_terminal_input() {
 }
 
 /// L-PIPE-001: a leading single-builtin stage (other than the streaming
-/// generators `yes`/`seq`) runs to completion before `head` reads, and never
+/// `yes`/`seq`/`cat`) runs to completion before `head` reads, and never
 /// sees SIGPIPE: bash reports `141 0` here.
 #[tokio::test]
 async fn l_pipe_001_stages_run_sequentially() {
     let mut bash = Bash::new();
     let result = bash
-        .exec("seq 20000 > /tmp/big; cat /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"")
+        .exec("seq 20000 > /tmp/big; grep . /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"")
         .await
         .unwrap();
     assert_eq!(result.stdout, "1\n0 0\n");
 }
 
+/// Plain `cat` streams as a stage, so it gets SIGPIPE like bash, and a
+/// `cat` between an endless loop and `head` no longer runs to a limit.
+#[tokio::test]
+async fn l_pipe_001_cat_streams() {
+    let mut bash = Bash::new();
+    let result = bash
+        .exec(
+            "seq 20000 > /tmp/big; cat /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"\n\
+             while :; do echo y; done | cat | head -2; echo \"${PIPESTATUS[*]}\"",
+        )
+        .await
+        .unwrap();
+    assert_eq!(result.stdout, "1\n141 0\ny\ny\n141 141 0\n");
+    assert!(result.stderr.is_empty(), "{}", result.stderr);
+}
+
 /// The generators stream: `yes | head -1` stops with SIGPIPE, as in bash.
 #[tokio::test]
 async fn l_pipe_001_generators_stream() {

knowledge/foundations/parallel-execution.md

@@ -64,6 +64,11 @@ Implemented
   matched by address, never to a command run from its arguments), so
   `seq 1000000 | head -1` stops after one chunk with `PIPESTATUS` `141 0`.
   Their output caps still apply when the reader takes everything.
+- Plain `cat` (no display flags) streams the same way and is also a
+  streaming filter: the interpreter hands it its input pipe through
+  `Context::stdin_stream` instead of collecting stdin up front, so
+  `while :; do echo; done | cat | head -1` stops with SIGPIPE. `cat` with
+  flags drains the input pipe and runs buffered.
 - From the first stage that runs shell code (loop, group, function, `eval`,
   nested shell), the rest run concurrently (`execute_streaming_stages`, on
   with `concurrent_jobs`). Each non-last stage is a forked interpreter like a

knowledge/log.md

@@ -4,6 +4,7 @@
 
 * **Feature**: `yes` and `seq` stream into pipelines. As a producer stage they write chunks into the pipe as they go (`Context::stdout_stream`), so `yes | head -1` and `seq 1000000 | head -1` stop after the first chunk with `PIPESTATUS` `141 0` instead of running to the output cap and printing a truncation notice. Step 2 of the roadmap (streaming builtin I/O), starting with the generators.
 * **Feature**: `sudo` and `busybox`, the top missing commands in the gap replay (`sudo tee /etc/x <<EOF`, `busybox tar --help`). `sudo` skips its options and runs the command (leading `VAR=VALUE` through `env`, `-s`/`-i` through `bash -c`; `-v`/`-k` succeed, `-l` lists ALL); `busybox APPLET` runs the builtin of that name, `--list` lists them (L-SUDO-001). `nohup`/`nice`/`flock` no longer steal the inner command's `--help`.
+* **Feature**: Plain `cat` streams through pipelines. As a stage it writes 4 KiB chunks into the stage pipe and, in a middle stage, reads its input pipe as it goes (new internal `Context::stdin_stream`), so `cat big | head -1` and `while :; do echo; done | cat | head -1` end with SIGPIPE (`141`) like bash instead of running to a limit. `cat` also reports a missing operand and continues with the rest, as GNU cat does. L-PIPE-001 narrowed.
 * **Fix**: `${x#pat}`/`${x%pat}` follow the glob grammar. A lone `*` matches empty for the short forms (`${x#*}` and `${x%*}` keep `x`; they dropped a character), and patterns with `?` or several `*` (`${x%?}`, `${p#*/*/}`, `${p##*.*.}`) now match instead of being compared as literal text. Single-`*` patterns keep their linear fast path.
 * **Fix**: Quoted or escaped glob characters are literal everywhere. `echo a\*`, `a"*"` and `a'?'` no longer glob; `case`/`[[ == ]]` patterns match `"$x"`, `'a*'`, `a\*` and `\[s\]` literally; `"*"zz*` with no match prints `*zz*` (escapes no longer leak); `"$x"zz*` no longer reads `$xzz`; `'a'$x` expands and `'a'*` globs; `a{1,2}"*"` still brace-expands. The lexer escapes quoted glob characters in mixed words (the existing `QuotedGlobWord` convention, now with quote markers); `expand_word` returns unescaped text for non-glob uses; patterns use `expand_pattern_word`. Known gap: data from an unquoted expansion inside such a word loses `\` before glob characters when nothing matches.
 * **Feature**: `fmt` builtin, on the uutils `fmt` engine (MIT, vendored: Knuth-Plass line breaking, paragraphs, `-c`/`-t`/`-s`/`-u`/`-p`, `-w`/`-g`, `-WIDTH`). Width rules and error messages follow GNU. Line breaks can differ from GNU `fmt` on long paragraphs (L-FMT-001); GNU's GPL code was not ported. Step 4 of the level-up roadmap (missing commands). Spec: `fmt.test.sh`. See foundations/builtins "fmt".

knowledge/operations/limitations.md

@@ -46,7 +46,7 @@ execution model. Evidence is a threat-model ID, a test, or `stance`
 | L-PROC-002 | Background jobs run concurrently (`jobs`, `wait -n`, `kill`, `ps`, `pgrep`), but cannot be stopped or resumed: `kill -STOP/-CONT` are ignored, there is no `suspend`/Ctrl-Z, `bg` is a no-op and `fg` just waits; jobs end when `exec()` returns | No terminal process groups in a virtual shell; a job must not outlive the call that owns it (TM-DOS-122) | `l_proc_002_no_job_control` |
 | L-PROC-003 | No process spawning; external commands run as builtins | Core sandbox model: no fork/exec escape surface | `l_proc_003_no_process_spawning` |
 | L-PROC-004 | `bash`/`sh` child shells nest at most 8 deep (and count against `max_function_depth`); deeper nesting fails `maximum nesting depth exceeded` | Child shells run in-process on one stack; unbounded nesting crashed the process (TM-DOS-125) | `l_proc_004_child_shell_depth` |
-| L-PIPE-001 | Pipeline stages stream only from the first stage that runs shell code (loop, group, function, `eval`). Leading single-builtin stages run to completion first and hand over their whole output, so they never get SIGPIPE (`cat big \| head -1` gives `PIPESTATUS` `0 0`, bash `141 0`); the generators `yes` and `seq` are the exception and stream. Downstream of a streaming stage, a single builtin reads all its input before it starts, except `head` and `read`: `while :; do echo; done \| cat \| head -1` runs until a limit. Pipes hold 4 KiB, not 64 KiB, so a finite loop writing more than that into an early-exiting reader exits 141 where bash would exit 0. Pipelines nested more than 4 subshells deep run their stages in sequence | Builtins return their output as one value, not a stream; streaming them needs a reader/writer `Context`. The smaller pipe bounds how far a producer runs ahead (each byte costs budgeted commands); the nesting cap bounds native stack (TM-DOS-124) | `l_pipe_001_stages_run_sequentially` |
+| L-PIPE-001 | Pipeline stages stream only from the first stage that runs shell code (loop, group, function, `eval`). Leading single-builtin stages run to completion first and hand over their whole output, so they never get SIGPIPE (`grep . big \| head -1` gives `PIPESTATUS` `0 0`, bash `141 0`); `yes`, `seq` and plain `cat` are the exception and stream. Downstream of a streaming stage, a single builtin reads all its input before it starts, except `head`, `read` and plain `cat`: `while :; do echo; done \| grep y \| head -1` runs until a limit. Pipes hold 4 KiB, not 64 KiB, so a finite loop writing more than that into an early-exiting reader exits 141 where bash would exit 0. Pipelines nested more than 4 subshells deep run their stages in sequence | Builtins return their output as one value, not a stream; streaming them needs a reader/writer `Context`. The smaller pipe bounds how far a producer runs ahead (each byte costs budgeted commands); the nesting cap bounds native stack (TM-DOS-124) | `l_pipe_001_stages_run_sequentially` |
 | L-SUDO-001 | `sudo` runs its command as the one sandbox user: `-u`, `-g`, `-i` change nothing, `whoami` still prints `sandbox`, and there is never a password prompt. `busybox APPLET` runs bashkit's builtin of that name, not BusyBox's own applet (flags and messages are bashkit's) | The sandbox has no privilege boundary to cross; the VFS has no permission enforcement (L-FS-002) | `sudo-busybox.test.sh` |
 | L-RAND-001 | `uuidgen` makes random (v4) UUIDs only (no `-t`, `-m`, `-s`); `openssl` implements only `rand` | Time/MAC-based UUIDs would leak host identity; a full TLS/crypto toolkit is out of scope | `uuidgen_time_based_unsupported` |
 | L-FS-001 | Symlinks are followed, but `..` after a linked directory resolves lexically (`/link/..` is the link's parent, the `cd -L` view), and `ln` without `-s` (and `link`) makes a symlink, not a hard link | The interpreter normalizes paths before the VFS sees them; the VFS has no inodes to share | `symlink.test.sh`, `ln_default_symbolic` |