fix(interpreter): charge input budget for numbered input redirects

crates/bashkit/Cargo.toml

@@ -450,3 +450,8 @@ wasm-bindgen-futures = { version = "0.4", optional = true }
 # not flag the vendored code. See knowledge/runtimes/coreutils-args-port.md.
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(feature, values("i18n-decimal"))'] }
+
+# The process feature is native test infrastructure only; keep WASM builds
+# on the workspace Tokio feature set without changing versions or the lock.
+[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
+tokio = { workspace = true, features = ["process"] }

crates/bashkit/src/fs/proc_sub_fds.rs

@@ -121,6 +121,15 @@ impl ProcSubFs {
         self.table().iter().any(|(n, _)| *n == fd)
     }
 
+    /// Buffered read size for admission; pipe stat keeps its POSIX zero size.
+    pub(crate) fn input_len(&self, path: &Path) -> Option<usize> {
+        match self.slot(path) {
+            Slot::Open(_, ProcSubData::Input(data)) => Some(data.len()),
+            Slot::Open(_, ProcSubData::Output(_)) => Some(0),
+            Slot::Vfs | Slot::Closed => None,
+        }
+    }
+
     pub(crate) fn open(&self, fd: i32, data: ProcSubData) {
         self.table().push((fd, data));
     }

crates/bashkit/src/interpreter/mod.rs

@@ -9426,7 +9426,7 @@ impl Interpreter {
             }
 
             // Handle input redirections first
-            let stdin = match self
+            let stdin_setup = match self
                 .process_input_redirections(
                     stdin,
                     &command.redirects,
@@ -9441,6 +9441,10 @@ impl Interpreter {
                 }
                 Err(e) => return Err(e),
             };
+            // Keep the materialized stdin's byte/text storage admitted
+            // for as long as the command retains it.
+            let _stdin_storage = stdin_setup.stdin_lease;
+            let stdin = stdin_setup.stdin;
 
             // `read -u FD` on a readable fd (`exec 3<f`, a coproc): its
             // next line is stdin.
@@ -12914,7 +12918,7 @@ impl Interpreter {
             // WTF: a coproc fd as a compound's stdin (`while read l; do
             // ...; done <&${C[0]}`) is read to end of input up front, not
             // line by line as the loop runs.
-            let stdin = match self
+            let stdin_setup = match self
                 .process_input_redirections(None, redirects, StdinDemand::All, HighFdStdin::Any)
                 .await
             {
@@ -12924,6 +12928,9 @@ impl Interpreter {
                 }
                 Err(e) => return Err(e),
             };
+            // Keep stdin's byte/text storage admitted through the body.
+            let _stdin_storage = stdin_setup.stdin_lease;
+            let stdin = stdin_setup.stdin;
             let prev_pipeline_stdin = if stdin.is_some() {
                 let prev = self.pipeline_stdin.take();
                 self.pipeline_stdin = stdin;
@@ -13020,10 +13027,12 @@ impl Interpreter {
                     .await
                 };
                 match read {
-                    Ok(content) => {
-                        if let Some(content) = content {
-                            // Account for the VFS-owned input while it and the
-                            // decoded substitution text are simultaneously live.
+                    Ok(setup) => {
+                        // Retain admitted input storage through substitution.
+                        let _stdin_storage = setup.stdin_lease;
+                        if let Some(content) = setup.stdin {
+                            // Cover the temporary byte copy made by the
+                            // command-substitution text boundary.
                             let _content_lease =
                                 self.execution_budget.lease_bytes(content.len())?;
                             stdout.try_push_str(&content.command_substitution_text())?;

crates/bashkit/src/stream.rs

@@ -17,6 +17,41 @@ pub struct StreamData {
 }
 
 impl StreamData {
+    /// Decode a VFS buffer whose byte capacity is already leased by its caller.
+    pub(crate) fn from_admitted_bytes(
+        bytes: Vec<u8>,
+        budget: &crate::limits::ExecutionBudget,
+    ) -> Result<(Self, Option<crate::limits::ExecutionBudgetLease>), crate::limits::LimitExceeded>
+    {
+        let mut text = crate::limits::BudgetedString::new(Some(budget))?;
+        for chunk in bytes.utf8_chunks() {
+            budget.consume_work(1 + chunk.valid().len().div_ceil(1024) as u64)?;
+            budget.consume_input(chunk.valid().len())?;
+            text.try_push_str(chunk.valid())?;
+            if !chunk.invalid().is_empty() {
+                budget.consume_input('\u{fffd}'.len_utf8())?;
+                text.try_push('\u{fffd}')?;
+            }
+        }
+        let (text, storage) = text.into_parts();
+        Ok((Self { bytes, text }, storage))
+    }
+
+    /// Copy both representations through the existing capacity-owning builders.
+    pub(crate) fn try_clone_budgeted(
+        &self,
+        budget: &crate::limits::ExecutionBudget,
+    ) -> Result<(Self, StreamStorage), crate::limits::LimitExceeded> {
+        budget.consume_work(1 + self.retained_len().div_ceil(1024) as u64)?;
+        let mut bytes = crate::limits::BudgetedBytes::new(Some(budget))?;
+        let mut text = crate::limits::BudgetedString::new(Some(budget))?;
+        bytes.try_extend_from_slice(&self.bytes)?;
+        text.try_push_str(&self.text)?;
+        let (bytes, byte_storage) = bytes.into_parts();
+        let (text, text_storage) = text.into_parts();
+        Ok((Self { bytes, text }, [byte_storage, text_storage]))
+    }
+
     /// Convert admitted text to the byte/text stream without an uncharged copy.
     pub(crate) fn from_budgeted_string(
         text: crate::limits::BudgetedString,
@@ -73,6 +108,12 @@ impl StreamData {
     pub fn len(&self) -> usize {
         self.bytes.len()
     }
+
+    /// Live bytes retained by this buffer: the raw bytes plus the decoded
+    /// text kept alongside them.
+    pub(crate) fn retained_len(&self) -> usize {
+        self.bytes.len().saturating_add(self.text.len())
+    }
     pub fn is_empty(&self) -> bool {
         self.bytes.is_empty()
     }
@@ -220,6 +261,43 @@ impl serde::Serialize for StreamData {
 mod tests {
     use super::StreamData;
 
+    #[test]
+    fn admitted_bytes_preserve_lossy_text_and_lease() {
+        let budget = crate::limits::ExecutionBudget::new(
+            &crate::ExecutionLimits::new(),
+            std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        );
+        let bytes = b"a\xff\0".to_vec();
+        let raw = budget.lease_bytes(bytes.capacity()).unwrap();
+        let (stream, text) = StreamData::from_admitted_bytes(bytes, &budget).unwrap();
+        assert_eq!(stream.as_bytes(), b"a\xff\0");
+        assert_eq!(stream.text_lossy(), "a\u{fffd}\0");
+        assert!(budget.live_bytes_for_test() >= 8);
+        drop((stream, raw, text));
+        assert_eq!(budget.live_bytes_for_test(), 0);
+    }
+
+    #[test]
+    fn budgeted_clone_exact_boundary_and_refund() {
+        for (limit, succeeds) in [(3, false), (4, true)] {
+            let budget = crate::limits::ExecutionBudget::new(
+                &crate::ExecutionLimits::new().max_live_intermediate_bytes(limit),
+                std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+            );
+            let source = StreamData::from("ab");
+            let result = source.try_clone_budgeted(&budget);
+            if succeeds {
+                let (copy, storage) = result.unwrap();
+                assert_eq!(copy, source);
+                assert_eq!(budget.live_bytes_for_test(), 4);
+                drop((copy, storage));
+            } else {
+                assert!(result.is_err());
+            }
+            assert_eq!(budget.live_bytes_for_test(), 0);
+        }
+    }
+
     #[test]
     fn budgeted_string_stream_admits_both_representations() {
         for limit in [7, 8] {