feat(interpreter): stream yes and seq into pipelines with SIGPIPE

crates/bashkit/src/builtins/mod.rs

@@ -950,6 +950,17 @@ pub struct Context<'a> {
     pub(crate) shell: Option<ShellRef<'a>>,
 }
 
+/// Stdout pipe of a streaming pipeline stage, see [`Context::stdout_stream`].
+pub(crate) struct StdoutStream(std::sync::Arc<crate::interpreter::pipe::Pipe>);
+
+impl StdoutStream {
+    /// Wait for room in the pipe, then write. False once nobody reads.
+    pub(crate) async fn write(&self, data: &[u8]) -> bool {
+        self.0.writable().await;
+        self.0.write(data)
+    }
+}
+
 impl<'a> Context<'a> {
     /// Exact pipeline stdin bytes.
     pub fn stdin_bytes(&self) -> Option<&[u8]> {
@@ -1032,6 +1043,17 @@ impl<'a> Context<'a> {
             .and_then(|shell| shell.execution_extensions.get::<T>())
     }
 
+    /// Streaming stdout when this builtin is a producer stage of a
+    /// concurrent pipeline (`yes | head -1`). Write chunks as they are made
+    /// instead of returning them in `ExecResult::stdout`; when a write
+    /// reports the reader gone, stop and exit 141 (SIGPIPE).
+    pub(crate) fn stdout_stream(&self) -> Option<StdoutStream> {
+        self.shell
+            .as_ref()
+            .and_then(|shell| shell.stdout_pipe.clone())
+            .map(StdoutStream)
+    }
+
     /// 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/builtins/seq.rs

@@ -18,6 +18,9 @@ use crate::interpreter::ExecResult;
 ///   -w         Equalize width by padding with leading zeroes
 pub struct Seq;
 
+/// Bytes per write when streaming into a pipeline (one pipe's capacity).
+const STREAM_CHUNK_BYTES: usize = 4 * 1024;
+
 #[async_trait]
 impl Builtin for Seq {
     async fn execute(&self, ctx: Context<'_>) -> Result<ExecResult> {
@@ -140,6 +143,10 @@ impl Builtin for Seq {
         };
 
         let mut output = String::new();
+        // Pipeline producer: flush full chunks as they are made, so
+        // `seq 1000000 | head -1` stops early with SIGPIPE.
+        let stream = ctx.stdout_stream();
+        let mut flushed = 0usize;
         let mut current = first;
         let mut first_item = true;
 
@@ -160,7 +167,17 @@ impl Builtin for Seq {
                 capped = Some(("line", SEQ_MAX_LINES));
                 break;
             }
-            if output.len() > SEQ_MAX_OUTPUT_BYTES {
+            if let Some(stream) = &stream
+                && output.len() >= STREAM_CHUNK_BYTES
+            {
+                ctx.consume_budget_work(1)?;
+                if !stream.write(output.as_bytes()).await {
+                    return Ok(ExecResult::with_code("", 141));
+                }
+                flushed += output.len();
+                output.clear();
+            }
+            if flushed + output.len() > SEQ_MAX_OUTPUT_BYTES {
                 capped = Some(("output byte", SEQ_MAX_OUTPUT_BYTES));
                 break;
             }
@@ -187,9 +204,15 @@ impl Builtin for Seq {
             current += increment;
         }
 
-        if !output.is_empty() {
+        if !first_item {
             output.push('\n');
         }
+        if let Some(stream) = &stream {
+            if !stream.write(output.as_bytes()).await {
+                return Ok(ExecResult::with_code("", 141));
+            }
+            output.clear();
+        }
 
         if let Some((what, limit)) = capped {
             return Ok(cap_exceeded("seq", output, what, limit));

crates/bashkit/src/builtins/yes.rs

@@ -15,6 +15,9 @@ use crate::interpreter::ExecResult;
 /// In bashkit, output is limited to avoid infinite loops.
 pub struct Yes;
 
+/// Bytes per write when streaming into a pipeline (one pipe's capacity).
+const STREAM_CHUNK_BYTES: usize = 4 * 1024;
+
 impl BuiltinHelper for Yes {
     const NAME: &'static str = "yes";
 }
@@ -66,6 +69,23 @@ impl Builtin for Yes {
         // THREAT[TM-DOS-109]: the cap always ends `yes`; say so on stderr so a
         // consumer wanting more lines than the cap can tell output was cut.
         let output = build_yes_output(&text);
+        if let Some(stream) = ctx.stdout_stream() {
+            // Pipeline producer: hand the reader one chunk at a time, so
+            // `yes | head -1` stops after the first chunk with SIGPIPE.
+            let lines = output.lines().count();
+            for chunk in output.as_bytes().chunks(STREAM_CHUNK_BYTES) {
+                ctx.consume_budget_work(1)?;
+                if !stream.write(chunk).await {
+                    return Ok(ExecResult::with_code("", 141));
+                }
+            }
+            return Ok(super::limits::cap_exceeded(
+                "yes",
+                String::new(),
+                "output",
+                format!("{lines} lines"),
+            ));
+        }
         let lines = output.lines().count();
         Ok(super::limits::cap_exceeded(
             "yes",

crates/bashkit/src/interpreter/mod.rs

@@ -16,7 +16,7 @@ mod brace_expansion;
 mod expansion;
 mod glob;
 mod jobs;
-mod pipe;
+pub(crate) mod pipe;
 mod redirection;
 mod state;
 mod time_command;
@@ -366,6 +366,8 @@ pub(crate) struct ShellRef<'a> {
     pub(crate) jobs: &'a SharedJobTable,
     /// Typed per-execution extensions for the current `exec*()` call.
     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>>,
 }
 
 // Interpreter-dispatched "special" builtins, listed here so the public
@@ -1413,6 +1415,12 @@ pub struct Interpreter {
     /// Stdout pipe of this forked pipeline stage: command boundaries wait
     /// here for room and stop with 141 once the reader is gone.
     pipe_out: Option<Arc<pipe::Pipe>>,
+    /// Address of the stage's own simple command when it is a streaming
+    /// builtin (`yes`, `seq`): only that command writes straight into
+    /// `pipe_out`, never a command run from its argument expansions.
+    stream_stdout_command: Option<usize>,
+    /// Pipe handed to the next builtin's `Context` (taken at dispatch).
+    builtin_stdout_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".
@@ -1986,6 +1994,8 @@ impl Interpreter {
             pipeline_stdin: None,
             pipe_in: None,
             pipe_out: None,
+            stream_stdout_command: None,
+            builtin_stdout_pipe: None,
             getopts_char_idx: 0,
             last_bg_pid: None,
             output_callback: None,
@@ -2141,6 +2151,8 @@ impl Interpreter {
             pipeline_stdin: None,
             pipe_in: None,
             pipe_out: None,
+            stream_stdout_command: None,
+            builtin_stdout_pipe: None,
             getopts_char_idx: self.getopts_char_idx,
             last_bg_pid: self.last_bg_pid.clone(),
             output_callback: None,
@@ -5826,6 +5838,19 @@ impl Interpreter {
         !matches!(name, "eval" | "source" | "." | "bash" | "sh")
             && !self.scoped.functions.contains_key(name)
             && !self.scoped.aliases.contains_key(name)
+            && !Self::streams_stdout(simple)
+    }
+
+    /// Builtins that write their stdout to a pipeline as they go
+    /// (`Context::stdout_stream`): generators whose output is unbounded or
+    /// large, so `yes | head -1` and `seq 1000000 | head -1` stop at the
+    /// first line with SIGPIPE instead of running into the output caps.
+    fn streams_stdout(simple: &SimpleCommand) -> bool {
+        simple.redirects.is_empty()
+            && matches!(
+                simple.name.parts.as_slice(),
+                [WordPart::Literal(name)] if matches!(name.as_str(), "yes" | "seq")
+            )
     }
 
     /// Run `commands` (the tail of a pipeline) concurrently: every stage but
@@ -5867,6 +5892,11 @@ impl Interpreter {
             let command = command.clone();
             stages.push(Box::pin(async move {
                 let _read_end = read_end;
+                if let Command::Simple(simple) = &command
+                    && Self::streams_stdout(simple)
+                {
+                    child.stream_stdout_command = Some(simple as *const SimpleCommand as usize);
+                }
                 let jobs = Arc::clone(&child.jobs);
                 let result = jobs::with_jobs(&jobs, child.execute_command(&command)).await;
                 jobs.finish_all().await;
@@ -6951,6 +6981,16 @@ impl Interpreter {
         stdin: Option<crate::StreamData>,
     ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<ExecResult>> + Send + 'a>> {
         Box::pin(async move {
+            // The stage's own `yes`/`seq` writes straight into the pipe.
+            self.builtin_stdout_pipe = if self.stream_stdout_command
+                == Some(command as *const SimpleCommand as usize)
+                && command.redirects.is_empty()
+                && !self.scoped.functions.contains_key(name)
+            {
+                self.pipe_out.clone()
+            } 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());
@@ -7354,6 +7394,7 @@ impl Interpreter {
                     limits: &self.limits,
                     jobs: &self.jobs,
                     execution_extensions,
+                    stdout_pipe: None,
                 };
                 let plan_ctx = builtins::Context {
                     args,
@@ -7419,6 +7460,7 @@ impl Interpreter {
                 limits: &self.limits,
                 jobs: &self.jobs,
                 execution_extensions,
+                stdout_pipe: self.builtin_stdout_pipe.take(),
             };
             let ctx = builtins::Context {
                 args,

crates/bashkit/tests/integration/limitations_evidence_tests.rs

@@ -171,17 +171,29 @@ async fn l_term_002_cat_does_not_wait_for_terminal_input() {
     );
 }
 
-/// L-PIPE-001: a leading single-builtin stage runs to its own cap before
-/// `head` reads, and never sees SIGPIPE.
+/// L-PIPE-001: a leading single-builtin stage (other than the streaming
+/// generators `yes`/`seq`) 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[*]}\"")
+        .await
+        .unwrap();
+    assert_eq!(result.stdout, "1\n0 0\n");
+}
+
+/// The generators stream: `yes | head -1` stops with SIGPIPE, as in bash.
+#[tokio::test]
+async fn l_pipe_001_generators_stream() {
     let mut bash = Bash::new();
     let result = bash
         .exec("yes | head -1; echo \"${PIPESTATUS[*]}\"")
         .await
         .unwrap();
-    assert_eq!(result.stdout, "y\n1 0\n");
-    assert!(result.stderr.contains("output limit"), "{}", result.stderr);
+    assert_eq!(result.stdout, "y\n141 0\n");
+    assert!(result.stderr.is_empty(), "{}", result.stderr);
 }
 
 /// L-PROC-004: child shells nest at most 8 deep.

crates/bashkit/tests/spec_cases/bash/pipe-generators.test.sh

@@ -0,0 +1,38 @@
+# Generators at the head of a pipeline stop when the reader goes away
+# (SIGPIPE, status 141) instead of running to an output cap.
+
+### yes_head_sigpipe
+yes | head -2
+echo "${PIPESTATUS[*]}"
+### expect
+y
+y
+141 0
+### end
+
+### seq_million_head
+seq 1000000 | head -1
+echo "${PIPESTATUS[*]}"
+### expect
+1
+141 0
+### end
+
+### seq_full_output_through_pipe
+seq 5000 | tail -1
+seq -s, 5 | cat
+seq 3 | wc -l
+### expect
+5000
+1,2,3,4,5
+3
+### end
+
+### yes_custom_text_head_bytes
+yes ab | head -c 7
+echo
+### expect
+ab
+ab
+a
+### end

knowledge/foundations/parallel-execution.md

@@ -57,6 +57,13 @@ Implemented
   first, one after another, on the cheap subshell snapshot
   (`enter_pipeline_stage`/`exit_pipeline_stage`): a builtin produces its
   output as one value anyway, and this keeps `echo x | grep x` fork-free.
+- The generators `yes` and `seq` (unredirected, not shadowed by a
+  function) also start the concurrent part: as a producer stage they write
+  4 KiB chunks straight into the pipe through `Context::stdout_stream`
+  (the interpreter hands the pipe only to the stage's own simple command,
+  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.
 - 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

@@ -2,6 +2,7 @@
 
 ## 2026-10-07
 
+* **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`.
 * **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.

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: `yes \| head -1` gives `PIPESTATUS` `1 0` plus yes's output-cap notice, where bash gives `141 0`. 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 (`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-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` |