diff --git a/CHANGELOG.md b/CHANGELOG.md index 917dedb7..f191da87 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/kaish-kernel/src/kernel.rs b/crates/kaish-kernel/src/kernel.rs index 59fe5fe1..5fddb785 100644 --- a/crates/kaish-kernel/src/kernel.rs +++ b/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, on_output: Option<&mut (dyn FnMut(&ExecResult) + Send)>, ) -> Result { - 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, - on_output: Option<&mut (dyn FnMut(&ExecResult) + Send)>, - ) -> Result<(ExecResult, Option)> { 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, on_output: Option<&mut (dyn FnMut(&ExecResult) + Send)>, - ) -> Result<(ExecResult, Option)> { + ) -> Result { 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 { diff --git a/crates/kaish-kernel/tests/cancellation_tests.rs b/crates/kaish-kernel/tests/cancellation_tests.rs index 2724aad8..e3fe08bb 100644 --- a/crates/kaish-kernel/tests/cancellation_tests.rs +++ b/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, diff --git a/crates/kaish-kernel/tests/streaming_terminal_events_tests.rs b/crates/kaish-kernel/tests/streaming_terminal_events_tests.rs new file mode 100644 index 00000000..4f50a0b5 --- /dev/null +++ b/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"); +}