fix(interpreter): run pipeline stages as subshells

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 |

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<crate::StreamData>,
+        subshell: bool,
+    ) -> Result<ExecResult> {
+        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 {

crates/bashkit/src/lib.rs

@@ -4872,15 +4872,18 @@ 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");
     }
 
     #[tokio::test]
     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");

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);
+}

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;

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");
+}

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 {:?}",

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

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

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

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

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:

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.

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