Streaming callers see every terminal event their result reports

CHANGELOG.md

@@ -19,6 +19,15 @@ breaking entries are marked **BREAKING**.
 
 ### Fixed
 
+- 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
+  instead of being dropped.
+- A cancel that fires before a program's first statement now stops it. An
+  embedder or job token cancelled that early let the program run to
+  completion.
+- A cancelled call exits 130 even when it ends a child by signal. A single
+  external command reported the child's 143 or 137 instead.
 - `grep` now reserves exit 1 for "no lines matched". An invalid pattern, an
   unreadable file, or a missing pattern argument exits 2, so a caller cannot
   read a broken search as a negative answer. `diff` argument errors exit 2 to

crates/kaish-kernel/src/kernel.rs

@@ -1796,9 +1796,10 @@ impl Kernel {
                         let diagnostic =
                             ExecResult::terminate_diagnostic(format!("timeout: timed out after {:?}", d));
                         res.code = 124;
-                        if res.err.is_empty() {
-                            res.err = diagnostic.clone();
-                        }
+                        // After any stderr the program already wrote, as the
+                        // `timeout` builtin does: `sleep: interrupted` alone
+                        // does not say a deadline fired.
+                        push_diagnostic(&mut res.err, &diagnostic);
                         timed_out = Some(diagnostic);
                     }
                     Ok((res, timed_out))
@@ -1941,28 +1942,20 @@ impl Kernel {
                     tracing::error!(job_id = %job_id, "background job output writer stopped before execution completed");
                 }
             };
-            // `run_watched` merges it into an Ok result only.
+            // `run_inner` merges it into an Ok result only.
             let embedder_baggage = opts.baggage.clone();
-            let outcome = fork.run_watched(&source, opts, None, Some(&mut on_output)).await;
+            let outcome = fork.run_inner(&source, opts, None, Some(&mut on_output)).await;
             if let Some(watcher) = embedder_watcher {
                 watcher.abort();
             }
 
-            // A timeout, runtime error, or cancellation is not a statement, so
-            // no callback carried its diagnostic to the stream.
+            // A runtime error or a cancellation is not a statement, so no
+            // callback carried its diagnostic to the stream. A timeout's
+            // diagnostic arrives through `on_output` like a statement's.
             let stream_needs_newline = !streamed.err.is_empty() && !streamed.err.ends_with('\n');
             let mut unstreamed_err = String::new();
             let mut result = match outcome {
-                Ok((mut result, timeout_diagnostic)) => {
-                    if let Some(diagnostic) = timeout_diagnostic {
-                        // The watchdog already wrote it into an empty `err`.
-                        if !result.err.trim_end().ends_with(diagnostic.trim_end()) {
-                            push_diagnostic(&mut result.err, &diagnostic);
-                        }
-                        unstreamed_err.push_str(&diagnostic);
-                    }
-                    result
-                }
+                Ok(result) => result,
                 Err(error) => {
                     let diagnostic =
                         ExecResult::terminate_diagnostic(format!("{:#}", classify_execute_error(error)));
@@ -2176,20 +2169,6 @@ impl Kernel {
         pipe_stdin: Option<crate::scheduler::PipeReader>,
         on_output: Option<&mut (dyn FnMut(&ExecResult) + Send)>,
     ) -> Result<ExecResult> {
-        self.run_watched(input, opts, pipe_stdin, on_output)
-            .await
-            .map(|(result, _timeout_diagnostic)| result)
-    }
-
-    /// [`Self::run_inner`], also returning the timeout diagnostic when the
-    /// watchdog deadline elapsed (see `run_under_watchdog`).
-    async fn run_watched(
-        &self,
-        input: &str,
-        opts: ExecuteOptions,
-        pipe_stdin: Option<crate::scheduler::PipeReader>,
-        on_output: Option<&mut (dyn FnMut(&ExecResult) + Send)>,
-    ) -> Result<(ExecResult, Option<String>)> {
         use opentelemetry::context::FutureExt;
 
         // Capture the embedder's baggage before `opts` is consumed so it can be
@@ -2204,9 +2183,9 @@ impl Kernel {
             None => self.execute_with_options_inner(input, opts, pipe_stdin, on_output).await,
         };
 
-        result.map(|(mut r, timeout_diagnostic)| {
+        result.map(|mut r| {
             crate::telemetry::merge_egress_baggage(&mut r, embedder_baggage);
-            (r, timeout_diagnostic)
+            r
         })
     }
 
@@ -2220,7 +2199,7 @@ impl Kernel {
         opts: ExecuteOptions,
         pipe_stdin: Option<crate::scheduler::PipeReader>,
         on_output: Option<&mut (dyn FnMut(&ExecResult) + Send)>,
-    ) -> Result<(ExecResult, Option<String>)> {
+    ) -> Result<ExecResult> {
         let _guard = self.acquire_execute_lock().await;
 
         // Always reset to a fresh internal token; this is the kernel's own
@@ -2280,11 +2259,14 @@ impl Kernel {
             if let Some(h) = watcher_handle {
                 h.abort();
             }
-            let diagnostic = "timeout: timed out after 0s";
-            return Ok((
-                ExecResult::failure(124, diagnostic.to_string()),
-                Some(ExecResult::terminate_diagnostic(diagnostic)),
-            ));
+            let result = ExecResult::failure(124, "timeout: timed out after 0s".to_string());
+            if let Some(on_output) = on_output {
+                // No statement ran to carry it.
+                let mut tail = ExecResult::success("");
+                tail.err = result.err.clone();
+                on_output(&tail);
+            }
+            return Ok(result);
         }
 
         // Apply per-call vars overlay (push frame + set_exported), wrapped in
@@ -2481,8 +2463,21 @@ impl Kernel {
         };
 
         let result = self
-            .run_under_watchdog(timeout, &effective_cancel, self.execute_streaming_inner(input, cb_ref))
-            .await;
+            .run_under_watchdog(timeout, &effective_cancel, self.execute_streaming_inner(input, &mut *cb_ref))
+            .await
+            .map(|(mut result, timeout_diagnostic)| {
+                // A deadline is not a statement, so no statement streamed it.
+                if let Some(diagnostic) = timeout_diagnostic {
+                    let mut tail = ExecResult::success("");
+                    tail.err = diagnostic;
+                    cb_ref(&tail);
+                } else if effective_cancel.is_cancelled() && !result.ok() {
+                    // The token, not the code: a killed child exits 128+signal,
+                    // and `exit 143` alone is not a cancel.
+                    result.code = 130;
+                }
+                result
+            });
 
         // Restore self.cancel_token to a fresh, uncancelled token so the
         // embedder's view of `Kernel::cancel()` stays predictable on the
@@ -2542,8 +2537,13 @@ impl Kernel {
 
         let mut result = ExecResult::success("");
 
-        // Reset cancellation token for this execution.
-        let cancel = self.reset_cancel();
+        // The caller installed this call's token. Resetting it here would
+        // discard a cancel that fired before the first statement.
+        let cancel = {
+            #[allow(clippy::expect_used)]
+            let token = self.cancel_token.lock().expect("cancel_token poisoned");
+            token.clone()
+        };
 
         for stmt in program.statements.into_iter() {
             if matches!(stmt, Stmt::Empty) {
@@ -2588,9 +2588,10 @@ impl Kernel {
                     accumulate_result(&mut result, &r);
                     result.set_output(last_output);
                 }
-                ControlFlow::Exit { code, result: carried } => {
+                ControlFlow::Exit { code, result: mut carried } => {
+                    // Into `carried`, as the other arms do, so `on_output` sees it.
                     if !drained_stderr.is_empty() {
-                        result.err.push_str(&drained_stderr);
+                        carried.err = format!("{}{}", drained_stderr, carried.err);
                     }
                     // Output produced before the exit — e.g. by the loop the
                     // `exit` ran inside — arrives on the signal. Emit it like
@@ -7974,6 +7975,31 @@ pub(crate) async fn kill_with_grace(
     child.wait().await
 }
 
+#[cfg(test)]
+#[allow(clippy::unwrap_used, clippy::expect_used)]
+mod statement_loop_cancel_tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn an_installed_cancelled_token_stops_the_statement_loop() {
+        // `execute_with_options_inner` installs the call's token before the
+        // loop runs. When an embedder or job token fired first, that token is
+        // already cancelled; the loop must stop on it, not replace it.
+        let kernel = Kernel::transient().expect("kernel");
+        let token = tokio_util::sync::CancellationToken::new();
+        token.cancel();
+        *kernel.cancel_token.lock().expect("cancel_token") = token;
+
+        let mut on_output = |_: &ExecResult| {};
+        let result = kernel
+            .execute_streaming_inner("echo a; echo b", &mut on_output)
+            .await
+            .expect("program runs");
+        assert_eq!(result.code, 130, "{result:?}");
+        assert_eq!(result.text_out(), "", "no statement may run under a cancelled token");
+    }
+}
+
 #[cfg(test)]
 #[allow(clippy::unwrap_used, clippy::expect_used)]
 mod argv_classify_tests {

crates/kaish-kernel/tests/cancellation_tests.rs

@@ -273,13 +273,9 @@ async fn kernel_cancel_kills_running_external() {
         .await
         .expect("execute");
 
-    // Cancel returns control with the kernel's "interrupted" path; exit code
-    // is 130 (SIGINT-style) on the cancellation checkpoint.
-    assert!(
-        result.code == 130 || result.code == 143,
-        "expected 130 or 143, got {}",
-        result.code,
-    );
+    // A cancelled call reports 130 whether a checkpoint or the killed child
+    // ended it; the child's own 128+SIGTERM (143) is not the call's status.
+    assert_eq!(result.code, 130, "cancellation must report 130: {result:?}");
 
     let pid = wait_for_pid(&pid_file, Duration::from_secs(2)).await.expect("pid_file");
     assert!(
@@ -404,15 +400,10 @@ async fn grace_escalation_sigkills_term_trapping_child() {
     let (pid, kill_requested) = ready;
     let elapsed = kill_requested.elapsed();
 
-    // 137 is 128 + SIGKILL(9): the child's own wait status, reported straight
-    // through. This is the sharpest evidence the test has — 143 (128 + SIGTERM)
-    // would mean plain SIGTERM did the job and no escalation ever happened.
-    assert_eq!(
-        result.code, 137,
-        "expected 137 (128 + SIGKILL) — the TERM-ignoring child should have been \
-         escalated to SIGKILL; 143 would mean SIGTERM killed it and the escalation \
-         never ran",
-    );
+    // A cancelled call reports 130 whatever signal ended the child, as a
+    // timeout reports 124. The escalation is proven below: the child ignores
+    // SIGTERM, so dying at all means SIGKILL, and not before the grace.
+    assert_eq!(result.code, 130, "a cancelled call must report 130: {result:?}");
 
     assert!(
         wait_for_dead(pid, Duration::from_secs(3)).await,

crates/kaish-kernel/tests/streaming_terminal_events_tests.rs

@@ -0,0 +1,60 @@
+//! A streaming caller sees everything the returned result reports.
+//!
+//! `execute_with_options_streaming` calls `on_output` once per top-level
+//! statement, and the REPL `-c` frontend prints only what `on_output`
+//! delivers. Output that reaches the returned `ExecResult` but never the
+//! callback is output that caller never shows. Each test compares the two.
+//!
+//! Only builtins run here, so these need no feature gate.
+
+#![allow(clippy::unwrap_used, clippy::expect_used)]
+
+use std::time::Duration;
+
+use kaish_kernel::interpreter::ExecResult;
+use kaish_kernel::{ExecuteOptions, Kernel};
+
+/// Run `program`, returning the stderr `on_output` delivered and the result.
+async fn run_streaming(program: &str, opts: ExecuteOptions) -> (String, ExecResult) {
+    let kernel = Kernel::transient().expect("kernel");
+    let mut streamed_err = String::new();
+    let mut on_output = |output: &ExecResult| streamed_err.push_str(&output.err);
+    let result = kernel
+        .execute_with_options_streaming(program, opts, &mut on_output)
+        .await
+        .expect("program runs");
+    (streamed_err, result)
+}
+
+#[tokio::test]
+async fn substitution_stderr_streams_from_an_ordinary_statement() {
+    // Control: an ordinary statement already streamed its drained stderr.
+    let (streamed, result) = run_streaming("echo $(ls /no-such-dir; echo 3)", ExecuteOptions::new()).await;
+    assert!(result.err.contains("no-such-dir"), "{result:?}");
+    assert_eq!(streamed, result.err);
+}
+
+#[tokio::test]
+async fn substitution_stderr_streams_from_an_exit_statement() {
+    let (streamed, result) = run_streaming("exit $(ls /no-such-dir; echo 3)", ExecuteOptions::new()).await;
+    assert_eq!(result.code, 3);
+    assert!(result.err.contains("no-such-dir"), "{result:?}");
+    assert_eq!(streamed, result.err, "the exit statement's stderr must reach the stream");
+}
+
+#[tokio::test]
+async fn timeout_is_named_after_earlier_stderr() {
+    let (streamed, result) = run_streaming(
+        "echo early >&2; sleep 30",
+        ExecuteOptions::new().with_timeout(Duration::from_millis(50)),
+    )
+    .await;
+    assert_eq!(result.code, 124);
+    let early = result.err.find("early").expect("statement stderr kept");
+    let timed_out = result
+        .err
+        .find("timeout: timed out after")
+        .unwrap_or_else(|| panic!("timeout diagnostic missing: {:?}", result.err));
+    assert!(early < timed_out, "{:?}", result.err);
+    assert_eq!(streamed, result.err, "the timeout diagnostic must reach the stream");
+}