From fb62bb83a96d36bba775476907ee1249e5805ab2 Mon Sep 17 00:00:00 2001 From: A Tobey Date: Sun, 13 Sep 2026 08:58:47 -0400 Subject: [PATCH 1/2] fix: whole-program jobs stream stdout as it is produced A job started with `Kernel::execute_background_with_options` published output only when each top-level statement finished. A `for` loop or a single `cargo build` showed nothing in `/v/jobs/N/stdout` until it ended, while the same command as `cmd &` streamed live after #449. A shell user expects to see output as it is produced. One flag governed both streams, so turning stdout streaming on for these jobs would also turn on the external stderr tee, and the per-statement writer would write that stderr a second time. The stderr tee now has its own per-job flag, `background_stream_stderr`: true for `cmd &`, false for a whole-program job. spawn.rs tees stderr only when both flags are on, so `cmd &` behaves as before. A whole-program job turns stdout streaming on, and its writer carries only each statement's stderr and the terminal diagnostic. Stderr stays per statement here. Live stderr for every job lands after the execution context is threaded through the interpreter (GH #369). The comment and `JobStreams::stderr` docs that said every pipeline stage tees stderr were wrong: First and Middle stages stream nothing. Both now say where stderr tees. Streaming stdout exposed a routing hole #449 left open. An embedder tool (`backend.call_tool`) returned its result without publishing it, and nothing else published stdout once #449 dropped the completion write, so `embedder_tool &` left `/v/jobs/N/stdout` empty. The whole-program writer had hidden this by publishing whole statement results. The backend-tool arm now publishes through `ExecContext::publish_job_stdout`, the helper gather's rows use. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 7 +- crates/kaish-kernel/src/kernel.rs | 123 +++++++++++++----- crates/kaish-kernel/src/scheduler/job.rs | 15 ++- crates/kaish-kernel/src/spawn.rs | 16 ++- crates/kaish-kernel/src/tools/context.rs | 19 ++- .../tests/job_live_output_tests.rs | 72 +++++++++- docs/EMBEDDING.md | 10 +- 7 files changed, 210 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ec418ad2..b4a76cfd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,11 +14,14 @@ breaking entries are marked **BREAKING**. - **`Kernel::execute_background_with_options`** — run a whole program as a job and get its `JobId`; a program that fails to parse or validate - registers no job. Output reaches the job streams after each top-level - statement. + registers no job. Stdout streams as the program runs; stderr reaches the + job stream after each top-level statement. ### Fixed +- An embedder tool's stdout now reaches its background job's stdout stream. + `embedder_tool &` left `/v/jobs/N/stdout` empty; the output was only in + the job's result. - A streaming caller (`kaish -c`, `execute_with_options_streaming`) now receives stderr drained during an `exit` statement and the watchdog's timeout line. The timeout line follows any stderr the program wrote diff --git a/crates/kaish-kernel/src/kernel.rs b/crates/kaish-kernel/src/kernel.rs index 6d19c79e..b30791d7 100644 --- a/crates/kaish-kernel/src/kernel.rs +++ b/crates/kaish-kernel/src/kernel.rs @@ -1450,7 +1450,11 @@ 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_output = true; + { + let mut ec = fork.exec_ctx.write().await; + ec.background_stream_output = true; + ec.background_stream_stderr = true; + } fork } @@ -1857,11 +1861,12 @@ impl Kernel { /// job. The fork runs the source with the same [`ExecuteOptions`] /// semantics as [`Self::execute_with_options`]. /// - /// Output reaches the job's streams after each top-level statement - /// finishes, not while a statement runs. A runtime error (exit 1), timeout - /// (exit 124), or cancellation (exit 130) ends the job's stderr with one - /// diagnostic, in the result and in the stream. `JobManager::cancel` and - /// `opts.cancel_token` both cancel the job. + /// Stdout reaches the job's stream as commands produce it, as in a `cmd &` + /// job: an external per chunk, a builtin when it returns. Stderr reaches + /// the stream after each top-level statement finishes. A runtime error + /// (exit 1), timeout (exit 124), or cancellation (exit 130) ends the job's + /// stderr with one diagnostic, in the result and in the stream. + /// `JobManager::cancel` and `opts.cancel_token` both cancel the job. /// /// # Errors /// @@ -1893,11 +1898,13 @@ 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 command - // publishing the same bytes as it runs would write them twice. + // Stdout streams as commands produce it, like a `cmd &` job. + // Stderr is written per statement below, so an external's stderr + // must not also tee live. let mut ec = fork.exec_ctx.write().await; ec.background_job = Some(job_id); - ec.background_stream_output = false; + ec.background_stream_output = true; + ec.background_stream_stderr = false; } // The job token is this call's cancel input, so `JobManager::cancel` @@ -1906,19 +1913,16 @@ impl Kernel { let embedder_cancel = opts.cancel_token.replace(cancel.clone()); let jobs = self.jobs.clone(); - let (output_tx, mut output_rx) = mpsc::unbounded_channel::(); + let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); - let output_jobs = jobs.clone(); - let output_writer = tokio::spawn(crate::telemetry::bind_current_context(async move { - let Some(streams) = output_jobs.streams(job_id).await else { + // Stdout reaches the job's stream from the commands themselves. + let stderr_jobs = jobs.clone(); + let stderr_writer = tokio::spawn(crate::telemetry::bind_current_context(async move { + let Some(streams) = stderr_jobs.streams(job_id).await else { return; }; - while let Some(output) = output_rx.recv().await { - match output.out_bytes() { - Some(bytes) => streams.stdout.write(bytes).await, - None => streams.stdout.write(output.text_out().as_bytes()).await, - } - streams.stderr.write(output.err.as_bytes()).await; + while let Some(err) = stderr_rx.recv().await { + streams.stderr.write(err.as_bytes()).await; } })); @@ -1938,8 +1942,8 @@ impl Kernel { accumulate_result(&mut streamed, output); // Unbounded: this synchronous callback cannot await a bounded // channel without dropping output or blocking execution. - if output_tx.send(output.clone()).is_err() { - tracing::error!(job_id = %job_id, "background job output writer stopped before execution completed"); + if !output.err.is_empty() && stderr_tx.send(output.err.clone()).is_err() { + tracing::error!(job_id = %job_id, "background job stderr writer stopped before execution completed"); } }; // `run_inner` merges it into an Ok result only. @@ -1977,22 +1981,22 @@ impl Kernel { unstreamed_err.push_str(&diagnostic); } if !unstreamed_err.is_empty() { - let mut tail = ExecResult::success(""); + let mut tail = String::new(); if stream_needs_newline { - tail.err.push('\n'); + tail.push('\n'); } - tail.err.push_str(&unstreamed_err); - if output_tx.send(tail).is_err() { - tracing::error!(job_id = %job_id, "background job output writer stopped before its final diagnostic"); + tail.push_str(&unstreamed_err); + if stderr_tx.send(tail).is_err() { + tracing::error!(job_id = %job_id, "background job stderr writer stopped before its final diagnostic"); } } - drop(output_tx); + drop(stderr_tx); - if let Err(error) = output_writer.await { + if let Err(error) = stderr_writer.await { result.code = 1; push_diagnostic( &mut result.err, - &ExecResult::terminate_diagnostic(format!("background job output writer failed: {error}")), + &ExecResult::terminate_diagnostic(format!("background job stderr writer failed: {error}")), ); } jobs.finalize_streams(job_id, &result).await; @@ -3470,6 +3474,7 @@ impl Kernel { kill_grace: ec.kill_grace, background_job: ec.background_job, background_stream_output: ec.background_stream_output, + background_stream_stderr: ec.background_stream_stderr, aliases: ec.aliases.clone(), ignore_config: ec.ignore_config.clone(), output_limit: ec.output_limit.clone(), @@ -3960,6 +3965,10 @@ impl Kernel { || tool_schema .as_ref() .is_some_and(|s| s.typed_substitution); + drop(scope); + // No builtin or external command produced this output, + // so nothing else publishes it to a background job. + ctx.publish_job_stdout(&result).await; return Ok(result); } Err(BackendError::ToolNotFound(_)) => { @@ -10393,6 +10402,62 @@ AFTER="yes"'"#) ); } + #[tokio::test] + async fn background_job_publishes_custom_tool_stdout() { + use crate::backend::testing::MockBackend; + use crate::backend::ToolResult; + + let jobs = Arc::new(JobManager::new()); + let (mock, calls) = MockBackend::new(); + let backend = mock.with_tool_result(|name| Ok(ToolResult::success(format!("tool:{name}\n")))); + let kernel = Kernel::with_backend( + Arc::new(backend), + KernelConfig::isolated().with_job_manager(jobs.clone()), + |_| {}, + |_| {}, + ) + .expect("kernel"); + + kernel.execute("embedder_tool &").await.expect("spawn"); + let id = crate::scheduler::JobId(1); + let result = jobs.wait(id).await.expect("job result"); + assert!(result.ok(), "background job failed: {result:?}"); + assert_eq!(calls.load(Ordering::SeqCst), 1, "custom tool must run in the fork"); + assert_eq!( + String::from_utf8(jobs.read_stdout(id).await.expect("stdout stream")).expect("utf8"), + "tool:embedder_tool\n", + "a custom tool's stdout is the job's stdout" + ); + } + + #[tokio::test] + async fn background_job_publishes_redispatched_custom_tool_stdout_once() { + use crate::backend::testing::MockBackend; + use crate::backend::ToolResult; + + let jobs = Arc::new(JobManager::new()); + let (mock, calls) = MockBackend::new(); + let backend = mock.with_tool_result(|name| Ok(ToolResult::success(format!("tool:{name}\n")))); + let kernel = Kernel::with_backend( + Arc::new(backend), + KernelConfig::isolated().with_job_manager(jobs.clone()), + |_| {}, + |_| {}, + ) + .expect("kernel"); + + kernel.execute("timeout 5 embedder_tool &").await.expect("spawn"); + let id = crate::scheduler::JobId(1); + let result = jobs.wait(id).await.expect("job result"); + assert!(result.ok(), "background job failed: {result:?}"); + assert_eq!(calls.load(Ordering::SeqCst), 1, "timeout must run the custom tool once"); + assert_eq!( + String::from_utf8(jobs.read_stdout(id).await.expect("stdout stream")).expect("utf8"), + "tool:embedder_tool\n", + "timeout publishes nothing when the tool it ran already published" + ); + } + #[tokio::test] async fn background_program_cancel_reaches_running_execution() { let jobs = Arc::new(JobManager::new()); diff --git a/crates/kaish-kernel/src/scheduler/job.rs b/crates/kaish-kernel/src/scheduler/job.rs index e0f83e2b..6bb6bab5 100644 --- a/crates/kaish-kernel/src/scheduler/job.rs +++ b/crates/kaish-kernel/src/scheduler/job.rs @@ -47,12 +47,15 @@ pub struct JobStreams { /// ([`JobManager::finalize_streams`]), so a reader can tell "no more /// coming" from "nothing yet". pub stdout: Arc, - /// 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. + /// The job's stderr. For a `cmd &` job, fed live per chunk by external + /// commands in stages that stream stdout, and at completion from the + /// job's captured `err` when nothing arrived live. 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. + /// + /// A whole-program job writes each top-level statement's stderr when the + /// statement finishes. pub stderr: Arc, } diff --git a/crates/kaish-kernel/src/spawn.rs b/crates/kaish-kernel/src/spawn.rs index cc966af5..f0592039 100644 --- a/crates/kaish-kernel/src/spawn.rs +++ b/crates/kaish-kernel/src/spawn.rs @@ -121,8 +121,10 @@ pub(crate) struct SpawnContext { pub job_manager: Option>, /// The background job this command runs for, if any. pub background_job: Option, - /// Whether this command's output is its background job's output. + /// Whether this command's stdout is its background job's stdout. pub background_stream_output: bool, + /// Whether this command's stderr also tees into its job's stderr stream. + pub background_stream_stderr: bool, } impl SpawnContext { @@ -136,6 +138,7 @@ impl SpawnContext { job_manager: ctx.job_manager.clone(), background_job: ctx.background_job, background_stream_output: ctx.background_stream_output, + background_stream_stderr: ctx.background_stream_stderr, } } } @@ -543,11 +546,11 @@ pub(crate) async fn spawn_process(request: SpawnRequest, spawn_ctx: &SpawnContex let stdout_clone = stdout_stream.clone(); let stderr_clone = stderr_stream.clone(); - // Only the stage whose stdout *is* the job's stdout tees: in + // Only the stage whose stdout *is* the job's stdout tees stdout: in // `a | b`, `a`'s bytes are `b`'s stdin, and teeing them would put // the pipeline's intermediate data into the node alongside its - // real output. stderr has no such routing — every stage's stderr - // is the job's stderr — so it tees from any position. + // real output. stderr tees wherever this command streams, unless + // its job writes stderr per statement. let stdout_tee = job_streams.as_ref().and_then(|s| { matches!( spawn_ctx.pipeline_position, @@ -555,7 +558,10 @@ pub(crate) async fn spawn_process(request: SpawnRequest, spawn_ctx: &SpawnContex ) .then(|| s.stdout.clone()) }); - let stderr_tee = job_streams.as_ref().map(|s| s.stderr.clone()); + let stderr_tee = job_streams + .as_ref() + .filter(|_| spawn_ctx.background_stream_stderr) + .map(|s| s.stderr.clone()); let stdout_task = stdout_pipe.map(|pipe| { tokio::spawn(async move { diff --git a/crates/kaish-kernel/src/tools/context.rs b/crates/kaish-kernel/src/tools/context.rs index 49ac7d5c..4a82ca5f 100644 --- a/crates/kaish-kernel/src/tools/context.rs +++ b/crates/kaish-kernel/src/tools/context.rs @@ -218,9 +218,13 @@ pub struct ExecContext { /// 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. + /// redirect, and in a scatter worker. pub background_stream_output: bool, + /// Whether an external command also tees its stderr into the job's stderr + /// stream wherever `background_stream_output` is on. Set once per job: + /// true for `cmd &`, false for a whole-program job, which writes each + /// statement's stderr when the statement finishes. + pub background_stream_stderr: bool, /// Command aliases (name → expansion string). pub aliases: HashMap, /// Ignore file configuration for file-walking tools. @@ -424,8 +428,8 @@ 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. + /// stdout is the job's stdout. For output no builtin or external command + /// produced: gather's rows and an embedder tool's result. pub(crate) async fn publish_job_stdout(&self, result: &ExecResult) { let (Some(job_id), true, PipelinePosition::Only | PipelinePosition::Last, Some(jobs)) = ( self.background_job, @@ -469,6 +473,7 @@ impl ExecContext { kill_grace: DEFAULT_KILL_GRACE, background_job: None, background_stream_output: false, + background_stream_stderr: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -511,6 +516,7 @@ impl ExecContext { kill_grace: DEFAULT_KILL_GRACE, background_job: None, background_stream_output: false, + background_stream_stderr: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -550,6 +556,7 @@ impl ExecContext { kill_grace: DEFAULT_KILL_GRACE, background_job: None, background_stream_output: false, + background_stream_stderr: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -589,6 +596,7 @@ impl ExecContext { kill_grace: DEFAULT_KILL_GRACE, background_job: None, background_stream_output: false, + background_stream_stderr: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -631,6 +639,7 @@ impl ExecContext { kill_grace: DEFAULT_KILL_GRACE, background_job: None, background_stream_output: false, + background_stream_stderr: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -670,6 +679,7 @@ impl ExecContext { kill_grace: DEFAULT_KILL_GRACE, background_job: None, background_stream_output: false, + background_stream_stderr: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -967,6 +977,7 @@ impl ExecContext { kill_grace: self.kill_grace, background_job: self.background_job, background_stream_output: self.background_stream_output, + background_stream_stderr: self.background_stream_stderr, 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 8d617d4c..23b021d8 100644 --- a/crates/kaish-kernel/tests/job_live_output_tests.rs +++ b/crates/kaish-kernel/tests/job_live_output_tests.rs @@ -21,7 +21,7 @@ use std::time::{Duration, Instant}; use kaish_kernel::ast::Value; use kaish_kernel::scheduler::JobId; -use kaish_kernel::{Kernel, KernelConfig}; +use kaish_kernel::{ExecuteOptions, Kernel, KernelConfig}; /// The execution core is hermetic — it never reads OS env — so PATH comes in /// through `initial_vars`, exactly as the REPL frontend supplies it. @@ -358,3 +358,73 @@ async fn scatter_worker_output_is_not_job_output() { 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:?}"); } + +// ── Whole-program jobs stream stdout like `&` jobs ── +// +// A shell user sees output as it is produced. Each liveness case emits, +// waits, and emits again; the first token must be readable while the job +// still runs and before the second arrives. + +/// Poll until `first` is in the job's stdout. The status is sampled before the +/// stream, so `running` plus `first` without `second` is a live write, not a +/// completion-time dump. +async fn assert_stdout_live(kernel: &Kernel, id: JobId, first: &str, second: &str) { + let deadline = Instant::now() + LIVE_TIMEOUT; + loop { + let status = status_of(kernel, id).await; + let out = stdout_of(kernel, id).await; + if out.contains(first) { + assert_eq!(status, "running", "stdout held {first:?} only after the job finished: {out:?}"); + assert!(!out.contains(second), "stdout arrived as one buffer: {out:?}"); + return; + } + assert_eq!(status, "running", "job finished before {first:?} reached stdout: {out:?}"); + assert!(Instant::now() < deadline, "{first:?} never reached stdout while the job ran"); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +async fn start_program(kernel: &Kernel, program: &str) -> JobId { + kernel + .execute_background_with_options(program, ExecuteOptions::new()) + .await + .expect("program rejected") +} + +#[tokio::test] +async fn whole_program_external_stdout_is_live() { + let kernel = kernel(); + let id = start_program(&kernel, "sh -c 'echo first; sleep 2; echo second'").await; + assert_stdout_live(&kernel, id, "first", "second").await; + assert_eq!(wait_done(&kernel, id).await, "done:0"); + assert_eq!(stdout_of(&kernel, id).await, "first\nsecond\n"); +} + +#[tokio::test] +async fn whole_program_builtin_output_inside_a_loop_is_live() { + let kernel = kernel(); + let id = start_program(&kernel, "for i in 1 2; do echo \"tick-$i\"; sleep 2; done").await; + assert_stdout_live(&kernel, id, "tick-1", "tick-2").await; + assert_eq!(wait_done(&kernel, id).await, "done:0"); + assert_eq!(stdout_of(&kernel, id).await, "tick-1\ntick-2\n"); +} + +#[tokio::test] +async fn whole_program_stderr_stream_matches_the_result_in_order() { + let kernel = kernel(); + let id = start_program(&kernel, "echo one >&2; sh -c 'echo two >&2'; echo three >&2").await; + let result = kernel.jobs().wait(id).await.expect("job result"); + let stream = stderr_of(&kernel, id).await; + assert_eq!(stream, "one\ntwo\nthree\n"); + assert_eq!(stream, result.err, "the stream and the result must agree"); +} + +#[tokio::test] +async fn whole_program_substitution_stderr_is_job_stderr() { + let kernel = kernel(); + let id = start_program(&kernel, "x=$(cat /kaish-no-such-file); echo \"got [$x]\"").await; + assert_eq!(wait_done(&kernel, id).await, "done:0"); + assert_eq!(stdout_of(&kernel, id).await, "got []\n"); + let err = stderr_of(&kernel, id).await; + assert_eq!(err.matches("kaish-no-such-file").count(), 1, "substitution stderr, once: {err:?}"); +} diff --git a/docs/EMBEDDING.md b/docs/EMBEDDING.md index f4252b86..4aac3532 100644 --- a/docs/EMBEDDING.md +++ b/docs/EMBEDDING.md @@ -1462,11 +1462,11 @@ let result = kernel.jobs().wait(id).await.expect("job remains tracked"); The job runs in a fork with the same options, tools, variables, and cwd as foreground execution. `JobManager::cancel` and the options' cancel token both -stop it with exit 130. Output reaches the job's stdout and stderr streams when -each top-level statement finishes, not while a statement runs. A runtime error -(exit 1), a timeout (exit 124), or a cancellation (exit 130) ends stderr with -one diagnostic line, in both the result and the stream. Shell `cmd &` jobs -stream external output as it arrives. +stop it with exit 130. Stdout reaches the job's stream as it is produced, the +same as a `cmd &` job; stderr reaches the stream when each top-level statement +finishes. A runtime error (exit 1), a timeout (exit 124), or a cancellation +(exit 130) ends stderr with one diagnostic line, in both the result and the +stream. ### JobFs for Background Job Observability From 0323c610e207a97448aeed67a40b9bd64021f0b4 Mon Sep 17 00:00:00 2001 From: A Tobey Date: Sun, 13 Sep 2026 09:15:14 -0400 Subject: [PATCH 2/2] fix: publish --help and AST output to a job's stdout stream Review of the previous commit found two stdout producers with no publish. `execute_command_depth` returns rendered `--help` before the tool runs, so `ls --help &` left `/v/jobs/N/stdout` empty (since #449) and a whole-program job would now lose it as well. `execute_streaming_inner` returns the AST dump without running a statement. Both publish through `ExecContext::publish_job_stdout`. `finalize_streams` documented that a whole-program job publishes each statement, which stopped being true; it now lists the producers. The `/v/jobs` help and EMBEDDING.md said stderr takes every stage's. An earlier stage's stderr and a builtin's reach the stream only at completion and only when nothing streamed live, so a last-stage external writing stderr hides them. The text now says so; live stderr for every stage is the later sinks work. The timeout guard test now says it pins "written once", not the backend publish itself. Co-Authored-By: DeepSeek V4 Flash Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 6 ++-- crates/kaish-help/content/en/vfs.md | 4 ++- crates/kaish-kernel/src/kernel.rs | 44 ++++++++++++++++++++++-- crates/kaish-kernel/src/scheduler/job.rs | 7 ++-- docs/EMBEDDING.md | 9 ++--- 5 files changed, 57 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b4a76cfd..884aa972 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,9 +19,9 @@ breaking entries are marked **BREAKING**. ### Fixed -- An embedder tool's stdout now reaches its background job's stdout stream. - `embedder_tool &` left `/v/jobs/N/stdout` empty; the output was only in - the job's result. +- An embedder tool's stdout and a tool's `--help` text now reach a background + job's stdout stream. `embedder_tool &` and `ls --help &` left + `/v/jobs/N/stdout` empty; the output was only in the job's result. - A streaming caller (`kaish -c`, `execute_with_options_streaming`) now receives stderr drained during an `exit` statement and the watchdog's timeout line. The timeout line follows any stderr the program wrote diff --git a/crates/kaish-help/content/en/vfs.md b/crates/kaish-help/content/en/vfs.md index 859038e4..ce06e51f 100644 --- a/crates/kaish-help/content/en/vfs.md +++ b/crates/kaish-help/content/en/vfs.md @@ -50,7 +50,9 @@ 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. +not. `stderr` fills live from an external command at the end of its pipeline. +Other stderr, from a builtin or an earlier stage, arrives when the job ends, +and only if nothing reached `stderr` live. 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 b30791d7..8c0a94c1 100644 --- a/crates/kaish-kernel/src/kernel.rs +++ b/crates/kaish-kernel/src/kernel.rs @@ -2520,8 +2520,12 @@ impl Kernel { { let scope = self.scope.read().await; if scope.show_ast() { + drop(scope); let output = format!("{:#?}\n", program); - return Ok(ExecResult::with_output(crate::interpreter::OutputData::text(output))); + let result = ExecResult::with_output(crate::interpreter::OutputData::text(output)); + // No statement runs, so nothing else publishes it to a background job. + self.exec_ctx.read().await.publish_job_stdout(&result).await; + return Ok(result); } } @@ -4043,7 +4047,10 @@ impl Kernel { let help_topic = crate::help::HelpTopic::Tool(name.to_string()); let ctx = self.exec_ctx.read().await; let content = crate::help::get_help(&help_topic, &ctx.tool_schemas); - return Ok(ExecResult::with_output(crate::interpreter::OutputData::text(content))); + let result = ExecResult::with_output(crate::interpreter::OutputData::text(content)); + // The tool never runs, so no builtin publish reaches a background job. + ctx.publish_job_stdout(&result).await; + return Ok(result); } // Snapshot exec_ctx into a local context and release the write lock @@ -10430,6 +10437,37 @@ AFTER="yes"'"#) ); } + #[tokio::test] + async fn background_program_publishes_tool_help() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::isolated().with_job_manager(jobs.clone())).expect("kernel"); + let id = kernel + .execute_background_with_options("ls --help", ExecuteOptions::new()) + .await + .expect("receipt"); + let result = jobs.wait(id).await.expect("job result"); + assert!(result.ok(), "{result:?}"); + assert!(!result.text_out().is_empty(), "the control must produce help text"); + let stream = String::from_utf8(jobs.read_stdout(id).await.expect("stdout stream")).expect("utf8"); + assert_eq!(stream, result.text_out(), "help text is the job's stdout"); + } + + #[tokio::test] + async fn background_job_publishes_tool_help() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::isolated().with_job_manager(jobs.clone())).expect("kernel"); + kernel.execute("ls --help &").await.expect("spawn"); + let id = crate::scheduler::JobId(1); + let result = jobs.wait(id).await.expect("job result"); + assert!(result.ok(), "{result:?}"); + assert!(!result.text_out().is_empty(), "the control must produce help text"); + let stream = String::from_utf8(jobs.read_stdout(id).await.expect("stdout stream")).expect("utf8"); + assert_eq!(stream, result.text_out(), "help text is the job's stdout"); + } + + /// Pins "written once": the backend arm and `timeout`'s own publish must + /// not both write. `background_job_publishes_custom_tool_stdout` is the + /// test that fails when the backend arm stops publishing. #[tokio::test] async fn background_job_publishes_redispatched_custom_tool_stdout_once() { use crate::backend::testing::MockBackend; @@ -10525,6 +10563,8 @@ AFTER="yes"'"#) let result = jobs.wait(id).await.expect("job result"); assert!(result.ok(), "{result:?}"); assert_eq!(result.text_out(), foreground.text_out()); + let stream = String::from_utf8(jobs.read_stdout(id).await.expect("stdout stream")).expect("utf8"); + assert_eq!(stream, result.text_out(), "the AST is the job's stdout"); } #[tokio::test] diff --git a/crates/kaish-kernel/src/scheduler/job.rs b/crates/kaish-kernel/src/scheduler/job.rs index 6bb6bab5..a0466d21 100644 --- a/crates/kaish-kernel/src/scheduler/job.rs +++ b/crates/kaish-kernel/src/scheduler/job.rs @@ -713,9 +713,10 @@ impl JobManager { /// Close a finished job's streams. /// - /// 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. + /// stdout is never written here. Every producer of a job's stdout + /// publishes as it runs: an external per chunk; a builtin, an embedder + /// tool, `--help`, or an AST dump when it returns; gather's rows. 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 diff --git a/docs/EMBEDDING.md b/docs/EMBEDDING.md index 4aac3532..31ddef1b 100644 --- a/docs/EMBEDDING.md +++ b/docs/EMBEDDING.md @@ -1509,10 +1509,11 @@ Three limits, stated because an embedder polling these needs to predict them: - **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`. + of it is published. `stderr` takes an external command's stderr live + when the command ends its pipeline. A builtin's stderr, and an earlier + stage's, reach the node at completion, and only when nothing arrived live; + otherwise they stay in the job's `ExecResult`. A whole-program job writes + each top-level statement's stderr when the statement finishes. - **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.