diff --git a/Cargo.lock b/Cargo.lock index 91a79a20..0b54e683 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -384,6 +384,12 @@ dependencies = [ "crypto-common 0.2.2", ] +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "enumflags2" version = "0.7.12" @@ -417,7 +423,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -487,6 +493,17 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-macro" version = "0.3.34" @@ -504,6 +521,12 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" +[[package]] +name = "futures-timer" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af43fadb8a98512d547e37b4e92e0ced13e205c061b87b4623eff01d918d6968" + [[package]] name = "futures-util" version = "0.3.34" @@ -913,6 +936,26 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "regex-automata" version = "0.4.18" @@ -979,7 +1022,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1042,6 +1085,31 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "schemars" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab508826f74a77ca9d5aba6ff19b522583ee3eaf28a19384ff3d0e5835fadf6e" +dependencies = [ + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1c3a92094fa7d61aa124645844facb6b554dfc797136d0f5fd1f890e2bffc69" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 3.0.3", +] + [[package]] name = "serde" version = "1.0.229" @@ -1072,6 +1140,17 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "serde_json" version = "1.0.151" @@ -1224,10 +1303,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1254,6 +1333,10 @@ dependencies = [ name = "tinybox-bus" version = "0.2.3" dependencies = [ + "futures-executor", + "futures-timer", + "futures-util", + "schemars", "serde", "serde_json", "tinystoragedrivers-types", @@ -1943,7 +2026,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/tinybox-bus/Cargo.toml b/crates/tinybox-bus/Cargo.toml index f8a879f2..6bb7b1b9 100644 --- a/crates/tinybox-bus/Cargo.toml +++ b/crates/tinybox-bus/Cargo.toml @@ -10,6 +10,12 @@ repository.workspace = true [dependencies] # The bus boundary contains data only; serialization is its sole dependency. serde.workspace = true +# OpenHuman config operations share this contract type for schema generation. +schemars = "1" +# Runtime-neutral future combinators for the optional native-shell client helper. +futures-util = { version = "0.3", features = ["async-await"] } +# Runtime-neutral delay used between bounded output polls. +futures-timer = "3" # Only the `storage` feature: pure collection and legacy-body vocabulary. serde_json = { workspace = true, optional = true } tinystoragedrivers-types = { workspace = true, optional = true } @@ -22,6 +28,7 @@ storage = ["dep:serde_json", "dep:tinystoragedrivers-types"] [dev-dependencies] serde_json.workspace = true +futures-executor = "0.3" [lints] workspace = true diff --git a/crates/tinybox-bus/src/grants.rs b/crates/tinybox-bus/src/grants.rs index 13258b10..b1d976c0 100644 --- a/crates/tinybox-bus/src/grants.rs +++ b/crates/tinybox-bus/src/grants.rs @@ -1,8 +1,9 @@ //! Host-configured filesystem grant discovery and protected-path vocabulary. +use schemars::JsonSchema; use serde::{Deserialize, Serialize}; /// Configuration used to discover toolchain and explicit filesystem grants. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(default)] pub struct LocalGrantConfig { /// Include existing toolchain homes and git configuration. diff --git a/crates/tinybox-bus/src/lib.rs b/crates/tinybox-bus/src/lib.rs index 274ff7ee..444e3166 100644 --- a/crates/tinybox-bus/src/lib.rs +++ b/crates/tinybox-bus/src/lib.rs @@ -1,4 +1,5 @@ -//! Serialized requests and results for the `TinyBox` module. +//! Serialized requests and results for the `TinyBox` module, plus runtime-neutral +//! native-shell lifecycle orchestration over a caller-supplied client. //! //! This crate has no bus connection, executor, native backend, or policy. //! A caller applies its own authorization before sending these requests. @@ -18,6 +19,12 @@ use std::collections::BTreeMap; mod grants; pub use grants::{LocalGrantConfig, LocalGrants, ProtectedPaths, ResolveLocalGrantsRequest}; +mod native_shell; +pub use native_shell::{ + NativeShellClient, NativeShellError, NativeShellOutput, NativeShellRunOutcome, + NativeShellRunRequest, NativeShellRunning, run_native_shell, +}; + mod jail; pub use jail::{JailEnforcement, NativeJailConfig, NativeJailRequest, NativeJailResourceRequest}; diff --git a/crates/tinybox-bus/src/lib_tests.rs b/crates/tinybox-bus/src/lib_tests.rs index 3e42c923..3a18e9ee 100644 --- a/crates/tinybox-bus/src/lib_tests.rs +++ b/crates/tinybox-bus/src/lib_tests.rs @@ -16,6 +16,14 @@ fn command_optional_fields_default_when_absent() -> Result<(), serde_json::Error Ok(()) } +#[test] +fn local_grant_config_exposes_a_json_schema() -> Result<(), serde_json::Error> { + let schema = schemars::schema_for!(LocalGrantConfig); + let schema = serde_json::to_value(schema)?; + assert!(schema.get("properties").is_some()); + Ok(()) +} + #[test] fn create_environment_defaults_and_workspace_round_trips() -> Result<(), serde_json::Error> { let request: CreateRequest = serde_json::from_str( diff --git a/crates/tinybox-bus/src/native_shell.rs b/crates/tinybox-bus/src/native_shell.rs new file mode 100644 index 00000000..df725606 --- /dev/null +++ b/crates/tinybox-bus/src/native_shell.rs @@ -0,0 +1,455 @@ +//! Runtime-neutral orchestration for one supervised native shell execution. + +use std::{ + future::Future, + pin::Pin, + time::{Duration, Instant}, +}; + +use futures_util::{FutureExt, pin_mut, select}; + +use crate::{ + CreateRequest, ExecutionState, OutputBatch, OutputChunk, ProcessRef, ReserveRequest, + ResourceId, ShellExecRequest, +}; + +/// Transport operations required to supervise a native shell. +/// +/// Adapters implement these methods with their own bus connection and runtime. +pub trait NativeShellClient: Sync { + /// Transport error returned by one bus operation. + type Error; + + /// Reserve a resource or process identifier. + fn reserve( + &self, + request: ReserveRequest, + ) -> impl Future> + Send; + /// Create the resource with the module-issued reservation. + fn create( + &self, + request: CreateRequest, + ) -> impl Future> + Send; + /// Start the requested native shell and return its owned process reference. + fn start_native_shell_exec( + &self, + request: ShellExecRequest, + ) -> impl Future> + Send; + /// Read output starting at the supplied sequence cursor. + fn read_output( + &self, + process: ProcessRef, + cursor: u64, + ) -> impl Future> + Send; + /// Cancel and reap a running native process. + fn cancel(&self, process: ProcessRef) -> impl Future> + Send; + /// Release retained output after the process reaches a terminal state. + fn release_output( + &self, + process: ProcessRef, + ) -> impl Future> + Send; + /// Close the resource after process cleanup. + fn close(&self, resource: ResourceId) -> impl Future> + Send; +} + +/// Inputs for resource creation and native shell execution. +/// +/// The `resource` and process reservation fields in `create` and `shell` are +/// replaced by identifiers reserved during the run. +#[derive(Debug, Clone)] +pub struct NativeShellRunRequest { + /// Resource configuration; its reservation identifier is ignored. + pub create: CreateRequest, + /// Shell settings; its resource and process identifiers are ignored. + pub shell: ShellExecRequest, + /// Return a running process after this duration if it is shorter than the timeout. + pub yield_after: Option, +} + +/// Collected native shell output after the module acknowledges process cleanup. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NativeShellOutput { + /// Process exit code, including nonzero exits. + pub exit_code: i32, + /// Complete stdout bytes. + pub stdout: Vec, + /// Complete stderr bytes. + pub stderr: Vec, + /// Whether the shell exceeded its request timeout. + pub timed_out: bool, +} + +/// Outcome of running a shell through `TinyBox`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NativeShellRunOutcome { + /// The process exited or timed out and its output was released. + Finished(NativeShellOutput), + /// The process is still running; the caller now owns its cleanup lifecycle. + Running(NativeShellRunning), +} + +/// A running process handed back at the configured yield point. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NativeShellRunning { + /// Resource that must be closed after the process is finished. + pub resource: ResourceId, + /// Owned process reference for polling, cancellation, or release. + pub process: ProcessRef, + /// Cursor for the next output read. + pub cursor: u64, + /// Stdout collected before yielding. + pub stdout: Vec, + /// Stderr collected before yielding. + pub stderr: Vec, +} + +/// Failure during one phase of native shell orchestration. +#[derive(Debug)] +pub enum NativeShellError { + /// A bus operation failed. + Transport { + /// Human-readable operation that failed. + operation: &'static str, + /// Underlying adapter error. + source: E, + }, + /// The module reported a terminal failed state. + Execution(crate::ExecutionFailure), + /// An output batch did not advance its cursor despite containing chunks. + InvalidOutputCursor, + /// Cancellation failed, so native cleanup is not acknowledged. + Cancellation(E), + /// Output release failed after the process had stopped. + OutputRelease(E), + /// Resource close failed after process cleanup. + ResourceClose(E), + /// Cleanup failed while another operation was already failing. + Cleanup { + /// Cleanup operation that failed. + operation: &'static str, + /// Cleanup adapter error. + source: E, + /// Earlier failure that led to cleanup. + during: Box, + }, +} + +/// Run one native shell and stream every output chunk to `on_chunk`. +/// +/// The caller must keep this future supervised while signaling `cancelled`; +/// dropping it cannot await native process cleanup. The helper reserves and +/// creates the resource, starts the process, polls replayable output, cancels +/// on timeout or caller cancellation, releases output, and closes the resource. +/// +/// # Errors +/// Returns the transport failure, module execution failure, or cleanup failure. +pub async fn run_native_shell( + client: &C, + request: NativeShellRunRequest, + cancelled: Fut, + mut on_chunk: F, +) -> Result> +where + C: NativeShellClient, + F: FnMut(&OutputChunk), + Fut: Future + Send, + F: Send, + C::Error: Send, +{ + let timeout_ms = request.shell.timeout_ms.max(1); + let yield_after = request + .yield_after + .filter(|after| *after < Duration::from_millis(timeout_ms)); + let yield_started = Instant::now(); + let resource = client + .reserve(ReserveRequest::Resource) + .await + .map_err(|source| NativeShellError::Transport { + operation: "reserve resource", + source, + })?; + let result = async { + let process = + start_native_shell(client, resource.clone(), request.create, request.shell).await?; + collect_native_shell( + client, + process, + NativeShellCollectionConfig { + resource: resource.clone(), + timeout_ms, + yield_after, + yield_started, + }, + cancelled, + &mut on_chunk, + ) + .await + } + .await; + let result = match result { + Ok(NativeShellRunOutcome::Running(running)) => { + return Ok(NativeShellRunOutcome::Running(running)); + } + result => result, + }; + let close_result = client.close(resource).await; + match (result, close_result) { + (Ok(output), Ok(())) => Ok(output), + (Err(error), Ok(())) => Err(error), + (Ok(_), Err(source)) => Err(NativeShellError::ResourceClose(source)), + (Err(error), Err(source)) => Err(NativeShellError::Cleanup { + operation: "close resource", + source, + during: Box::new(error), + }), + } +} + +async fn start_native_shell( + client: &C, + resource: ResourceId, + mut create: CreateRequest, + mut shell: ShellExecRequest, +) -> Result> { + create.resource = resource.clone(); + client + .create(create) + .await + .map_err(|source| NativeShellError::Transport { + operation: "create resource", + source, + })?; + let process_id = client + .reserve(ReserveRequest::Process(resource.clone())) + .await + .map_err(|source| NativeShellError::Transport { + operation: "reserve process", + source, + })?; + shell.resource = resource; + shell.process = process_id; + client + .start_native_shell_exec(shell) + .await + .map_err(|source| NativeShellError::Transport { + operation: "start native shell", + source, + }) +} + +async fn collect_native_shell( + client: &C, + process: ProcessRef, + config: NativeShellCollectionConfig, + cancelled: Fut, + on_chunk: &mut F, +) -> Result> +where + C: NativeShellClient, + F: FnMut(&OutputChunk), + Fut: Future + Send, + F: Send, + C::Error: Send, +{ + let deadline = Instant::now() + Duration::from_millis(config.timeout_ms); + let mut capture = NativeShellCapture::default(); + let cancelled = cancelled.fuse(); + pin_mut!(cancelled); + let mut cancellation_requested = false; + let mut timed_out = false; + loop { + let poll = next_native_shell_batch( + client, + &process, + capture.cursor, + deadline, + cancellation_requested, + cancelled.as_mut(), + ) + .await?; + match poll { + NativeShellPoll::Cancelled => { + cancellation_requested = true; + } + NativeShellPoll::TimedOut => { + cancellation_requested = true; + timed_out = true; + } + NativeShellPoll::Batch(batch) => { + let has_chunks = !batch.chunks.is_empty(); + if !capture.append(&batch, on_chunk) { + let primary = NativeShellError::InvalidOutputCursor; + if let Err(source) = client.cancel(process.clone()).await { + return Err(NativeShellError::Cleanup { + operation: "cancel after invalid output cursor", + source, + during: Box::new(primary), + }); + } + if let Err(source) = client.release_output(process.clone()).await { + return Err(NativeShellError::Cleanup { + operation: "release output after invalid cursor", + source, + during: Box::new(primary), + }); + } + return Err(primary); + } + if has_chunks && !matches!(&batch.state, ExecutionState::Running) { + continue; + } + if let Some(output) = + finish_native_shell(client, &process, &mut capture, batch.state, timed_out) + .await? + { + return Ok(NativeShellRunOutcome::Finished(output)); + } + if !cancellation_requested + && config + .yield_after + .is_some_and(|after| config.yield_started.elapsed() >= after) + { + return Ok(NativeShellRunOutcome::Running(NativeShellRunning { + resource: config.resource, + process, + cursor: capture.cursor, + stdout: capture.stdout, + stderr: capture.stderr, + })); + } + } + } + } +} + +async fn finish_native_shell( + client: &C, + process: &ProcessRef, + capture: &mut NativeShellCapture, + state: ExecutionState, + timed_out: bool, +) -> Result, NativeShellError> { + let output = match state { + ExecutionState::Running => return Ok(None), + ExecutionState::Exited { exit_code } => capture.finish(exit_code, timed_out), + ExecutionState::Cancelled => capture.finish(-1, timed_out), + ExecutionState::Failed { + code: crate::ExecutionFailure::TimedOut, + } => capture.finish(-1, true), + ExecutionState::Failed { code } => { + client + .release_output(process.clone()) + .await + .map_err(NativeShellError::OutputRelease)?; + return Err(NativeShellError::Execution(code)); + } + }; + client + .release_output(process.clone()) + .await + .map_err(NativeShellError::OutputRelease)?; + Ok(Some(output)) +} + +struct NativeShellCollectionConfig { + resource: ResourceId, + timeout_ms: u64, + yield_after: Option, + yield_started: Instant, +} + +#[derive(Default)] +struct NativeShellCapture { + cursor: u64, + stdout: Vec, + stderr: Vec, +} + +impl NativeShellCapture { + fn append(&mut self, batch: &OutputBatch, on_chunk: &mut F) -> bool + where + F: FnMut(&OutputChunk), + { + if !batch.chunks.is_empty() && batch.next_sequence <= self.cursor { + return false; + } + for chunk in &batch.chunks { + on_chunk(chunk); + match chunk.stream { + crate::OutputStream::Stdout => self.stdout.extend_from_slice(&chunk.bytes), + crate::OutputStream::Stderr => self.stderr.extend_from_slice(&chunk.bytes), + } + } + self.cursor = batch.next_sequence; + true + } + + fn finish(&mut self, exit_code: i32, timed_out: bool) -> NativeShellOutput { + NativeShellOutput { + exit_code, + stdout: std::mem::take(&mut self.stdout), + stderr: std::mem::take(&mut self.stderr), + timed_out, + } + } +} + +enum NativeShellPoll { + Batch(OutputBatch), + Cancelled, + TimedOut, +} + +async fn next_native_shell_batch( + client: &C, + process: &ProcessRef, + cursor: u64, + deadline: Instant, + cancellation_requested: bool, + mut cancelled: Pin<&mut futures_util::future::Fuse>, +) -> Result> +where + C: NativeShellClient, + C::Error: Send, + Fut: Future + Send, +{ + let remaining = deadline.saturating_duration_since(Instant::now()); + let poll = client.read_output(process.clone(), cursor).fuse(); + let delay = futures_timer::Delay::new(if cancellation_requested { + Duration::from_millis(20) + } else { + remaining.min(Duration::from_millis(20)) + }) + .fuse(); + pin_mut!(poll, delay); + if cancellation_requested { + select! { + result = poll => result + .map(NativeShellPoll::Batch) + .map_err(|source| NativeShellError::Transport { operation: "read output", source }), + } + } else { + select! { + result = poll => match result { + Ok(batch) => Ok(NativeShellPoll::Batch(batch)), + Err(source) => { + client.cancel(process.clone()).await.map_err(NativeShellError::Cancellation)?; + client.release_output(process.clone()).await.map_err(NativeShellError::OutputRelease)?; + Err(NativeShellError::Transport { operation: "read output", source }) + } + }, + () = cancelled => { + client.cancel(process.clone()).await.map_err(NativeShellError::Cancellation)?; + Ok(NativeShellPoll::Cancelled) + }, + () = delay => { + client.cancel(process.clone()).await.map_err(NativeShellError::Cancellation)?; + Ok(NativeShellPoll::TimedOut) + } + } + } +} + +#[cfg(test)] +#[path = "native_shell_tests.rs"] +mod tests; diff --git a/crates/tinybox-bus/src/native_shell_tests.rs b/crates/tinybox-bus/src/native_shell_tests.rs new file mode 100644 index 00000000..16079606 --- /dev/null +++ b/crates/tinybox-bus/src/native_shell_tests.rs @@ -0,0 +1,569 @@ +//! Tests for runtime-neutral native shell orchestration. +use super::*; +use crate::{ + ExecutionFailure, HostConfig, NetworkPolicy, OutputStream, ResourceLimits, SelectShellRequest, + Workspace, +}; +use std::{collections::BTreeMap, sync::Mutex}; + +struct MockClient { + calls: Mutex>, + reads: Mutex, + behavior: MockBehavior, +} + +#[derive(Clone, Copy)] +enum MockBehavior { + Normal, + FailCreate, + Cancelled, + ModuleTimeout, + Running, + FailResourceReserve, + FailProcessReserve, + FailStart, + FailRead, + FailCancel, + FailRelease, + FailClose, + FailedExecution, + InvalidCursor, + FailedExecutionCloseFail, + TimedOut, + InvalidCursorCancelFail, + InvalidCursorReleaseFail, + RunningStderr, +} + +impl NativeShellClient for MockClient { + type Error = &'static str; + + async fn reserve(&self, request: ReserveRequest) -> Result { + std::future::ready(()).await; + self.calls.lock().map_err(|_| "poisoned")?.push("reserve"); + if matches!(self.behavior, MockBehavior::FailResourceReserve) + && matches!(request, ReserveRequest::Resource) + || matches!(self.behavior, MockBehavior::FailProcessReserve) + && matches!(request, ReserveRequest::Process(_)) + { + return Err("reserve failed"); + } + Ok(match request { + ReserveRequest::Resource => ResourceId("resource-1".into()), + ReserveRequest::Process(_) => ResourceId("process-1".into()), + _ => return Err("unexpected reservation"), + }) + } + + async fn create(&self, request: CreateRequest) -> Result<(), Self::Error> { + std::future::ready(()).await; + self.calls.lock().map_err(|_| "poisoned")?.push("create"); + assert_eq!(request.resource, ResourceId("resource-1".into())); + if matches!(self.behavior, MockBehavior::FailCreate) { + return Err("create failed"); + } + Ok(()) + } + + async fn start_native_shell_exec( + &self, + request: ShellExecRequest, + ) -> Result { + std::future::ready(()).await; + self.calls.lock().map_err(|_| "poisoned")?.push("start"); + assert_eq!(request.process, ResourceId("process-1".into())); + if matches!(self.behavior, MockBehavior::FailStart) { + return Err("start failed"); + } + Ok(ProcessRef { + resource: request.resource, + process: request.process, + }) + } + + async fn read_output(&self, _: ProcessRef, cursor: u64) -> Result { + std::future::ready(()).await; + self.calls.lock().map_err(|_| "poisoned")?.push("read"); + let cancelled = self + .calls + .lock() + .map_err(|_| "poisoned")? + .contains(&"cancel"); + if matches!(self.behavior, MockBehavior::TimedOut) && !cancelled { + std::future::pending::<()>().await; + } + if matches!(self.behavior, MockBehavior::FailRead) { + return Err("read failed"); + } + let mut reads = self.reads.lock().map_err(|_| "poisoned")?; + *reads = reads.wrapping_add(1); + let state = if matches!( + self.behavior, + MockBehavior::Cancelled | MockBehavior::TimedOut + ) { + if cancelled { + ExecutionState::Cancelled + } else { + ExecutionState::Running + } + } else if matches!(self.behavior, MockBehavior::ModuleTimeout) { + ExecutionState::Failed { + code: ExecutionFailure::TimedOut, + } + } else if matches!(self.behavior, MockBehavior::RunningStderr) && cancelled { + ExecutionState::Cancelled + } else if matches!( + self.behavior, + MockBehavior::Running | MockBehavior::RunningStderr + ) { + ExecutionState::Running + } else if matches!( + self.behavior, + MockBehavior::FailedExecution | MockBehavior::FailedExecutionCloseFail + ) { + ExecutionState::Failed { + code: ExecutionFailure::OutputLimit, + } + } else { + ExecutionState::Exited { exit_code: 7 } + }; + Ok(OutputBatch { + chunks: if cursor == 0 { + vec![OutputChunk { + sequence: 0, + stream: if matches!( + self.behavior, + MockBehavior::RunningStderr + | MockBehavior::InvalidCursor + | MockBehavior::InvalidCursorCancelFail + | MockBehavior::InvalidCursorReleaseFail + ) { + OutputStream::Stderr + } else { + OutputStream::Stdout + }, + bytes: b"done".to_vec(), + }] + } else { + Vec::new() + }, + next_sequence: if matches!( + self.behavior, + MockBehavior::InvalidCursor + | MockBehavior::InvalidCursorCancelFail + | MockBehavior::InvalidCursorReleaseFail + ) { + 0 + } else if cursor == 0 { + 1 + } else { + cursor + }, + state, + }) + } + + async fn cancel(&self, _: ProcessRef) -> Result<(), Self::Error> { + std::future::ready(()).await; + self.calls.lock().map_err(|_| "poisoned")?.push("cancel"); + if matches!( + self.behavior, + MockBehavior::FailCancel | MockBehavior::InvalidCursorCancelFail + ) { + return Err("cancel failed"); + } + Ok(()) + } + + async fn release_output(&self, _: ProcessRef) -> Result<(), Self::Error> { + std::future::ready(()).await; + self.calls.lock().map_err(|_| "poisoned")?.push("release"); + if matches!( + self.behavior, + MockBehavior::FailRelease | MockBehavior::InvalidCursorReleaseFail + ) { + return Err("release failed"); + } + Ok(()) + } + + async fn close(&self, _: ResourceId) -> Result<(), Self::Error> { + std::future::ready(()).await; + self.calls.lock().map_err(|_| "poisoned")?.push("close"); + if matches!( + self.behavior, + MockBehavior::FailClose | MockBehavior::FailedExecutionCloseFail + ) { + return Err("close failed"); + } + Ok(()) + } +} + +#[test] +fn runs_and_cleans_up_a_shell_while_streaming_chunks() -> Result<(), &'static str> { + futures_executor::block_on(runs_and_cleans_up_a_shell()) +} + +async fn runs_and_cleans_up_a_shell() -> Result<(), &'static str> { + let client = MockClient { + calls: Mutex::new(Vec::new()), + reads: Mutex::new(0), + behavior: MockBehavior::Normal, + }; + let mut chunks = Vec::new(); + let output = run_native_shell(&client, request(), std::future::pending(), |chunk| { + chunks.push(chunk.clone()); + }) + .await + .map_err(|_| "run failed")?; + let NativeShellRunOutcome::Finished(output) = output else { + return Err("expected a finished outcome"); + }; + assert_eq!(output.exit_code, 7); + assert_eq!(output.stdout, b"done"); + assert!(!output.timed_out); + assert_eq!(chunks.len(), 1); + assert_eq!( + *client.calls.lock().map_err(|_| "poisoned")?, + [ + "reserve", "create", "reserve", "start", "read", "read", "release", "close" + ] + ); + Ok(()) +} + +#[test] +fn cancellation_reaps_the_process_before_releasing_and_closing() -> Result<(), &'static str> { + futures_executor::block_on(async { + let client = MockClient { + calls: Mutex::new(Vec::new()), + reads: Mutex::new(0), + behavior: MockBehavior::Cancelled, + }; + let output = run_native_shell(&client, request(), cancel_now(), ignore_chunk) + .await + .map_err(|_| "run failed")?; + let NativeShellRunOutcome::Finished(output) = output else { + return Err("expected a finished outcome"); + }; + assert_eq!(output.exit_code, -1); + assert!(!output.timed_out); + let calls = client.calls.lock().map_err(|_| "poisoned")?; + assert_eq!(&calls[..4], &["reserve", "create", "reserve", "start"]); + let cancel = calls + .iter() + .position(|call| *call == "cancel") + .ok_or("cancel missing")?; + let release = calls + .iter() + .position(|call| *call == "release") + .ok_or("release missing")?; + let close = calls + .iter() + .position(|call| *call == "close") + .ok_or("close missing")?; + assert!(cancel < release && release < close); + Ok(()) + }) +} + +#[test] +fn create_failure_still_releases_the_resource_reservation() -> Result<(), &'static str> { + futures_executor::block_on(async { + let client = MockClient { + calls: Mutex::new(Vec::new()), + reads: Mutex::new(0), + behavior: MockBehavior::FailCreate, + }; + let error = run_native_shell(&client, request(), never_cancel(), ignore_chunk) + .await + .err() + .ok_or("expected failure")?; + assert!(matches!( + error, + NativeShellError::Transport { + operation: "create resource", + .. + } + )); + assert_eq!( + *client.calls.lock().map_err(|_| "poisoned")?, + ["reserve", "create", "close"] + ); + Ok(()) + }) +} + +#[test] +fn module_timeout_returns_collected_output_as_a_result() -> Result<(), &'static str> { + futures_executor::block_on(async { + let client = MockClient { + calls: Mutex::new(Vec::new()), + reads: Mutex::new(0), + behavior: MockBehavior::ModuleTimeout, + }; + let outcome = run_native_shell(&client, request(), never_cancel(), ignore_chunk) + .await + .map_err(|_| "run failed")?; + let NativeShellRunOutcome::Finished(output) = outcome else { + return Err("expected a finished outcome"); + }; + assert!(output.timed_out); + assert_eq!(output.stdout, b"done"); + assert_eq!(output.exit_code, -1); + Ok(()) + }) +} + +#[test] +fn yield_preserves_process_resource_cursor_and_partial_output() -> Result<(), &'static str> { + futures_executor::block_on(async { + let client = MockClient { + calls: Mutex::new(Vec::new()), + reads: Mutex::new(0), + behavior: MockBehavior::Running, + }; + let mut request = request(); + request.yield_after = Some(std::time::Duration::ZERO); + let outcome = run_native_shell(&client, request, never_cancel(), ignore_chunk) + .await + .map_err(|_| "run failed")?; + let NativeShellRunOutcome::Running(running) = outcome else { + return Err("expected a running outcome"); + }; + assert_eq!(running.resource, ResourceId("resource-1".into())); + assert_eq!(running.process.process, ResourceId("process-1".into())); + assert_eq!(running.cursor, 1); + assert_eq!(running.stdout, b"done"); + assert_eq!( + *client.calls.lock().map_err(|_| "poisoned")?, + ["reserve", "create", "reserve", "start", "read"] + ); + Ok(()) + }) +} + +#[test] +fn reserve_start_and_read_failures_are_reported_and_cleaned_up() -> Result<(), &'static str> { + for (behavior, operation) in [ + (MockBehavior::FailResourceReserve, "reserve resource"), + (MockBehavior::FailProcessReserve, "reserve process"), + (MockBehavior::FailStart, "start native shell"), + (MockBehavior::FailRead, "read output"), + ] { + futures_executor::block_on(async { + let client = mock(behavior); + let error = run_native_shell(&client, request(), never_cancel(), ignore_chunk) + .await + .err() + .ok_or("operation should fail")?; + if operation == "read output" { + assert!(matches!( + error, + NativeShellError::Transport { + operation: "read output", + .. + } + )); + assert!( + client + .calls + .lock() + .map_err(|_| "poisoned")? + .contains(&"cancel") + ); + assert!( + client + .calls + .lock() + .map_err(|_| "poisoned")? + .contains(&"release") + ); + } else { + assert!( + matches!(error, NativeShellError::Transport { operation: actual, .. } if actual == operation) + ); + } + Ok::<(), &'static str>(()) + })?; + } + Ok(()) +} + +#[test] +fn terminal_execution_and_cleanup_failures_keep_the_primary_error() -> Result<(), &'static str> { + futures_executor::block_on(async { + let client = mock(MockBehavior::FailedExecution); + assert!(matches!( + run_native_shell(&client, request(), never_cancel(), ignore_chunk).await, + Err(NativeShellError::Execution(ExecutionFailure::OutputLimit)) + )); + Ok::<(), &'static str>(()) + })?; + futures_executor::block_on(async { + let client = mock(MockBehavior::FailClose); + assert!(matches!( + run_native_shell(&client, request(), never_cancel(), ignore_chunk).await, + Err(NativeShellError::ResourceClose("close failed")) + )); + Ok::<(), &'static str>(()) + })?; + futures_executor::block_on(async { + let client = mock(MockBehavior::FailedExecutionCloseFail); + let err = run_native_shell(&client, request(), never_cancel(), ignore_chunk) + .await + .err() + .ok_or("expected cleanup failure")?; + assert!(matches!( + err, + NativeShellError::Cleanup { + operation: "close resource", + .. + } + )); + Ok::<(), &'static str>(()) + })?; + Ok(()) +} + +#[test] +fn invalid_cursor_and_cancel_cleanup_errors_are_reported() { + futures_executor::block_on(async { + let client = mock(MockBehavior::InvalidCursor); + assert!(matches!( + run_native_shell(&client, request(), never_cancel(), ignore_chunk).await, + Err(NativeShellError::InvalidOutputCursor) + )); + }); + futures_executor::block_on(async { + let client = mock(MockBehavior::FailCancel); + assert!(matches!( + run_native_shell(&client, request(), cancel_now(), ignore_chunk).await, + Err(NativeShellError::Cancellation("cancel failed")) + )); + }); + futures_executor::block_on(async { + let client = mock(MockBehavior::FailRelease); + assert!(matches!( + run_native_shell(&client, request(), never_cancel(), ignore_chunk).await, + Err(NativeShellError::OutputRelease("release failed")) + )); + }); + futures_executor::block_on(async { + let client = mock(MockBehavior::InvalidCursorCancelFail); + assert!(matches!( + run_native_shell(&client, request(), never_cancel(), ignore_chunk).await, + Err(NativeShellError::Cleanup { + operation: "cancel after invalid output cursor", + .. + }) + )); + }); + futures_executor::block_on(async { + let client = mock(MockBehavior::InvalidCursorReleaseFail); + assert!(matches!( + run_native_shell(&client, request(), never_cancel(), ignore_chunk).await, + Err(NativeShellError::Cleanup { + operation: "release output after invalid cursor", + .. + }) + )); + }); +} + +#[test] +fn timeout_cancels_and_returns_collected_output() -> Result<(), &'static str> { + futures_executor::block_on(async { + let client = mock(MockBehavior::TimedOut); + let mut req = request(); + req.shell.timeout_ms = 1; + let outcome = run_native_shell(&client, req, never_cancel(), ignore_chunk) + .await + .map_err(|_| "run failed")?; + let NativeShellRunOutcome::Finished(output) = outcome else { + return Err("expected completion"); + }; + assert!(output.timed_out); + assert_eq!(output.exit_code, -1); + assert!( + client + .calls + .lock() + .map_err(|_| "poisoned")? + .contains(&"cancel") + ); + Ok(()) + }) +} + +#[test] +fn running_poll_without_yield_accumulates_stderr_until_timeout() -> Result<(), &'static str> { + futures_executor::block_on(async { + let client = mock(MockBehavior::RunningStderr); + let mut req = request(); + req.shell.timeout_ms = 1; + req.yield_after = Some(std::time::Duration::from_secs(5)); + let outcome = run_native_shell(&client, req, never_cancel(), ignore_chunk) + .await + .map_err(|_| "run failed")?; + let NativeShellRunOutcome::Finished(output) = outcome else { + return Err("expected completion"); + }; + assert!(output.timed_out); + assert_eq!(output.stderr, b"done"); + assert_eq!(output.stdout.len(), 0); + Ok(()) + }) +} + +fn mock(behavior: MockBehavior) -> MockClient { + MockClient { + calls: Mutex::new(Vec::new()), + reads: Mutex::new(0), + behavior, + } +} + +fn never_cancel() -> impl std::future::Future { + std::future::pending::<()>() +} + +fn cancel_now() -> impl std::future::Future { + std::future::ready(()) +} + +fn ignore_chunk(_: &OutputChunk) {} + +fn request() -> NativeShellRunRequest { + NativeShellRunRequest { + create: CreateRequest { + resource: ResourceId("placeholder-resource".into()), + backend: "passthrough".into(), + host: HostConfig::Local, + workspace: Workspace::Directory("/workspace".into()), + network: NetworkPolicy::Open, + resources: ResourceLimits::default(), + ports: Vec::new(), + env: BTreeMap::new(), + }, + shell: ShellExecRequest { + process: ResourceId("placeholder-process".into()), + resource: ResourceId("placeholder-resource".into()), + shell: SelectShellRequest { + command: "echo done".into(), + preserve_pipeline_status: false, + runtime_path: None, + python_utf8: false, + }, + cwd: None, + extra_env: BTreeMap::new(), + state_dir: "/state".into(), + timeout_ms: 1_000, + stdin: None, + jail: None, + hide_window: false, + }, + yield_after: None, + } +} diff --git a/crates/tinybox-jail/src/detect.rs b/crates/tinybox-jail/src/detect.rs index c1a8d801..3ed3a39d 100644 --- a/crates/tinybox-jail/src/detect.rs +++ b/crates/tinybox-jail/src/detect.rs @@ -62,6 +62,27 @@ pub fn pick_backend() -> Arc { pick_from(candidates()) } +/// Returns the stable name of the backend selected by platform detection. +#[must_use] +pub fn backend_name() -> &'static str { + pick_backend().name() +} + +/// Reports whether this build and running Linux kernel support Landlock. +/// +/// This is always false off Linux and when the `landlock` feature is disabled. +#[must_use] +pub fn landlock_available() -> bool { + #[cfg(target_os = "linux")] + { + crate::linux::LandlockBackend::new().is_available() + } + #[cfg(not(target_os = "linux"))] + { + false + } +} + /// Select an available backend from the ordered platform candidates. fn pick_from(backends: Vec>) -> Arc { for backend in backends { diff --git a/crates/tinybox-jail/src/detect_tests.rs b/crates/tinybox-jail/src/detect_tests.rs index b943f705..6c760c46 100644 --- a/crates/tinybox-jail/src/detect_tests.rs +++ b/crates/tinybox-jail/src/detect_tests.rs @@ -17,6 +17,22 @@ fn unavailable_backend_rejects_spawning() { #[test] fn backend_detection_returns_a_backend() { assert_ne!(pick_backend().name().len(), 0); + assert_eq!(backend_name(), pick_backend().name()); +} + +#[cfg(target_os = "linux")] +#[test] +fn landlock_detection_matches_the_landlock_backend() { + assert_eq!( + landlock_available(), + crate::linux::LandlockBackend::new().is_available() + ); +} + +#[cfg(not(target_os = "linux"))] +#[test] +fn landlock_is_unavailable_off_linux() { + assert!(!landlock_available()); } #[test]