A job's stdout stream holds only the job's own output

CHANGELOG.md

@@ -31,6 +31,10 @@ breaking entries are marked **BREAKING**.
 - A runtime fault keeps the output that ran before it. `echo left &&
   x=$((1/0))` shows `left`, and a function that faults exits 1 with what it
   printed and the full cause instead of only the outermost message.
+- A background job's stdout stream (`/v/jobs/N/stdout`) now holds only the
+  job's own output, in order: builtin output is published when each builtin
+  returns, and output captured by `$(...)`, redirected with `>`, or read by
+  `scatter` no longer appears in it.
 - `grep` now reserves exit 1 for "no lines matched". An invalid pattern, an
   unreadable file, or a missing pattern argument exits 2, so a caller cannot
   read a broken search as a negative answer. `diff` argument errors exit 2 to

crates/kaish-help/content/en/vfs.md

@@ -45,11 +45,12 @@ jobs --cleanup             # remove completed jobs
 ```
 
 `stdout` and `stderr` fill as an external command emits. A builtin does not
-stream: it returns its whole output when it finishes, so `echo hi &` fills
-the node in one write at the end — and so does `cargo build | tee log &`,
-because `tee` is a builtin. Drop the `| tee`; the job's stream is the log.
-Only the last stage of a pipeline reaches `stdout` (an earlier stage's output
-is the next stage's stdin); `stderr` takes every stage's.
+stream: it writes its whole output when it returns, so `echo hi &` fills the
+node in one write — and so does `cargo build | tee log &`, because `tee` is a
+builtin. Drop the `| tee`; the job's stream is the log.
+Only the job's own output reaches `stdout`. An earlier pipeline stage's
+output, `$(...)` output, a redirected stdout, and a scatter worker's output do
+not. `stderr` takes every stage's.
 
 Each node holds at most 10MB and evicts its oldest bytes past that. Redirect
 to a file (`cargo build > /tmp/build.log 2>&1 &`) when the whole output

crates/kaish-kernel/src/kernel.rs

@@ -1450,7 +1450,7 @@ impl Kernel {
         job_id: crate::scheduler::JobId,
     ) -> Arc<Self> {
         let fork = self.fork_inner(cancel, Some(job_id)).await;
-        fork.exec_ctx.write().await.background_stream_external_output = true;
+        fork.exec_ctx.write().await.background_stream_output = true;
         fork
     }
 
@@ -1893,11 +1893,11 @@ impl Kernel {
         let job_id = self.jobs.register(source.to_owned(), result_rx).await;
         self.jobs.set_cancel_token(job_id, cancel.clone()).await;
         {
-            // This job publishes whole statement results below; a child's
-            // drain task teeing the same bytes would write them twice.
+            // This job publishes whole statement results below; a command
+            // publishing the same bytes as it runs would write them twice.
             let mut ec = fork.exec_ctx.write().await;
             ec.background_job = Some(job_id);
-            ec.background_stream_external_output = false;
+            ec.background_stream_output = false;
         }
 
         // The job token is this call's cancel input, so `JobManager::cancel`
@@ -3469,7 +3469,7 @@ impl Kernel {
             kill_children_on_parent_death: ec.kill_children_on_parent_death,
             kill_grace: ec.kill_grace,
             background_job: ec.background_job,
-            background_stream_external_output: ec.background_stream_external_output,
+            background_stream_output: ec.background_stream_output,
             aliases: ec.aliases.clone(),
             ignore_config: ec.ignore_config.clone(),
             output_limit: ec.output_limit.clone(),
@@ -4074,6 +4074,20 @@ impl Kernel {
         // The builtin's own `parsed.global.apply(ctx)` becomes idempotent.
         GlobalFlags::apply_from_args(&tool_args, raw_argv, &mut *ctx);
 
+        // A builtin's output is a value until it returns. When that output is
+        // its job's stdout, it is published after `--json` is applied below.
+        // A builtin that re-dispatched (`timeout`) publishes nothing if the
+        // command it ran already reached the stream.
+        let job_stdout = match (ctx.background_job, ctx.background_stream_output, ctx.pipeline_position) {
+            (Some(job_id), true, PipelinePosition::Only | PipelinePosition::Last) => {
+                self.jobs.streams(job_id).await.map(|streams| streams.stdout)
+            }
+            _ => None,
+        };
+        let written_before = match &job_stdout {
+            Some(stdout) => stdout.stats().await.total_written,
+            None => 0,
+        };
         let mut result = tool.execute(tool_args, &mut *ctx).await;
         // A command substitution binds `.data` only when it is the result's
         // VALUE. `--json` and the pipeline sideband read `.data` either way,
@@ -4140,6 +4154,14 @@ impl Kernel {
         // tool owns its own output (renders --json itself), in which case we
         // leave its bytes untouched.
         let result = finalize_output(result, ctx.output_format, owns_output);
+        if let Some(stdout) = job_stdout
+            && stdout.stats().await.total_written == written_before
+        {
+            match result.out_bytes() {
+                Some(bytes) => stdout.write(bytes).await,
+                None => stdout.write(result.text_out().as_bytes()).await,
+            }
+        }
 
         Ok(result)
     }
@@ -5187,6 +5209,10 @@ impl Kernel {
 
     async fn execute_block_capturing(&self, stmts: &[Stmt]) -> Result<ExecResult> {
         let _depth = self.enter_recursion("command substitution")?;
+        // Captured output is a value, not job output: nothing inside publishes
+        // to a job stream. Restored on every exit from the block below.
+        let stream_output = std::mem::replace(&mut self.exec_ctx.write().await.background_stream_output, false);
+        let outcome: Result<ExecResult> = async {
         // Accumulate stdout as raw bytes so a binary-producing statement
         // (`$(dd …)`, `$(base64 -d …)`) isn't lossy-decoded here before the
         // caller can preserve it. The final result is text iff valid UTF-8.
@@ -5258,6 +5284,10 @@ impl Kernel {
         result.data_is_value = last_data.is_some();
         result.data = last_data;
         Ok(result)
+        }
+        .await;
+        self.exec_ctx.write().await.background_stream_output = stream_output;
+        outcome
     }
 
     /// Evaluate `$(( text ))`'s content. Takes the sync fast path
@@ -6412,6 +6442,8 @@ impl Kernel {
         }
 
         // 1. Sync ctx → self internals
+        // The stream flag is per dispatch; the kernel's own value returns after.
+        let saved_stream_output;
         {
             let mut scope = self.scope.write().await;
             *scope = ctx.scope.clone();
@@ -6435,6 +6467,14 @@ impl Kernel {
             ec.ignore_config = ctx.ignore_config.clone();
             ec.output_limit = ctx.output_limit.clone();
             ec.pipeline_position = ctx.pipeline_position;
+            // A command nested in this dispatch runs as its own single-command
+            // pipeline (`Only`), so the flag, not its position, carries whether
+            // this stage's stdout is the job's stdout.
+            saved_stream_output = std::mem::replace(
+                &mut ec.background_stream_output,
+                ctx.background_stream_output
+                    && matches!(ctx.pipeline_position, PipelinePosition::Only | PipelinePosition::Last),
+            );
             ec.cancel = ctx.cancel.clone();
             ec.watchdog = ctx.watchdog.clone();
         }
@@ -6445,7 +6485,9 @@ impl Kernel {
         // the same boundary bash draws by running each stage in a subshell.
         // Whatever output the statement produced before the signal still comes
         // back and still reaches the pipe.
-        let result = match self.execute_stmt_flow(stmt).await? {
+        let flow = self.execute_stmt_flow(stmt).await;
+        self.exec_ctx.write().await.background_stream_output = saved_stream_output;
+        let result = match flow? {
             ControlFlow::Normal(result)
             | ControlFlow::Break { result, .. }
             | ControlFlow::Continue { result, .. }
@@ -6497,6 +6539,8 @@ impl Kernel {
         }
 
         // 1. Sync ctx → self internals
+        // The stream flag is per dispatch; the kernel's own value returns after.
+        let saved_stream_output;
         {
             let mut scope = self.scope.write().await;
             *scope = ctx.scope.clone();
@@ -6526,6 +6570,14 @@ impl Kernel {
             ec.ignore_config = ctx.ignore_config.clone();
             ec.output_limit = ctx.output_limit.clone();
             ec.pipeline_position = ctx.pipeline_position;
+            // A command nested in this dispatch runs as its own single-command
+            // pipeline (`Only`), so the flag, not its position, carries whether
+            // this stage's stdout is the job's stdout.
+            saved_stream_output = std::mem::replace(
+                &mut ec.background_stream_output,
+                ctx.background_stream_output
+                    && matches!(ctx.pipeline_position, PipelinePosition::Only | PipelinePosition::Last),
+            );
             // Sync the cancel token from ctx → ec. Builtins like `timeout`
             // swap ctx.cancel to a derived child token before re-dispatching;
             // execute_command's snapshot reads ec.cancel (kept aligned by
@@ -6538,7 +6590,9 @@ impl Kernel {
         }
 
         // 2. Execute via the full dispatch chain
-        let result = self.execute_command(&cmd.name, &cmd.args).await?;
+        let result = self.execute_command(&cmd.name, &cmd.args).await;
+        self.exec_ctx.write().await.background_stream_output = saved_stream_output;
+        let result = result?;
 
         // 3. Sync self → ctx
         {

crates/kaish-kernel/src/scheduler/job.rs

@@ -35,18 +35,21 @@ pub struct JobStreams {
     ///   command running for this job — but only from the stage whose stdout
     ///   *is* the job's stdout (`Only` or `Last` in the pipeline), so
     ///   `a | b` streams `b` and not `a`'s bytes on their way into `b`.
-    /// * **At completion**, from the job's captured `ExecResult`, and only
-    ///   when nothing was streamed live. That covers a builtin-only job
-    ///   (`echo hi &`): a builtin returns its output as a value when it
-    ///   finishes, so there is no byte stream to tee.
+    /// * **When a builtin returns**, from its result after `--json` is
+    ///   applied, under the same `Only`/`Last` rule. A builtin that
+    ///   re-dispatched (`timeout`) publishes only if the command it ran wrote
+    ///   nothing.
+    ///
+    /// Output with another destination is never published: a `$(...)`
+    /// capture, a stdout redirect, a scatter worker's stdout.
     ///
     /// Whichever fed it, the stream is closed once the job's result is in
     /// ([`JobManager::finalize_streams`]), so a reader can tell "no more
     /// coming" from "nothing yet".
     pub stdout: Arc<BoundedStream>,
-    /// The job's stderr. Same two feeds as [`Self::stdout`], except the live
-    /// one takes **every** stage's stderr, not just the last — stderr is not
-    /// piped between stages. The consequence, stated rather than papered
+    /// The job's stderr. Fed live per chunk by external commands from
+    /// **every** stage — stderr is not piped between stages — and at
+    /// completion from the job's captured `err` when nothing arrived live. The consequence, stated rather than papered
     /// over: in a job mixing builtins and externals, once any external has
     /// written stderr the completion write is skipped, so a builtin stage's
     /// stderr stays in the job's `ExecResult` and does not reach this stream.
@@ -705,15 +708,12 @@ impl JobManager {
         Some(stream.read().await)
     }
 
-    /// Close out a finished job's streams: write the captured result into a
-    /// stream that received nothing live, then close both.
+    /// Close a finished job's streams.
     ///
-    /// The conditional is the no-double-write rule. A stream with live bytes
-    /// in it already holds exactly what the child emitted; writing
-    /// `result.text_out()` on top would repeat all of it. A stream with no
-    /// live bytes belongs to a job with nothing to tee — a builtin returns
-    /// its output as a value, not as a pipe — and would otherwise read empty
-    /// forever.
+    /// stdout is never written here. Every command whose output is the job's
+    /// stdout published it while running, and a whole-program job publishes
+    /// each statement; writing the captured result on top would repeat it.
+    /// stderr takes the captured `err` only when nothing reached it live.
     ///
     /// Called by the background task that owns the job, before it hands the
     /// result over, so a reader that sees a terminal `status` also sees a
@@ -723,14 +723,6 @@ impl JobManager {
             return;
         };
 
-        if streams.stdout.stats().await.total_written == 0 {
-            // Raw bytes when the payload is binary; `text_out` would decode it
-            // lossily and corrupt what a caller reads back out of the node.
-            match result.out_bytes() {
-                Some(bytes) => streams.stdout.write(bytes).await,
-                None => streams.stdout.write(result.text_out().as_bytes()).await,
-            }
-        }
         if streams.stderr.stats().await.total_written == 0 {
             streams.stderr.write(result.err.as_bytes()).await;
         }

crates/kaish-kernel/src/scheduler/pipeline.rs

@@ -89,6 +89,17 @@ fn fault_result(error: anyhow::Error) -> ExecResult {
     result
 }
 
+/// Whether a stage's redirects send its stdout away from its pipeline
+/// position: to a file (`>`, `>>`, `&>`) or to stderr (`>&2`).
+fn redirects_stdout(stage: &PipelineStage) -> bool {
+    stage.redirects().iter().any(|redirect| {
+        matches!(
+            redirect.kind,
+            RedirectKind::StdoutOverwrite | RedirectKind::StdoutAppend | RedirectKind::Both | RedirectKind::MergeStdout
+        )
+    })
+}
+
 /// Apply redirects to an execution result.
 ///
 /// Pre-execution redirects (Stdin, HereDoc) should be handled before calling.
@@ -605,11 +616,18 @@ impl PipelineRunner {
         // Set pipeline position for stdio inheritance decisions
         ctx.pipeline_position = PipelinePosition::Only;
 
+        // A redirected stdout goes to its target, not to a job's stream.
+        let stream_output = ctx.background_stream_output;
+        if redirects_stdout(stage) {
+            ctx.background_stream_output = false;
+        }
+
         // Execute via dispatcher (full resolution chain)
         let result = match dispatch_stage(stage, ctx, dispatcher).await {
             Ok(result) => result,
             Err(e) => fault_result(e),
         };
+        ctx.background_stream_output = stream_output;
 
         // Apply post-execution redirects
         apply_redirects(result, stage.redirects(), ctx, dispatcher).await
@@ -755,6 +773,11 @@ impl PipelineRunner {
                 // dropped structured data (`seq 1 3 | jq .` → text → parse error).
                 stage_ctx.stdin_data_rx = data_receiver;
 
+                // A redirected stdout goes to its target, not to a job's stream.
+                if redirects_stdout(&stage) {
+                    stage_ctx.background_stream_output = false;
+                }
+
                 // Execute the stage
                 let mut result = match dispatch_stage(&stage, &mut stage_ctx, &*task_dispatcher).await {
                     Ok(result) => result,

crates/kaish-kernel/src/scheduler/scatter.rs

@@ -172,7 +172,10 @@ impl ScatterGatherRunner {
             };
             (text, data)
         } else {
+            // The stages before `scatter` produce its input, not job output.
+            let stream_output = std::mem::replace(&mut ctx.background_stream_output, false);
             let mut result = runner.run_sequential(pre_scatter, ctx, &*self.sequential_dispatcher).await;
+            ctx.background_stream_output = stream_output;
             // GH #250: `run_sequential` never applies the output-limit spill
             // check or the `did_spill` -> exit-3 remap
             // (`output_limit::apply_spill_contract`) — that seam only
@@ -233,6 +236,9 @@ impl ScatterGatherRunner {
         // Run post-gather commands if any. A failed gather short-circuits —
         // feeding partial/failed output onward would propagate corruption.
         if post_gather.is_empty() || gathered.code != 0 {
+            // gather's rows are built here rather than by a dispatched
+            // command, so nothing else publishes them to a job stream.
+            ctx.publish_job_stdout(&gathered).await;
             gathered
         } else {
             ctx.set_stdin_with_data(
@@ -293,6 +299,8 @@ impl ScatterGatherRunner {
             // `'static`), so the child MUST be built here and MOVED into the
             // spawn — it cannot be constructed inside the closure.
             let mut worker_ctx = base_ctx.child_for_pipeline();
+            // A worker's stdout is gather's input, not job output.
+            worker_ctx.background_stream_output = false;
             // Per-worker TYPED binding — the same json→Value conversion the
             // for-loop uses for `$(cmd)` items (GH #73), so a record element
             // subscripts as `${ITEM[k]}`.

crates/kaish-kernel/src/spawn.rs

@@ -121,8 +121,8 @@ pub(crate) struct SpawnContext {
     pub job_manager: Option<Arc<JobManager>>,
     /// The background job this command runs for, if any.
     pub background_job: Option<JobId>,
-    /// Whether a child process writes its output directly to the job stream.
-    pub background_stream_external_output: bool,
+    /// Whether this command's output is its background job's output.
+    pub background_stream_output: bool,
 }
 
 impl SpawnContext {
@@ -135,7 +135,7 @@ impl SpawnContext {
             pipeline_position: ctx.pipeline_position,
             job_manager: ctx.job_manager.clone(),
             background_job: ctx.background_job,
-            background_stream_external_output: ctx.background_stream_external_output,
+            background_stream_output: ctx.background_stream_output,
         }
     }
 }
@@ -328,7 +328,7 @@ pub(crate) async fn spawn_process(request: SpawnRequest, spawn_ctx: &SpawnContex
     let job_streams = match (
         &spawn_ctx.job_manager,
         spawn_ctx.background_job,
-        spawn_ctx.background_stream_external_output,
+        spawn_ctx.background_stream_output,
     ) {
         (Some(jobs), Some(job_id), true) => jobs.streams(job_id).await,
         _ => None,