diff --git a/CHANGELOG.md b/CHANGELOG.md index a70e13c1..ec418ad2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/kaish-help/content/en/vfs.md b/crates/kaish-help/content/en/vfs.md index 595086fd..859038e4 100644 --- a/crates/kaish-help/content/en/vfs.md +++ b/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 diff --git a/crates/kaish-kernel/src/kernel.rs b/crates/kaish-kernel/src/kernel.rs index cd82cad6..6d19c79e 100644 --- a/crates/kaish-kernel/src/kernel.rs +++ b/crates/kaish-kernel/src/kernel.rs @@ -1450,7 +1450,7 @@ impl Kernel { job_id: crate::scheduler::JobId, ) -> Arc { 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 { 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 = 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 { diff --git a/crates/kaish-kernel/src/scheduler/job.rs b/crates/kaish-kernel/src/scheduler/job.rs index d390d3f1..e0f83e2b 100644 --- a/crates/kaish-kernel/src/scheduler/job.rs +++ b/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, - /// 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; } diff --git a/crates/kaish-kernel/src/scheduler/pipeline.rs b/crates/kaish-kernel/src/scheduler/pipeline.rs index 242ca61d..ded0542b 100644 --- a/crates/kaish-kernel/src/scheduler/pipeline.rs +++ b/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, diff --git a/crates/kaish-kernel/src/scheduler/scatter.rs b/crates/kaish-kernel/src/scheduler/scatter.rs index 0e8ef35c..28e1b0d5 100644 --- a/crates/kaish-kernel/src/scheduler/scatter.rs +++ b/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]}`. diff --git a/crates/kaish-kernel/src/spawn.rs b/crates/kaish-kernel/src/spawn.rs index 7b73dbc5..cc966af5 100644 --- a/crates/kaish-kernel/src/spawn.rs +++ b/crates/kaish-kernel/src/spawn.rs @@ -121,8 +121,8 @@ pub(crate) struct SpawnContext { pub job_manager: Option>, /// The background job this command runs for, if any. pub background_job: Option, - /// 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, diff --git a/crates/kaish-kernel/src/tools/context.rs b/crates/kaish-kernel/src/tools/context.rs index 9e6430eb..49ac7d5c 100644 --- a/crates/kaish-kernel/src/tools/context.rs +++ b/crates/kaish-kernel/src/tools/context.rs @@ -215,10 +215,12 @@ pub struct ExecContext { /// job for `kill - %N` and tees its output into the job's streams. /// `None` for foreground execution. pub background_job: Option, - /// Whether external child output is copied directly into the background - /// job's streams. Shell `&` jobs use this for chunk-live output; the - /// whole-program background API writes complete statement results itself. - pub background_stream_external_output: bool, + /// Whether this command's stdout is its background job's stdout, so its + /// output is published to the job's stream: an external per chunk, a + /// builtin when it returns. False inside `$(...)`, under a stdout + /// redirect, in a scatter worker, and for whole-program jobs, which + /// publish complete statement results themselves. + pub background_stream_output: bool, /// Command aliases (name → expansion string). pub aliases: HashMap, /// Ignore file configuration for file-walking tools. @@ -421,6 +423,27 @@ fn concurrent_change_error(resolved: &Path) -> crate::backend::BackendError { } impl ExecContext { + /// Publish `result`'s stdout to this context's background job, when that + /// stdout is the job's stdout. For output built outside a dispatched + /// command, such as gather's rows; a dispatched command publishes its own. + pub(crate) async fn publish_job_stdout(&self, result: &ExecResult) { + let (Some(job_id), true, PipelinePosition::Only | PipelinePosition::Last, Some(jobs)) = ( + self.background_job, + self.background_stream_output, + self.pipeline_position, + self.job_manager.as_ref(), + ) else { + return; + }; + let Some(streams) = jobs.streams(job_id).await else { + return; + }; + match result.out_bytes() { + Some(bytes) => streams.stdout.write(bytes).await, + None => streams.stdout.write(result.text_out().as_bytes()).await, + } + } + /// Create a new execution context with a VFS (uses LocalBackend without tools). /// /// This constructor is for backward compatibility and tests that don't need tool dispatch. @@ -445,7 +468,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, - background_stream_external_output: false, + background_stream_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -487,7 +510,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, - background_stream_external_output: false, + background_stream_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -526,7 +549,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, - background_stream_external_output: false, + background_stream_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -565,7 +588,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, - background_stream_external_output: false, + background_stream_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -607,7 +630,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, - background_stream_external_output: false, + background_stream_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -646,7 +669,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, - background_stream_external_output: false, + background_stream_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -943,7 +966,7 @@ impl ExecContext { kill_children_on_parent_death: self.kill_children_on_parent_death, kill_grace: self.kill_grace, background_job: self.background_job, - background_stream_external_output: self.background_stream_external_output, + background_stream_output: self.background_stream_output, aliases: self.aliases.clone(), ignore_config: self.ignore_config.clone(), output_limit: self.output_limit.clone(), diff --git a/crates/kaish-kernel/tests/job_live_output_tests.rs b/crates/kaish-kernel/tests/job_live_output_tests.rs index 6bb426c3..8d617d4c 100644 --- a/crates/kaish-kernel/tests/job_live_output_tests.rs +++ b/crates/kaish-kernel/tests/job_live_output_tests.rs @@ -297,3 +297,64 @@ async fn reading_an_unknown_job_is_none() { assert!(kernel.jobs().read_stdout(JobId(99)).await.is_none()); assert!(kernel.jobs().read_stderr(JobId(99)).await.is_none()); } + +// ── Routing: only output that is the job's own stdout reaches the stream ── +// +// The live tee decides at spawn time, before a capture or redirect takes the +// bytes, and the completion write fills a stream only when nothing streamed. +// Each case below mixes a builtin with an external, or gives the external's +// output a destination other than the job's stdout. + +/// Run `program` as job 1 to completion and return its stdout stream. +async fn job_stdout(kernel: &Kernel, program: &str) -> String { + kernel.execute(program).await.expect("spawn failed"); + let id = JobId(1); + assert_eq!(wait_done(kernel, id).await, "done:0"); + stdout_of(kernel, id).await +} + +#[tokio::test] +async fn builtin_and_external_output_both_reach_the_stream_in_order() { + let kernel = kernel(); + let out = job_stdout( + &kernel, + "if true; then echo builtin-a; sh -c 'echo external-b'; echo builtin-c; fi &", + ) + .await; + assert_eq!(out, "builtin-a\nexternal-b\nbuiltin-c\n"); +} + +#[tokio::test] +async fn captured_external_output_is_not_job_output() { + let kernel = kernel(); + let out = job_stdout( + &kernel, + "if true; then x=$(sh -c 'echo captured'); echo \"got $x\"; fi &", + ) + .await; + assert_eq!(out, "got captured\n"); +} + +#[tokio::test] +async fn redirected_external_output_is_not_job_output() { + let kernel = kernel(); + let path = std::env::temp_dir().join(format!("kaish-job-redirect-{}.txt", std::process::id())); + let program = format!( + "if true; then sh -c 'echo to-file' > {}; echo after; fi &", + path.display() + ); + let out = job_stdout(&kernel, &program).await; + let written = std::fs::read_to_string(&path).expect("redirect target written"); + std::fs::remove_file(&path).expect("remove redirect target"); + assert_eq!(written, "to-file\n", "the redirect must still receive the bytes"); + assert_eq!(out, "after\n"); +} + +#[tokio::test] +async fn scatter_worker_output_is_not_job_output() { + let kernel = kernel(); + let out = job_stdout(&kernel, "seq 1 2 | scatter | sh -c 'echo worker' | gather &").await; + assert!(!out.lines().any(|line| line == "worker"), "a worker's raw stdout leaked: {out:?}"); + assert_eq!(out.lines().count(), 2, "one gather record per worker: {out:?}"); + assert!(out.contains("\"out\":\"worker\""), "gather's records must reach the stream: {out:?}"); +} diff --git a/docs/EMBEDDING.md b/docs/EMBEDDING.md index b6d85c40..f4252b86 100644 --- a/docs/EMBEDDING.md +++ b/docs/EMBEDDING.md @@ -1494,23 +1494,25 @@ cat /v/jobs/1/stdout # Whatever the build has printed so far `stdout` and `stderr` are live for an **external** command run by the job: its drain task tees each 8 KiB chunk into the node as the child emits it. -GH #240 had removed both nodes because they filled only once, at completion, -while four docs promised a live stream — they are back on the terms the docs -always claimed. +A builtin publishes its stdout when it returns, so a job mixing both reads +in order. GH #240 had removed both nodes because they filled only once, at +completion, while four docs promised a live stream — they are back on the +terms the docs always claimed. Three limits, stated because an embedder polling these needs to predict them: - **A builtin is not a live producer.** A kaish builtin returns its whole - output as a value when it finishes, so `echo hi &` fills the node in one - write at completion — and so does `cargo build 2>&1 | tee build.log &`, + output as a value when it finishes, so `echo hi &` writes the node once, + when `echo` returns — and so does `cargo build 2>&1 | tee build.log &`, because kaish's `tee` is a builtin. Drop the `| tee`: the job's own stream *is* the log. -- **Only the last stage of a pipeline reaches `stdout`.** An upstream stage's - output is the next stage's stdin, not the job's stdout. `stderr` takes every - stage's, since stderr is not piped. One consequence: 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 the node. +- **Only the job's own stdout reaches `stdout`.** An upstream stage's output + is the next stage's stdin, `$(...)` output is a value, a redirected stdout + goes to its target, and a scatter worker's stdout is gather's input; none + of it is published. `stderr` takes every external stage's stderr live, + since stderr is not piped. A builtin's stderr reaches the node at + completion, and only when no external wrote stderr first; otherwise it + stays in the job's `ExecResult`. - **Each node is a 10 MB ring** that evicts its oldest bytes. A job that outruns it loses its head, not its tail; redirect to a file (`cmd > /tmp/out.log &`) when the whole output matters.