fix(streaming): preserve routed byte output and enforce builtin budgets

crates/bashkit/docs/streaming.md

@@ -70,6 +70,13 @@ sandbox delivery truncation.
 before the next execution. Deadlines cover pending sinks as well as engine work.
 Command I/O and filesystem handles expire when the execution ends.
 
+Streaming builtin readers share the aggregate input budget. Record buffers
+and whole-input collectors reserve intermediate memory before growth; even a
+sink-only execution can reject an oversized record. Streaming `cat`, `grep`
+and `tr` output follows descriptor routing without disabling live delivery.
+Process substitutions preserve raw bytes but remain buffered; `$(...)` is
+still a text boundary.
+
 Use buffered execution when `after_tool` or `after_exec` must inspect or replace
 whole output. Live execution rejects those hooks and `ExecOptions` output
 observers; use async byte sinks for live delivery.

crates/bashkit/src/builtins/cat.rs

@@ -73,7 +73,8 @@ impl Builtin for Cat {
             return stream_plain(&ctx, &files, &stream).await;
         }
 
-        let mut raw = Vec::new();
+        let budget = ctx.shell.as_ref().map(|s| s.execution_budget);
+        let mut raw = crate::limits::BudgetedBytes::new(budget)?;
         let mut stderr = String::new();
         // Buffered stdin is read once: `cat - -` sees it at the first `-`.
         let mut stdin_used = false;
@@ -82,10 +83,10 @@ impl Builtin for Cat {
                 if let Some(stdin) = ctx.stdin
                     && !std::mem::replace(&mut stdin_used, true)
                 {
-                    raw.extend_from_slice(stdin.as_bytes());
+                    raw.try_extend_from_slice(stdin.as_bytes())?;
                 }
                 if let Some(input) = ctx.stdin_stream() {
-                    raw.extend_from_slice(&input.read_to_end().await);
+                    raw.try_extend_from_slice(&input.read_to_end().await?)?;
                 }
             } else {
                 let path = if Path::new(file).is_absolute() {
@@ -94,7 +95,7 @@ impl Builtin for Cat {
                     vfs_join(ctx.cwd, file).to_string_lossy().into_owned()
                 };
                 match ctx.fs.read_file(Path::new(&path)).await {
-                    Ok(bytes) => raw.extend_from_slice(&bytes),
+                    Ok(bytes) => raw.try_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);
@@ -104,6 +105,7 @@ impl Builtin for Cat {
             }
         }
 
+        let (raw, _raw_storage) = raw.into_parts();
         let mut result = if plain {
             ExecResult::ok_bytes(raw)
         } else {
@@ -136,7 +138,7 @@ async fn stream_plain(
     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 {
+            if !stream.write(chunk).await? {
                 return Ok(false);
             }
         }
@@ -155,7 +157,7 @@ async fn stream_plain(
             };
             if let Some(input) = ctx.stdin_stream() {
                 while ok {
-                    let chunk = input.read().await;
+                    let chunk = input.read().await?;
                     if chunk.is_empty() {
                         break;
                     }

crates/bashkit/src/builtins/cuttr.rs

@@ -357,28 +357,29 @@ impl Builtin for Tr {
         // `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 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 {
+            while let Some(piece) = input.next(cut).await? {
                 ctx.consume_budget_work(1)?;
-                let output = match spec.apply(&piece.into()) {
+                let piece = ctx.collected_input(piece)?;
+                let output = match spec.apply(&piece) {
                     Ok(output) => output,
                     Err(msg) => return Ok(Self::err(&msg, 1)),
                 };
-                if !output.is_empty() && !out.write(&output).await {
+                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();
+        let stdin = ctx.stdin_to_end().await?;
         match spec.apply(&stdin) {
             Ok(output) => Ok(ExecResult::ok_bytes(output)),
             Err(msg) => Ok(Self::err(&msg, 1)),

crates/bashkit/src/builtins/grep.rs

@@ -900,18 +900,18 @@ async fn scan_input(
     let opts = sh.opts;
     let term = if opts.null_data { b'\0' } else { b'\n' };
     let (name, mut input, whole_binary) = match source {
-        Source::Stdin(name) => (name, super::InputChunks::stdin(ctx), false),
+        Source::Stdin(name) => (name, super::InputChunks::stdin(ctx)?, false),
         Source::File(name, bytes) => {
             let nul = bytes.contains(&0);
-            (name, super::InputChunks::bytes(bytes), nul)
+            (name, super::InputChunks::bytes(ctx, bytes)?, nul)
         }
     };
     let mut scan = FileScan::new(&name, opts);
     if whole_binary {
         scan.set_binary(opts);
     }
     while !scan.done {
-        let Some(piece) = input.next(super::cut_records(term)).await else {
+        let Some(piece) = input.next(super::cut_records(term)).await? else {
             break;
         };
         ctx.consume_budget_work(1)?;
@@ -928,7 +928,7 @@ async fn scan_input(
         if let Some(s) = stream
             && !out.is_empty()
         {
-            if !s.write(out).await {
+            if !s.write(out).await? {
                 return Ok((scan.nlines, true));
             }
             out.clear();
@@ -938,7 +938,7 @@ async fn scan_input(
     if let Some(s) = stream
         && !out.is_empty()
     {
-        if !s.write(out).await {
+        if !s.write(out).await? {
             return Ok((n, true));
         }
         out.clear();
@@ -969,7 +969,11 @@ impl Builtin for Grep {
         for pattern_file in &opts.pattern_files {
             let content = match read_operand_file(&ctx, pattern_file).await {
                 Ok(c) => c,
+                Err(error @ (crate::Error::ResourceLimit(_) | crate::Error::Cancelled)) => {
+                    return Err(error);
+                }
                 Err(reason) => {
+                    let reason = crate::error::io_error_reason(&reason);
                     return Ok(ExecResult::err(
                         format!("grep: {pattern_file}: {reason}\n"),
                         2,
@@ -991,7 +995,11 @@ impl Builtin for Grep {
                     opts.exclude_patterns
                         .extend(text.lines().filter(|l| !l.is_empty()).map(str::to_string));
                 }
+                Err(error @ (crate::Error::ResourceLimit(_) | crate::Error::Cancelled)) => {
+                    return Err(error);
+                }
                 Err(reason) => {
+                    let reason = crate::error::io_error_reason(&reason);
                     return Ok(ExecResult::err(
                         format!("grep: {exclude_file}: {reason}\n"),
                         2,
@@ -1182,18 +1190,11 @@ fn resolve(cwd: &std::path::Path, file: &str) -> std::path::PathBuf {
 
 /// Read a file named on the command line (`-f`, `--exclude-from`); `-`
 /// is standard input. `Err` is the strerror-style reason.
-async fn read_operand_file(ctx: &Context<'_>, name: &str) -> std::result::Result<Vec<u8>, String> {
+async fn read_operand_file(ctx: &Context<'_>, name: &str) -> Result<Vec<u8>> {
     if name == "-" {
-        return Ok(ctx
-            .stdin_to_end()
-            .await
-            .map(|s| s.as_bytes().to_vec())
-            .unwrap_or_default());
+        return Ok(ctx.stdin_to_end().await?.as_bytes().to_vec());
     }
-    ctx.fs
-        .read_file(&resolve(ctx.cwd, name))
-        .await
-        .map_err(|e| crate::error::io_error_reason(&e))
+    ctx.fs.read_file(&resolve(ctx.cwd, name)).await
 }
 
 /// A resolved directory ancestor. Sibling aliases remain independent searches.