Whole-program jobs stream stdout as it is produced

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 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

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

crates/kaish-kernel/src/kernel.rs

@@ -1450,7 +1450,11 @@ 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_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::<ExecResult>();
+        let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::<String>();
 
-        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;
@@ -2516,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);
             }
         }
 
@@ -3470,6 +3478,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 +3969,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(_)) => {
@@ -4034,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
@@ -10393,6 +10409,93 @@ 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_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;
+        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());
@@ -10460,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]

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<BoundedStream>,
-    /// 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<BoundedStream>,
 }
 
@@ -710,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

crates/kaish-kernel/src/spawn.rs

@@ -121,8 +121,10 @@ 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 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,19 +546,22 @@ 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,
                 PipelinePosition::Only | PipelinePosition::Last
             )
             .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 {