diff --git a/CHANGELOG.md b/CHANGELOG.md index 267bc033..917dedb7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,13 @@ breaking entries are marked **BREAKING**. ## [Unreleased] +### Added + +- **`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. + ### Fixed - `grep` now reserves exit 1 for "no lines matched". An invalid pattern, an diff --git a/crates/kaish-kernel/src/kernel.rs b/crates/kaish-kernel/src/kernel.rs index bf9fdc2d..59fe5fe1 100644 --- a/crates/kaish-kernel/src/kernel.rs +++ b/crates/kaish-kernel/src/kernel.rs @@ -1449,7 +1449,9 @@ impl Kernel { cancel: tokio_util::sync::CancellationToken, job_id: crate::scheduler::JobId, ) -> Arc { - self.fork_inner(cancel, Some(job_id)).await + let fork = self.fork_inner(cancel, Some(job_id)).await; + fork.exec_ctx.write().await.background_stream_external_output = true; + fork } /// Shared fork implementation. Caller decides the cancellation token and @@ -1740,7 +1742,9 @@ impl Kernel { // call to mis-take. Ok(result) }; - let result = self.run_under_watchdog(timeout, &cancel, work).await?; + // The argv door returns one command's result; it has no statement + // stream that could miss the timeout diagnostic. + let (result, _timeout_diagnostic) = self.run_under_watchdog(timeout, &cancel, work).await?; self.update_last_result(&result).await; Ok(result) } @@ -1757,12 +1761,17 @@ impl Kernel { /// watchdog handle from `exec_ctx` on the way out (a patient hold against a /// stale handle would silently suspend nothing). Callers must short-circuit a /// `Some(Duration::ZERO)` timeout (return 124 without spawning) before calling. + /// + /// The second value is the timeout diagnostic when the deadline elapsed, + /// so a caller that streams statements can report a timeout that no + /// statement produced; an exit code of 124 alone does not mean the + /// deadline fired. async fn run_under_watchdog( &self, timeout: Option, cancel: &tokio_util::sync::CancellationToken, work: F, - ) -> Result + ) -> Result<(ExecResult, Option)> where F: std::future::Future>, { @@ -1782,19 +1791,22 @@ impl Kernel { timer.abort(); match r { Ok(mut res) => { + let mut timed_out = None; if elapsed.load(std::sync::atomic::Ordering::SeqCst) { + let diagnostic = + ExecResult::terminate_diagnostic(format!("timeout: timed out after {:?}", d)); res.code = 124; if res.err.is_empty() { - res.err = - ExecResult::terminate_diagnostic(format!("timeout: timed out after {:?}", d)); + res.err = diagnostic.clone(); } + timed_out = Some(diagnostic); } - Ok(res) + Ok((res, timed_out)) } Err(e) => Err(e), } } else { - work.await + work.await.map(|res| (res, None)) }; // The timer task is gone (fired or aborted); drop the stale handle. @@ -1837,6 +1849,226 @@ impl Kernel { self.run_inner(input, opts, None, None).await.map_err(classify_execute_error) } + /// Start a complete kaish program as a background job. + /// + /// The source is parsed and validated on the job's fork before this + /// returns a [`crate::scheduler::JobId`]; a rejected program registers no + /// 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. + /// + /// # Errors + /// + /// Returns [`KernelError::Parse`] or [`KernelError::Validation`] before + /// registering a job. + pub async fn execute_background_with_options( + &self, + source: &str, + mut opts: ExecuteOptions, + ) -> Result { + use tokio::sync::{mpsc, oneshot}; + + // Validate the fork that will run: it holds its own copy of user + // tools and scope, which a later snapshot of the parent could differ + // from. + // The job parses and validates again when it runs; the cost buys a + // rejection before a job exists, without threading a parsed program + // through the execute doors. AST mode prints before validating, so it + // is not validated here either. + let cancel = tokio_util::sync::CancellationToken::new(); + let fork = self.fork_inner(cancel.clone(), None).await; + let program = Self::parse_program(source).map_err(classify_execute_error)?; + let show_ast = fork.scope.read().await.show_ast(); + if !show_ast { + fork.validate_parsed(&program, source).await.map_err(classify_execute_error)?; + } + + let (result_tx, result_rx) = oneshot::channel(); + 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. + let mut ec = fork.exec_ctx.write().await; + ec.background_job = Some(job_id); + ec.background_stream_external_output = false; + } + + // The job token is this call's cancel input, so `JobManager::cancel` + // reaches builtins and child process groups. An embedder token + // forwards into it. + let embedder_cancel = opts.cancel_token.replace(cancel.clone()); + + let jobs = self.jobs.clone(); + let (output_tx, mut output_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 { + 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; + } + })); + + let source = source.to_owned(); + tokio::spawn(crate::telemetry::bind_current_context(async move { + let embedder_watcher = embedder_cancel.map(|embedder_cancel| { + let cancel = cancel.clone(); + tokio::spawn(async move { + embedder_cancel.cancelled().await; + cancel.cancel(); + }) + }); + // The statements as streamed: a runtime error returns no result, + // and the output that ran before it must survive. + let mut streamed = ExecResult::success(""); + let mut on_output = |output: &ExecResult| { + 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"); + } + }; + // `run_watched` 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; + 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. + 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 + } + Err(error) => { + let diagnostic = + ExecResult::terminate_diagnostic(format!("{:#}", classify_execute_error(error))); + let mut result = streamed; + result.code = 1; + push_diagnostic(&mut result.err, &diagnostic); + unstreamed_err.push_str(&diagnostic); + crate::telemetry::merge_egress_baggage(&mut result, embedder_baggage); + result + } + }; + // The token, not the code: a cancelled external exits 128+signal, + // and `exit 130` alone is not a cancel. A job that succeeded before + // the token tripped keeps its result. + if cancel.is_cancelled() && !result.ok() { + result.code = 130; + let diagnostic = ExecResult::terminate_diagnostic("job cancelled"); + push_diagnostic(&mut result.err, &diagnostic); + unstreamed_err.push_str(&diagnostic); + } + if !unstreamed_err.is_empty() { + let mut tail = ExecResult::success(""); + if stream_needs_newline { + tail.err.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"); + } + } + drop(output_tx); + + if let Err(error) = output_writer.await { + result.code = 1; + push_diagnostic( + &mut result.err, + &ExecResult::terminate_diagnostic(format!("background job output writer failed: {error}")), + ); + } + jobs.finalize_streams(job_id, &result).await; + // The receiver is gone only when the job was removed from the manager. + let _ = result_tx.send(result); + })); + + Ok(job_id) + } + + /// Parse kaish source, tagging a failure as [`KernelError::Parse`] so + /// `classify_execute_error` can recover the structured rejection at the + /// public execute boundary. + fn parse_program(input: &str) -> Result { + parse(input).map_err(|errors| { + let msg = errors + .iter() + .map(|e| e.format(input)) + .collect::>() + .join("\n"); + let message = format!("parse error:\n{}", msg); + anyhow::Error::from(KernelError::Parse { errors, message }) + }) + } + + /// Validate a parsed program before it runs. An error-severity issue + /// returns [`KernelError::Validation`]. The returned string holds the + /// warnings whose code opts into agent surfacing, one per line; other + /// warnings stay trace-only (every external command fires an + /// `UndefinedCommand` warning). + async fn validate_parsed(&self, program: &crate::ast::Program, input: &str) -> Result { + let mut surfaced_warnings = String::new(); + if self.skip_validation { + return Ok(surfaced_warnings); + } + + // Catalog first: neither guard should ride the other's await, and + // `validate()` is synchronous, so neither rides one after this. + let catalog = { self.exec_ctx.read().await.tool_schemas.clone() }; + let user_tools = self.user_tools.read().await; + let validator = Validator::new(&self.tools, &user_tools, &catalog); + let issues = validator.validate(program); + + let errors: Vec<_> = issues + .iter() + .filter(|i| i.severity == Severity::Error) + .collect(); + if !errors.is_empty() { + let error_msg = errors + .iter() + .map(|e| e.format(input)) + .collect::>() + .join("\n"); + let message = format!("validation failed:\n{}", error_msg); + let issues: Vec = + errors.into_iter().cloned().collect(); + return Err(anyhow::Error::from(KernelError::Validation { issues, message })); + } + + for warning in issues.iter().filter(|i| i.severity == Severity::Warning) { + tracing::trace!("validation: {}", warning.format(input)); + if warning.code.surfaces_to_agent() { + surfaced_warnings.push_str(&warning.format(input)); + surfaced_warnings.push('\n'); + } + } + Ok(surfaced_warnings) + } + /// Same as [`Self::execute_with_options`] but with a per-statement output /// callback. The callback fires after each top-level statement so the /// embedder (REPL, MCP streaming) can flush output incrementally. @@ -1944,6 +2176,20 @@ 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 @@ -1958,9 +2204,9 @@ impl Kernel { None => self.execute_with_options_inner(input, opts, pipe_stdin, on_output).await, }; - result.map(|mut r| { + result.map(|(mut r, timeout_diagnostic)| { crate::telemetry::merge_egress_baggage(&mut r, embedder_baggage); - r + (r, timeout_diagnostic) }) } @@ -1974,7 +2220,7 @@ impl Kernel { opts: ExecuteOptions, pipe_stdin: Option, on_output: Option<&mut (dyn FnMut(&ExecResult) + Send)>, - ) -> Result { + ) -> Result<(ExecResult, Option)> { let _guard = self.acquire_execute_lock().await; // Always reset to a fresh internal token; this is the kernel's own @@ -2034,7 +2280,11 @@ impl Kernel { if let Some(h) = watcher_handle { h.abort(); } - return Ok(ExecResult::failure(124, "timeout: timed out after 0s".to_string())); + let diagnostic = "timeout: timed out after 0s"; + return Ok(( + ExecResult::failure(124, diagnostic.to_string()), + Some(ExecResult::terminate_diagnostic(diagnostic)), + )); } // Apply per-call vars overlay (push frame + set_exported), wrapped in @@ -2265,18 +2515,7 @@ impl Kernel { input: &str, on_output: &mut (dyn FnMut(&ExecResult) + Send), ) -> Result { - let program = parse(input).map_err(|errors| { - let msg = errors - .iter() - .map(|e| e.format(input)) - .collect::>() - .join("\n"); - let message = format!("parse error:\n{}", msg); - // Tagged so `classify_execute_error` can recover the structured - // rejection at the public execute-surface boundary; every other - // `?` in this function propagates a plain, untagged `anyhow::Error`. - anyhow::Error::from(KernelError::Parse { errors, message }) - })?; + let program = Self::parse_program(input)?; // AST display mode: show AST instead of executing { @@ -2287,48 +2526,7 @@ impl Kernel { } } - // Pre-execution validation. Most warnings stay trace-only (every - // external command fires an `UndefinedCommand` warning), but a warning - // whose code opts into agent surfacing is collected here and prepended - // to the result's stderr at each return point below. - let mut surfaced_warnings = String::new(); - if !self.skip_validation { - // Catalog first: neither guard should ride the other's await, and - // `validate()` is synchronous, so neither rides one after this. - let catalog = { self.exec_ctx.read().await.tool_schemas.clone() }; - let user_tools = self.user_tools.read().await; - let validator = Validator::new(&self.tools, &user_tools, &catalog); - let issues = validator.validate(&program); - - // Collect errors (warnings are logged but don't prevent execution) - let errors: Vec<_> = issues - .iter() - .filter(|i| i.severity == Severity::Error) - .collect(); - - if !errors.is_empty() { - let error_msg = errors - .iter() - .map(|e| e.format(input)) - .collect::>() - .join("\n"); - let message = format!("validation failed:\n{}", error_msg); - let issues: Vec = - errors.into_iter().cloned().collect(); - // Tagged the same way as the parse rejection above. - return Err(anyhow::Error::from(KernelError::Validation { issues, message })); - } - - // Log warnings via tracing (trace level to avoid noise); surface the - // opted-in ones to the agent so the guidance is actually seen. - for warning in issues.iter().filter(|i| i.severity == Severity::Warning) { - tracing::trace!("validation: {}", warning.format(input)); - if warning.code.surfaces_to_agent() { - surfaced_warnings.push_str(&warning.format(input)); - surfaced_warnings.push('\n'); - } - } - } + let surfaced_warnings = self.validate_parsed(&program, input).await?; // Surface opted-in validation warnings to the streaming frontend once, // before any command output. The streaming consumer (`-c`, REPL) prints @@ -3213,6 +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, aliases: ec.aliases.clone(), ignore_config: ec.ignore_config.clone(), output_limit: ec.output_limit.clone(), @@ -7391,6 +7590,14 @@ fn push_stdout_of(accumulated: &mut ExecResult, new: &ExecResult) { } } +/// Append a terminated diagnostic to `err` on its own line. +fn push_diagnostic(err: &mut String, diagnostic: &str) { + if !err.is_empty() && !err.ends_with('\n') { + err.push('\n'); + } + err.push_str(diagnostic); +} + fn accumulate_result(accumulated: &mut ExecResult, new: &ExecResult) { push_stdout_of(accumulated, new); accumulated.err.push_str(&new.err); @@ -9964,6 +10171,231 @@ AFTER="yes"'"#) assert!(stdout.text_out().contains("hello")); } + #[tokio::test] + async fn background_program_preflights_before_receipt() { + let kernel = Kernel::transient().expect("kernel"); + + let error = kernel + .execute_background_with_options("if then", ExecuteOptions::new()) + .await + .expect_err("invalid source must not receive a job id"); + assert!(matches!(error, KernelError::Parse { .. }), "unexpected error: {error}"); + assert!(kernel.jobs.list().await.is_empty(), "rejected source must not register a job"); + } + + #[tokio::test] + async fn background_program_preserves_options_custom_tools_and_statement_output() { + 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}")))); + let kernel = Kernel::with_backend( + Arc::new(backend), + KernelConfig::isolated().with_job_manager(jobs.clone()), + |_| {}, + |_| {}, + ) + .expect("kernel"); + + let id = kernel + .execute_background_with_options( + "echo before; embedder_tool; echo $WHO; pwd", + ExecuteOptions::new() + .with_var("WHO", Value::String("Amy".to_owned())) + .with_cwd(PathBuf::from("/workspace")), + ) + .await + .expect("receipt"); + let result = jobs.wait(id).await.expect("job result"); + assert!(result.ok(), "background program failed: {result:?}"); + assert_eq!(result.text_out(), "before\ntool:embedder_toolAmy\n/workspace"); + 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"), + result.text_out(), + "statement output must be complete and must not duplicate custom-tool output" + ); + } + + #[tokio::test] + async fn background_program_cancel_reaches_running_execution() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::repl().with_job_manager(jobs.clone())).expect("kernel"); + let id = kernel + .execute_background_with_options("sleep 30; echo unreachable", ExecuteOptions::new()) + .await + .expect("receipt"); + + assert!(jobs.cancel(id).await, "registered job needs a cancellation token"); + let result = tokio::time::timeout(Duration::from_secs(5), jobs.wait(id)) + .await + .expect("cancelled job did not settle") + .expect("job result"); + assert_eq!(result.code, 130, "cancellation must report the normal cancellation code: {result:?}"); + assert!(!result.text_out().contains("unreachable")); + let stderr = String::from_utf8(jobs.read_stderr(id).await.expect("stderr stream")).expect("utf8"); + assert!(stderr.ends_with("job cancelled\n"), "job stream must name the cancellation: {stderr:?}"); + assert!(result.err.ends_with("job cancelled\n"), "result must name the cancellation: {result:?}"); + } + + #[tokio::test] + async fn background_program_cancel_of_external_reports_cancellation() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::repl().with_job_manager(jobs.clone())).expect("kernel"); + let id = kernel + .execute_background_with_options("/usr/bin/sleep 30", ExecuteOptions::new()) + .await + .expect("receipt"); + + // Cancel only once the child exists, so the signal reaches it rather + // than a program that has not started. + let started = tokio::time::timeout(Duration::from_secs(5), async { + while jobs.job_pgids(id).await.is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await; + assert!(started.is_ok(), "external child never recorded its process group"); + + assert!(jobs.cancel(id).await); + let result = tokio::time::timeout(Duration::from_secs(5), jobs.wait(id)) + .await + .expect("cancelled job did not settle") + .expect("job result"); + let stderr = String::from_utf8(jobs.read_stderr(id).await.expect("stderr stream")).expect("utf8"); + assert_eq!(result.code, 130, "a cancelled external must report cancellation, not its signal: {result:?}"); + assert!(stderr.ends_with("job cancelled\n"), "job stream must name the cancellation: {stderr:?}"); + } + + #[tokio::test] + async fn background_program_in_ast_mode_matches_foreground() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::isolated().with_job_manager(jobs.clone())).expect("kernel"); + let on = kernel.execute("kaish-ast --on").await.expect("ast mode on"); + assert_eq!(on.text_out(), "AST mode: ON\n", "the control must actually enable AST mode"); + + // `break` outside a loop fails validation, but AST mode prints before + // validating; the background door must not reject what the foreground shows. + let foreground = kernel.execute("break").await.expect("foreground prints the AST"); + let id = kernel + .execute_background_with_options("break", ExecuteOptions::new()) + .await + .expect("AST mode must not reject the program before it runs"); + let result = jobs.wait(id).await.expect("job result"); + assert!(result.ok(), "{result:?}"); + assert_eq!(result.text_out(), foreground.text_out()); + } + + #[tokio::test] + async fn background_program_timeout_keeps_statement_stderr_and_terminal_diagnostic() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::repl().with_job_manager(jobs.clone())).expect("kernel"); + let id = kernel + .execute_background_with_options( + "echo early >&2; sleep 30", + ExecuteOptions::new().with_timeout(Duration::from_millis(50)), + ) + .await + .expect("receipt"); + let result = tokio::time::timeout(Duration::from_secs(5), jobs.wait(id)) + .await.expect("timed out job did not settle").expect("job result"); + let stderr = String::from_utf8(jobs.read_stderr(id).await.expect("stderr stream")).expect("utf8"); + assert_eq!(result.code, 124, "timeout must report 124: {result:?}"); + for (name, text) in [("result", result.err.as_str()), ("stream", stderr.as_str())] { + let early = text.find("early").unwrap_or_else(|| panic!("{name} lost statement stderr: {text:?}")); + let timed_out = text + .find("timeout: timed out after") + .unwrap_or_else(|| panic!("{name} lost the timeout diagnostic: {text:?}")); + assert!(early < timed_out, "{name} must keep statement stderr before the timeout: {text:?}"); + } + } + + #[tokio::test] + async fn background_program_runtime_error_keeps_prior_output_and_cause() { + 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( + "echo out; echo early >&2; x=$((1/0)); echo unreachable", + ExecuteOptions::new(), + ) + .await + .expect("receipt"); + let result = jobs.wait(id).await.expect("job result"); + let stdout = String::from_utf8(jobs.read_stdout(id).await.expect("stdout stream")).expect("utf8"); + let stderr = String::from_utf8(jobs.read_stderr(id).await.expect("stderr stream")).expect("utf8"); + + assert_eq!(result.code, 1, "a runtime fault fails the job: {result:?}"); + assert_eq!(result.text_out(), "out\n", "output before the fault must survive in the result"); + assert_eq!(stdout, "out\n"); + for (name, text) in [("result", result.err.as_str()), ("stream", stderr.as_str())] { + let early = text.find("early").unwrap_or_else(|| panic!("{name} lost statement stderr: {text:?}")); + let cause = text + .find("divides by zero") + .unwrap_or_else(|| panic!("{name} lost the fault's cause: {text:?}")); + assert!(early < cause, "{name} must keep statement stderr before the fault: {text:?}"); + } + } + + #[tokio::test] + async fn background_program_exit_code_alone_is_not_a_terminal_event() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::isolated().with_job_manager(jobs.clone())).expect("kernel"); + for code in [124, 130] { + let id = kernel + .execute_background_with_options(&format!("exit {code}"), ExecuteOptions::new()) + .await + .expect("receipt"); + let result = jobs.wait(id).await.expect("job result"); + let stderr = String::from_utf8(jobs.read_stderr(id).await.expect("stderr stream")).expect("utf8"); + assert_eq!(result.code, code); + assert_eq!(stderr, "", "`exit {code}` was neither timed out nor cancelled"); + assert_eq!(result.err, "", "`exit {code}` was neither timed out nor cancelled"); + } + } + + #[tokio::test] + async fn background_program_mixed_builtin_and_external_streams_once() { + let jobs = Arc::new(JobManager::new()); + let kernel = Kernel::new(KernelConfig::repl().with_job_manager(jobs.clone())).expect("kernel"); + let id = kernel + .execute_background_with_options( + "echo builtin-before; /usr/bin/printf 'external\\n'; echo builtin-after", + ExecuteOptions::new(), + ) + .await + .expect("receipt"); + + let result = jobs.wait(id).await.expect("job result"); + assert!(result.ok(), "background program failed: {result:?}"); + let stream = String::from_utf8(jobs.read_stdout(id).await.expect("stdout stream")).expect("utf8"); + assert_eq!(stream, "builtin-before\nexternal\nbuiltin-after\n"); + assert_eq!(stream, result.text_out(), "external tee must neither erase builtin output nor duplicate itself"); + } + + #[tokio::test] + async fn background_programs_share_an_embedder_job_manager_across_kernels() { + let jobs = Arc::new(JobManager::new()); + let first = Kernel::new(KernelConfig::isolated().with_job_manager(jobs.clone())) + .expect("first kernel"); + let second = Kernel::new(KernelConfig::isolated().with_job_manager(jobs.clone())) + .expect("second kernel"); + + let first_id = first + .execute_background_with_options("echo first", ExecuteOptions::new()) + .await + .expect("first receipt"); + let second_id = second + .execute_background_with_options("echo second", ExecuteOptions::new()) + .await + .expect("second receipt"); + assert_ne!(first_id, second_id, "the shared manager assigns one job namespace"); + assert_eq!(jobs.wait(first_id).await.expect("first result").text_out(), "first\n"); + assert_eq!(jobs.wait(second_id).await.expect("second result").text_out(), "second\n"); + } + #[tokio::test] async fn test_heredoc_piped_to_command() { // Bug 4: heredoc content should pipe through to next command diff --git a/crates/kaish-kernel/src/spawn.rs b/crates/kaish-kernel/src/spawn.rs index 1c441fc1..7b73dbc5 100644 --- a/crates/kaish-kernel/src/spawn.rs +++ b/crates/kaish-kernel/src/spawn.rs @@ -121,6 +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, } impl SpawnContext { @@ -133,6 +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, } } } @@ -322,8 +325,12 @@ pub(crate) async fn spawn_process(request: SpawnRequest, spawn_ctx: &SpawnContex // them itself. This is what makes `/v/jobs/{id}/stdout` grow while a // `cargo build &` is still building (GH #240 removed the node rather // than wire this tee; the tee is the half that was missing). - let job_streams = match (&spawn_ctx.job_manager, spawn_ctx.background_job) { - (Some(jobs), Some(job_id)) => jobs.streams(job_id).await, + let job_streams = match ( + &spawn_ctx.job_manager, + spawn_ctx.background_job, + spawn_ctx.background_stream_external_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 c007d31d..9e6430eb 100644 --- a/crates/kaish-kernel/src/tools/context.rs +++ b/crates/kaish-kernel/src/tools/context.rs @@ -215,6 +215,10 @@ 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, /// Command aliases (name → expansion string). pub aliases: HashMap, /// Ignore file configuration for file-walking tools. @@ -441,6 +445,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, + background_stream_external_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -482,6 +487,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, + background_stream_external_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -520,6 +526,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, + background_stream_external_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -558,6 +565,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, + background_stream_external_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -599,6 +607,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, + background_stream_external_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -637,6 +646,7 @@ impl ExecContext { kill_children_on_parent_death: false, kill_grace: DEFAULT_KILL_GRACE, background_job: None, + background_stream_external_output: false, aliases: HashMap::new(), ignore_config: IgnoreConfig::none(), output_limit: OutputLimitConfig::none(), @@ -933,6 +943,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, aliases: self.aliases.clone(), ignore_config: self.ignore_config.clone(), output_limit: self.output_limit.clone(), diff --git a/docs/EMBEDDING.md b/docs/EMBEDDING.md index 1fc94325..4b03d06c 100644 --- a/docs/EMBEDDING.md +++ b/docs/EMBEDDING.md @@ -1441,6 +1441,30 @@ last kernel built wins for both. A hermetic kernel (`NoLocal`, or any that manager. Share a manager between kernels configured alike, or accept the last writer. +### Whole-program asynchronous execution + +`Kernel::execute_background_with_options(source, options)` starts one complete +kaish program as a job and returns its `JobId`. The source is parsed and +validated on the job's fork first; a rejected program returns +`KernelError::Parse` or `KernelError::Validation` and registers no job. It uses +the kernel's `JobManager`, so an embedder that builds a kernel per request +supplies a shared manager with `KernelConfig::with_job_manager` first. + +```rust +let id = kernel + .execute_background_with_options("echo first; long_task; echo last", ExecuteOptions::new()) + .await?; +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. + ### JobFs for Background Job Observability The kernel automatically mounts `JobFs` at `/v/jobs`, exposing background