From 9a406b97ef73f555f50e7abf57cbc6d7aff73904 Mon Sep 17 00:00:00 2001 From: Mykhailo Chalyi Date: Wed, 7 Oct 2026 04:59:54 +0000 Subject: [PATCH] fix(interpreter): run pipeline stages as subshells Every stage of a multi-command pipeline now rolls back its state changes like bash; shopt -s lastpipe keeps the last stage in the shell. Streaming callers no longer see a non-last stage's output. Records L-PIPE-001 (stages still run sequentially). Claude-Session: https://claude.ai/code/session_019aFikmptPc91Fj4N2iDXQA --- crates/bashkit/docs/compatibility.md | 2 +- crates/bashkit/src/interpreter/mod.rs | 92 +++++++++++++++---- crates/bashkit/src/lib.rs | 7 +- .../integration/limitations_evidence_tests.rs | 13 +++ crates/bashkit/tests/integration/main.rs | 1 + .../tests/integration/pipeline_tests.rs | 48 ++++++++++ .../tests/integration/threat_model_tests.rs | 9 +- .../tests/spec_cases/bash/arrays.test.sh | 11 +-- .../bash/blackbox-edge-cases.test.sh | 3 +- .../spec_cases/bash/pipes-redirects.test.sh | 47 ++++++++++ .../spec_cases/bash/read-builtin.test.sh | 3 +- knowledge/foundations/parallel-execution.md | 13 +++ knowledge/log.md | 1 + knowledge/operations/limitations.md | 3 +- 14 files changed, 218 insertions(+), 35 deletions(-) create mode 100644 crates/bashkit/tests/integration/pipeline_tests.rs diff --git a/crates/bashkit/docs/compatibility.md b/crates/bashkit/docs/compatibility.md index 33df5221c..eb5c3c618 100644 --- a/crates/bashkit/docs/compatibility.md +++ b/crates/bashkit/docs/compatibility.md @@ -206,7 +206,7 @@ Chrono's validated strftime implementation, plus GNU `%N`, `%3N`, `%6N`, and | Operator | Status | Example | Notes | |----------|--------|---------|-------| -| `\|` | ✅ | `cmd1 \| cmd2` | Pipeline | +| `\|` | ✅ | `cmd1 \| cmd2` | Pipeline; each stage is a subshell, `shopt -s lastpipe` keeps the last stage in the shell | | `&&` | ✅ | `cmd1 && cmd2` | AND list | | `\|\|` | ✅ | `cmd1 \|\| cmd2` | OR list | | `;` | ✅ | `cmd1; cmd2` | Sequential | diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index 11a0a7b52..bb39f86eb 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -5342,26 +5342,32 @@ impl Interpreter { let mut pipeline_stderr_truncated = false; let mut pipe_statuses = Vec::new(); + // Every stage of a multi-command pipeline runs in a subshell, except + // the last one under `shopt -s lastpipe` (bash, job control off). + let multi = pipeline.commands.len() > 1; + let lastpipe = self + .scoped + .variables + .get("SHOPT_lastpipe") + .is_some_and(|v| v == "1"); + for (i, command) in pipeline.commands.iter().enumerate() { let is_last = i == pipeline.commands.len() - 1; - - let result = match command { - Command::Simple(simple) => { - self.execution_budget.consume_work(1)?; - self.counters.tick_command(&self.limits)?; - self.execute_simple_command(simple, stdin_data.take()) - .await? - } - _ => { - // Compound commands, lists, etc. in pipeline: - // set pipeline_stdin so inner commands (read, cat, etc.) can consume it - let prev_pipeline_stdin = self.pipeline_stdin.take(); - self.pipeline_stdin = stdin_data.take(); - let result = self.execute_command(command).await?; - self.pipeline_stdin = prev_pipeline_stdin; - result - } + let subshell = multi && !(is_last && lastpipe); + // A non-last stage's stdout feeds the next stage, never the + // streaming observer. + let saved_callback = if is_last { + None + } else { + self.output_callback.take() }; + let result = self + .execute_pipeline_stage(command, stdin_data.take(), subshell) + .await; + if let Some(cb) = saved_callback { + self.output_callback = Some(cb); + } + let result = result?; pipe_statuses.push(result.exit_code); if !pipeline_stderr_truncated { @@ -5410,6 +5416,58 @@ impl Interpreter { Ok(last_result) } + /// Run one pipeline stage with `stdin` from the previous stage. + /// + /// With `subshell`, state changes (variables, cwd, options, fds) are + /// rolled back afterwards and `exit`/`return` end only the stage. + async fn execute_pipeline_stage( + &mut self, + command: &Command, + stdin: Option, + subshell: bool, + ) -> Result { + let saved = subshell.then(|| { + ( + self.snapshot_subshell_state(), + self.call_stack.clone(), + self.coproc_buffers.clone(), + ) + }); + let result = match command { + Command::Simple(simple) => { + let ticked = self + .execution_budget + .consume_work(1) + .and_then(|()| self.counters.tick_command(&self.limits)); + match ticked { + Ok(()) => self.execute_simple_command(simple, stdin).await, + Err(e) => Err(e.into()), + } + } + _ => { + // Compound commands, lists, etc. in pipeline: + // set pipeline_stdin so inner commands (read, cat, etc.) can consume it + let prev_pipeline_stdin = std::mem::replace(&mut self.pipeline_stdin, stdin); + let result = self.execute_command(command).await; + self.pipeline_stdin = prev_pipeline_stdin; + result + } + }; + let Some((snap, call_stack, coproc)) = saved else { + return result; + }; + self.restore_subshell_state(snap); + self.call_stack = call_stack; + self.coproc_buffers = coproc; + let mut result = result?; + if let ControlFlow::Exit(code) | ControlFlow::Return(code) = result.control_flow { + result.exit_code = code; + result.control_flow = ControlFlow::None; + } + result.errexit_suppressed = false; + Ok(result) + } + /// Check if a command is the empty sentinel produced by the parser for trailing `&`. fn is_empty_sentinel(cmd: &Command) -> bool { if let Command::Simple(sc) = cmd { diff --git a/crates/bashkit/src/lib.rs b/crates/bashkit/src/lib.rs index eb1e08524..b63bcb88c 100644 --- a/crates/bashkit/src/lib.rs +++ b/crates/bashkit/src/lib.rs @@ -4872,7 +4872,10 @@ mod tests { #[tokio::test] async fn test_read_basic() { let mut bash = Bash::new(); - let result = bash.exec("echo hello | read VAR; echo $VAR").await.unwrap(); + let result = bash + .exec("shopt -s lastpipe; echo hello | read VAR; echo $VAR") + .await + .unwrap(); assert_eq!(result.stdout, "hello\n"); } @@ -4880,7 +4883,7 @@ mod tests { async fn test_read_multiple_vars() { let mut bash = Bash::new(); let result = bash - .exec("echo 'a b c' | read X Y Z; echo $X $Y $Z") + .exec("shopt -s lastpipe; echo 'a b c' | read X Y Z; echo $X $Y $Z") .await .unwrap(); assert_eq!(result.stdout, "a b c\n"); diff --git a/crates/bashkit/tests/integration/limitations_evidence_tests.rs b/crates/bashkit/tests/integration/limitations_evidence_tests.rs index 60af06b12..c798db898 100644 --- a/crates/bashkit/tests/integration/limitations_evidence_tests.rs +++ b/crates/bashkit/tests/integration/limitations_evidence_tests.rs @@ -164,3 +164,16 @@ async fn l_term_002_read_does_not_wait_for_terminal_input() { term.screen_text() ); } + +/// L-PIPE-001: stages run one after another, so the producer runs to its own +/// cap before `head` reads, and never sees SIGPIPE. +#[tokio::test] +async fn l_pipe_001_stages_run_sequentially() { + 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); +} diff --git a/crates/bashkit/tests/integration/main.rs b/crates/bashkit/tests/integration/main.rs index 9640fe053..24dd28af9 100644 --- a/crates/bashkit/tests/integration/main.rs +++ b/crates/bashkit/tests/integration/main.rs @@ -105,6 +105,7 @@ pub mod network_security_tests; pub mod output_truncation_tests; pub mod parallel_sessions_tests; pub mod partial_parse_tests; +pub mod pipeline_tests; pub mod proptest_differential; pub mod python_integration_tests; pub mod python_security_tests; diff --git a/crates/bashkit/tests/integration/pipeline_tests.rs b/crates/bashkit/tests/integration/pipeline_tests.rs new file mode 100644 index 000000000..fc8838a64 --- /dev/null +++ b/crates/bashkit/tests/integration/pipeline_tests.rs @@ -0,0 +1,48 @@ +//! Pipeline execution: stage isolation and streaming output. +//! +//! Bash-parity cases live in `spec_cases/bash/pipes-redirects.test.sh`; these +//! cover what spec files cannot see, like the streaming output callback. + +use bashkit::Bash; +use std::sync::{Arc, Mutex}; + +/// Output of a non-last stage feeds the next stage; it must never reach the +/// streaming callback directly (`for ... | tac` printed in input order). +#[tokio::test] +async fn non_last_stage_output_is_not_streamed() { + let streamed = Arc::new(Mutex::new(String::new())); + let sink = streamed.clone(); + let mut bash = Bash::new(); + let result = bash + .exec_streaming( + "for i in 1 2 3; do echo $i; done | tac\nf() { echo a; echo b; }; f | sort -r", + Box::new(move |stdout, _stderr| { + sink.lock().unwrap().push_str(&stdout.to_string()); + }), + ) + .await + .unwrap(); + assert_eq!(result.stdout, "3\n2\n1\nb\na\n"); + assert_eq!(*streamed.lock().unwrap(), "3\n2\n1\nb\na\n"); +} + +/// The last stage still streams as it runs. +#[tokio::test] +async fn last_stage_output_streams() { + let chunks = Arc::new(Mutex::new(Vec::new())); + let sink = chunks.clone(); + let mut bash = Bash::new(); + let result = bash + .exec_streaming( + "echo go | while read -r w; do echo $w-1; echo $w-2; done", + Box::new(move |stdout, _stderr| { + if !stdout.is_empty() { + sink.lock().unwrap().push(stdout.to_string()); + } + }), + ) + .await + .unwrap(); + assert_eq!(result.stdout, "go-1\ngo-2\n"); + assert_eq!(chunks.lock().unwrap().concat(), "go-1\ngo-2\n"); +} diff --git a/crates/bashkit/tests/integration/threat_model_tests.rs b/crates/bashkit/tests/integration/threat_model_tests.rs index 0343f2268..4feebd19f 100644 --- a/crates/bashkit/tests/integration/threat_model_tests.rs +++ b/crates/bashkit/tests/integration/threat_model_tests.rs @@ -2804,7 +2804,7 @@ mod variable_namespace_injection { async fn tm_inj_009_read_rejects_internal_prefixes() { for prefix in INTERNAL_PREFIXES { let result = exec(&format!( - "echo injected | read {prefix}; echo ${{{prefix}:-blocked}}" + "shopt -s lastpipe; echo injected | read {prefix}; echo ${{{prefix}:-blocked}}" )) .await; assert!( @@ -2817,14 +2817,17 @@ mod variable_namespace_injection { #[tokio::test] async fn tm_inj_009_read_allows_normal_vars() { - let result = exec("echo hello | read MY_VAR; echo $MY_VAR").await; + let result = exec("shopt -s lastpipe; echo hello | read MY_VAR; echo $MY_VAR").await; assert_eq!(result.exit_code, 0); assert!(result.stdout.contains("hello")); } #[tokio::test] async fn tm_inj_009_read_array_rejects_internal_prefixes() { - let result = exec("echo 'a b c' | read -a _NAMEREF_x; echo ${_NAMEREF_x:-blocked}").await; + let result = exec( + "shopt -s lastpipe; echo 'a b c' | read -a _NAMEREF_x; echo ${_NAMEREF_x:-blocked}", + ) + .await; assert!( result.stdout.contains("blocked"), "read -a should block _NAMEREF_x: got {:?}", diff --git a/crates/bashkit/tests/spec_cases/bash/arrays.test.sh b/crates/bashkit/tests/spec_cases/bash/arrays.test.sh index e83eda183..c350cc94b 100644 --- a/crates/bashkit/tests/spec_cases/bash/arrays.test.sh +++ b/crates/bashkit/tests/spec_cases/bash/arrays.test.sh @@ -156,9 +156,8 @@ b ### end ### mapfile_basic -### bash_diff: pipes don't create subshells in bashkit (stateless model) -# mapfile reads lines into array from pipe -printf 'a\nb\nc\n' | mapfile -t lines; echo ${#lines[@]}; echo ${lines[0]}; echo ${lines[1]}; echo ${lines[2]} +# mapfile reads lines into array from pipe (lastpipe keeps the array) +shopt -s lastpipe; printf 'a\nb\nc\n' | mapfile -t lines; echo ${#lines[@]}; echo ${lines[0]}; echo ${lines[1]}; echo ${lines[2]} ### expect 3 a @@ -167,17 +166,15 @@ c ### end ### readarray_alias -### bash_diff: pipes don't create subshells in bashkit (stateless model) # readarray is an alias for mapfile -printf 'x\ny\n' | readarray -t arr; echo ${arr[0]} ${arr[1]} +shopt -s lastpipe; printf 'x\ny\n' | readarray -t arr; echo ${arr[0]} ${arr[1]} ### expect x y ### end ### mapfile_default_name -### bash_diff: pipes don't create subshells in bashkit (stateless model) # mapfile default array name is MAPFILE -printf 'hello\nworld\n' | mapfile -t; echo ${MAPFILE[0]}; echo ${MAPFILE[1]} +shopt -s lastpipe; printf 'hello\nworld\n' | mapfile -t; echo ${MAPFILE[0]}; echo ${MAPFILE[1]} ### expect hello world diff --git a/crates/bashkit/tests/spec_cases/bash/blackbox-edge-cases.test.sh b/crates/bashkit/tests/spec_cases/bash/blackbox-edge-cases.test.sh index 9da998612..fb9d8a138 100644 --- a/crates/bashkit/tests/spec_cases/bash/blackbox-edge-cases.test.sh +++ b/crates/bashkit/tests/spec_cases/bash/blackbox-edge-cases.test.sh @@ -91,9 +91,8 @@ printf '%o\n' 255 ### end ### read_n_chars -### bash_diff: pipe creates subshell in real bash, read -n result lost; bashkit keeps it # Read specific number of characters -echo "hello" | read -n 3 x; echo "$x" +shopt -s lastpipe; echo "hello" | read -n 3 x; echo "$x" ### expect hel ### end diff --git a/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh b/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh index ffb1fa646..48e2ef383 100644 --- a/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh +++ b/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh @@ -285,3 +285,50 @@ test -f /tmp/nullcmd_assign && echo exists 1 exists ### end + +### pipeline_last_stage_is_subshell +# Without lastpipe every pipeline stage runs in a subshell +echo x | read v +echo "${v-unset}" +n=0 +printf 'a\nb\n' | while read -r l; do n=$((n+1)); done +echo "n=$n" +### expect +unset +n=0 +### end + +### pipeline_lastpipe_runs_last_stage_in_shell +# shopt -s lastpipe keeps the last stage in the current shell +shopt -s lastpipe +echo x | read v +echo "v=$v" +n=0 +printf 'a\nb\n' | while read -r l; do n=$((n+1)); done +echo "n=$n" +### expect +v=x +n=2 +### end + +### pipeline_first_stage_state_does_not_leak +# Assignments and cd in a non-last stage stay in that stage +x=1 +x=2 | cat +cd / +cd /tmp | true +echo "x=$x pwd=$PWD" +### expect +x=1 pwd=/ +### end + +### pipeline_stage_exit_only_ends_the_stage +# exit inside a pipeline stage ends that stage, not the shell +echo hi | exit 3 +echo "after $?" +exit 4 | cat +echo "still here" +### expect +after 3 +still here +### end diff --git a/crates/bashkit/tests/spec_cases/bash/read-builtin.test.sh b/crates/bashkit/tests/spec_cases/bash/read-builtin.test.sh index 21bd2c754..4170ea1e2 100644 --- a/crates/bashkit/tests/spec_cases/bash/read-builtin.test.sh +++ b/crates/bashkit/tests/spec_cases/bash/read-builtin.test.sh @@ -1,6 +1,5 @@ ### read_basic -### bash_diff -echo "hello" | read var; echo "$var" +shopt -s lastpipe; echo "hello" | read var; echo "$var" ### expect hello ### end diff --git a/knowledge/foundations/parallel-execution.md b/knowledge/foundations/parallel-execution.md index ac5f7abd3..855e042bd 100644 --- a/knowledge/foundations/parallel-execution.md +++ b/knowledge/foundations/parallel-execution.md @@ -46,6 +46,19 @@ Implemented becomes 128+signal. - Limit: `max_background_jobs` (TM-DOS-122). +## Pipelines + +- Every stage of a multi-command pipeline runs in a subshell: variables, + cwd, options and fds changed in a stage are rolled back after it, and + `exit`/`return` end only that stage. `shopt -s lastpipe` keeps the last + stage in the current shell (bash does this when job control is off, which + is always the case here). So `echo x | read v` leaves `v` unset, as in bash. +- Stages reuse the cheap subshell snapshot (`execute_pipeline_stage`), not a + fork: they still run one after another (L-PIPE-001). +- A non-last stage's stdout feeds the next stage only; the streaming output + callback is suspended while it runs, so `for ... | tac` streams the + reversed lines, not the loop's. + ## Benchmark Run `cargo bench --bench parallel_execution` when changes touch: diff --git a/knowledge/log.md b/knowledge/log.md index 6d6973d3b..b2c747d4e 100644 --- a/knowledge/log.md +++ b/knowledge/log.md @@ -2,6 +2,7 @@ ## 2026-10-07 +* **Fix**: Pipeline stages are subshells, as in bash: `echo x | read v` leaves `v` unset and `cd /tmp | true` keeps the cwd; `shopt -s lastpipe` keeps the last stage in the shell. Streaming callers no longer see a non-last stage's output (`for ...; done | tac` streamed the loop's lines in input order). Stages still run sequentially (new L-PIPE-001). See [Parallel Execution](foundations/parallel-execution.md#pipelines). * **Performance**: CPython `random` seeds lazily. The snapshot's `random._inst` gets a subclass whose entry points (`random`, `getrandbits`, `getstate`, `seed`, `setstate`) seed from `os.urandom` on first use, then switch it back to `Random` and rebind those five module functions. Calls that never touch `random` skip the reseed (`pass` 3.7 to 3.5 ms). See [CPython WebAssembly Runtime](runtimes/cpython-wasm.md). * **Performance**: Recorded CPython benches after #2534/#2535 (`criterion-python-vm-linux-x86_64-1791345434.md`): first call ~19 ms, warm `print(1)` 4.4 ms (was 6.3), 64 parallel sessions 157 ms (was 291), load ~165 calls/s (was ~90). `/benches` now shows Python startup (CPython vs Monty) from these reports. See [Performance Results](operations/performance-results.md). * **Feature**: Secret generators. `$SRANDOM` (bash 5.1, 32 bits from `getrandom`), `uuidgen` (v4 only, L-RAND-001) and `openssl rand [-hex|-base64] [-out] N` (only `rand`; capped at 1 MiB, TM-DOS-123). All draw from the OS CSPRNG like `/dev/urandom`, never the `$RANDOM` LCG. diff --git a/knowledge/operations/limitations.md b/knowledge/operations/limitations.md index fc0171aec..3ec89d677 100644 --- a/knowledge/operations/limitations.md +++ b/knowledge/operations/limitations.md @@ -45,6 +45,7 @@ execution model. Evidence is a threat-model ID, a test, or `stance` | L-PROC-001 | `exec` does not replace the process; `exec cmd` runs cmd then stops execution (fd redirects work) | True process replace would break sandbox containment | TM-ESC-005 | | 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-PIPE-001 | Pipeline stages run one after another, not concurrently: a stage finishes before the next one reads its output. An endless producer piped into an early-exiting reader (`yes \| head -1`, `while :; do echo; done \| head`) runs until its own cap or a limit, and no stage gets SIGPIPE (exit 141) | Stages hand over complete output buffers; streaming stages over a bounded pipe are the next step of the fidelity roadmap | `l_pipe_001_stages_run_sequentially` | | 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` 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` | | L-ROOTFS-001 | Default rootfs is static and read-only: `/proc` has no `self`, pid dirs, `uptime` or live counters; `/etc/passwd` has no root entry; `/dev/zero` yields 1 MiB per read; `/bin`, `/usr/bin` are stubs that dispatch builtins | Host state must not leak (TM-INF-003, TM-ISO-018); fixed values keep runs deterministic | `rootfs_layout`, `threat_etc_passwd_blocked` | @@ -84,7 +85,7 @@ Target: IEEE 1003.1-2024 Shell Command Language. | Special built-in utilities | Substantial | 14/15; `exec` partial (L-PROC-001); `times` returns zeros; `trap` per L-SIG-001 | | Quoting, redirections, compound commands, functions | Full | | | Word expansions | Substantial | Most expansions supported | -| Pipelines and lists | Full | `\|`, `&&`, `\|\|`, `;`, `&`+`wait`, `!` | +| Pipelines and lists | Full | `\|`, `&&`, `\|\|`, `;`, `&`+`wait`, `!`; every stage is a subshell, `shopt -s lastpipe` keeps the last one in the shell (L-PIPE-001: stages are not concurrent) | ## Shell Features