From 2b18072fd08a01042afc421f660c3747ce57360b Mon Sep 17 00:00:00 2001 From: A Tobey Date: Sat, 12 Sep 2026 16:36:25 -0400 Subject: [PATCH] fix: a job's stdout stream holds only the job's own output A background job's stdout stream was fed two ways: external commands teed each chunk when their pipeline position was Only or Last, and the job's captured result filled the stream at completion only when nothing had streamed. Position alone could not see where output was going, and the completion rule lost everything once any external had written: if true; then echo a; sh -c 'echo b'; echo c; fi & -> "b" if true; then x=$(sh -c 'echo captured'); echo "got $x"; fi & -> "captured" if true; then sh -c 'echo to-file' > f; echo after; fi & -> "to-file" seq 1 2 | scatter | sh -c 'echo worker' | gather & -> "worker\nworker" The routing decision now travels as `background_stream_output`, renamed from `background_stream_external_output` because it governs builtins too. It is off inside `$(...)`, for a stage that redirects stdout, for scatter workers and the stages before `scatter`, and for a compound stage that is not Only or Last, whose inner commands otherwise run as Only. Each dispatch sets it for its own command and restores the kernel's value afterward, so one command's setting does not leak into the next. A builtin publishes its output when it returns, after `--json` is applied, unless the command it re-dispatched already reached the stream (`timeout 5 echo hi &`). gather publishes its rows, which no dispatched command produces. `finalize_streams` no longer writes stdout: every producer publishes as it runs, and a completion write would only repeat or hide a routing hole. stderr keeps its completion rule. Co-Authored-By: DeepSeek V4 Flash Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 4 ++ crates/kaish-help/content/en/vfs.md | 11 +-- crates/kaish-kernel/src/kernel.rs | 68 +++++++++++++++++-- crates/kaish-kernel/src/scheduler/job.rs | 38 ++++------- crates/kaish-kernel/src/scheduler/pipeline.rs | 23 +++++++ crates/kaish-kernel/src/scheduler/scatter.rs | 8 +++ crates/kaish-kernel/src/spawn.rs | 8 +-- crates/kaish-kernel/src/tools/context.rs | 45 +++++++++--- .../tests/job_live_output_tests.rs | 61 +++++++++++++++++ docs/EMBEDDING.md | 24 ++++--- 10 files changed, 229 insertions(+), 61 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 917dedb7..5651c6c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,10 @@ breaking entries are marked **BREAKING**. ### Fixed +- 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 59fe5fe1..b58518d8 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 } @@ -1892,11 +1892,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` @@ -3411,7 +3411,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(), @@ -4016,6 +4016,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, @@ -4082,6 +4096,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) } @@ -5127,6 +5149,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. @@ -5188,6 +5214,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 @@ -6338,6 +6368,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(); @@ -6361,6 +6393,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(); } @@ -6371,7 +6411,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, .. } @@ -6423,6 +6465,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(); @@ -6452,6 +6496,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 @@ -6464,7 +6516,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 03a182d9..a9c4ea4a 100644 --- a/crates/kaish-kernel/src/scheduler/pipeline.rs +++ b/crates/kaish-kernel/src/scheduler/pipeline.rs @@ -74,6 +74,17 @@ fn finalize_scatter_gather_error(result: ExecResult, format: Option`, `>>`, `&>`) 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. @@ -590,11 +601,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) => ExecResult::failure(1, e.to_string()), }; + ctx.background_stream_output = stream_output; // Apply post-execution redirects apply_redirects(result, stage.redirects(), ctx, dispatcher).await @@ -740,6 +758,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 4b03d06c..7c062ba7 100644 --- a/docs/EMBEDDING.md +++ b/docs/EMBEDDING.md @@ -1491,23 +1491,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.