diff --git a/crates/openhuman-core/src/modules/README.md b/crates/openhuman-core/src/modules/README.md index 0221166e453..84442b41c2e 100644 --- a/crates/openhuman-core/src/modules/README.md +++ b/crates/openhuman-core/src/modules/README.md @@ -288,7 +288,7 @@ download, its `dlopen`, or its resident memory. | [`registry.rs`](registry.rs) | The compiled-in `ALL` table and `find(id)`. Records live in [`registry/`](registry/README.md), one `records_*.rs` file per module family. | | [`types.rs`](types.rs) | `ModuleRecord`, `PlatformAsset`, `LoadPolicy`, `ModuleSource`, `ModuleState`, `ModuleStatus`. | | [`host.rs`](host.rs) | The module bus: the dedicated runtime, `ModuleRuntime`, `runtime()`, `is_started()`. | -| [`ops.rs`](ops.rs) | Resolution (`ensure_loaded*`, `resolve`, `load_cached`, `load_local`), per-module configuration, overrides, the bundled and install directories, and status reporting (`list`, `state_of`). | +| [`ops.rs`](ops.rs) | Resolution (`ensure_loaded*`, `resolve`, `load_cached`, `load_local`), per-module configuration, overrides, the bundled and install directories, and status reporting (`list`, `state_of`). Modules that take host callbacks (TinySecurity, TinyMCP, TinyEventQueues, TinyChannels, TinyComputer) load on the module runtime through [`ops/callback_load.rs`](ops/callback_load.rs): `load` for the storage and computer first loads, `confirm` for every later load; TinySecurity's first load prepares its own per-candidate state in [`security/host_callbacks.rs`](security/host_callbacks.rs) over the same `Resolver::ensure_loaded_with_host_callback_factory`. Each takes a local-file override or staged artifact first, else the pinned release from the bundle, then the digest-verified release cache, then a download into it. | | [`boot.rs`](boot.rs) | Startup loading: search-path artifacts, then eager records. | | [`schemas.rs`](schemas.rs) | The `modules` RPC namespace and its handlers. | diff --git a/crates/openhuman-core/src/modules/ops.rs b/crates/openhuman-core/src/modules/ops.rs index 6a476f0d6c3..1b2e3053149 100644 --- a/crates/openhuman-core/src/modules/ops.rs +++ b/crates/openhuman-core/src/modules/ops.rs @@ -39,6 +39,7 @@ use tinybus::module::{ }; use tinybus::Proxy; +pub(super) mod callback_load; mod list; pub use list::list; @@ -155,10 +156,13 @@ pub fn resolver(config: &Config) -> Resolver { } } ModuleSource::Release if requires_verified_local_callbacks(record.id) => { - // These modules require a per-module storage callback, which - // TinyBus can bind only on its verified local-file path. - // Route a staged release through that rung; the callback-aware - // client below supplies the unique destination at load time. + // A staged artifact (`OPENHUMAN_MODULE_SOURCE_DIR` or + // `OPENHUMAN_MODULE_PATH`) wins over the release, as a local + // file: the search-path rung would admit it without the + // callbacks these modules need. With none staged the source + // stays the pinned release, which the callback loads in + // `callback_load` take from the installer bundle or the + // digest-verified release cache. if let Some(path) = source::local_artifact(record.id) { sources.insert( record.id.to_string(), @@ -251,41 +255,18 @@ type StorageCallbackLoads = tokio::sync::Mutex< static STORAGE_CALLBACK_LOADS: std::sync::LazyLock = std::sync::LazyLock::new(|| tokio::sync::Mutex::new(std::collections::HashMap::new())); +/// Load a storage-backed module with its storage callback, on the module +/// runtime so a caller that stops waiting cannot strand the binding. async fn ensure_loaded_with_storage_callback( resolver: &Resolver, config: &Config, id: &str, ) -> Result<(), String> { - let mut loads = STORAGE_CALLBACK_LOADS.lock().await; - if let Some((binding, _served)) = loads.get(id) { - if matches!(state_of(id), ModuleState::Ready) { - // Already serving with this binding: a binding is spent once the - // module takes it, so re-presenting it would be refused. The module - // is process-wide, so every later workspace's client finds it - // already serving. - return Ok(()); - } - let mut module_config = module_config(config, id).await; - module_config["storage_callback"] = serde_json::json!(binding.unique_name().as_str()); - return resolver - .ensure_loaded_with_host_callbacks(id, std::sync::Arc::clone(binding), module_config) - .await; - } - - let runtime = resolver - .runtime() - .await - .map_err(|error| error.to_string())?; - let (binding, served) = super::storage_host::prepare_module_callbacks(runtime) - .await - .map_err(|error| error.to_string())?; - let mut module_config = module_config(config, id).await; - module_config["storage_callback"] = serde_json::json!(binding.unique_name().as_str()); - resolver - .ensure_loaded_with_host_callbacks(id, std::sync::Arc::clone(&binding), module_config) - .await?; - loads.insert(id.to_owned(), (binding, served)); - Ok(()) + let (task_resolver, config, id) = (resolver.clone(), config.clone(), id.to_owned()); + callback_load::detached(resolver, async move { + callback_load::load_with_storage_callback(&task_resolver, &config, &id).await + }) + .await } #[cfg(feature = "security-module")] @@ -342,7 +323,11 @@ async fn ensure_loaded_with_security_callback( config: &Config, id: &str, ) -> Result<(), String> { - super::security::host_callbacks::ensure_loaded(resolver, config, id).await + let (task_resolver, config, id) = (resolver.clone(), config.clone(), id.to_owned()); + callback_load::detached(resolver, async move { + super::security::host_callbacks::ensure_loaded(&task_resolver, &config, &id).await + }) + .await } #[cfg(feature = "security-module")] @@ -351,27 +336,11 @@ async fn ensure_loaded_with_computer_callback( config: &Config, id: &str, ) -> Result<(), String> { - let mut loads = COMPUTER_CALLBACK_LOADS.lock().await; - if let Some(binding) = loads.get(id) { - let module_config = module_config_with_consequence_callback(config, id, binding).await; - return resolver - .ensure_loaded_with_host_callbacks(id, std::sync::Arc::clone(binding), module_config) - .await; - } - - let runtime = resolver - .runtime() - .await - .map_err(|error| error.to_string())?; - let binding = super::browser_egress_host::prepare_consequence_callback(runtime) - .await - .map_err(|error| error.to_string())?; - let module_config = module_config_with_consequence_callback(config, id, &binding).await; - resolver - .ensure_loaded_with_host_callbacks(id, std::sync::Arc::clone(&binding), module_config) - .await?; - loads.insert(id.to_owned(), binding); - Ok(()) + let (task_resolver, config, id) = (resolver.clone(), config.clone(), id.to_owned()); + callback_load::detached(resolver, async move { + callback_load::load_with_computer_callback(&task_resolver, &config, &id).await + }) + .await } #[cfg(feature = "security-module")] @@ -712,3 +681,7 @@ mod client_tests; #[cfg(test)] #[path = "ops_resolver_tests.rs"] mod resolver_tests; + +#[cfg(test)] +#[path = "ops_callback_release_tests.rs"] +mod callback_release_tests; diff --git a/crates/openhuman-core/src/modules/ops/callback_load.rs b/crates/openhuman-core/src/modules/ops/callback_load.rs new file mode 100644 index 00000000000..a235b96879b --- /dev/null +++ b/crates/openhuman-core/src/modules/ops/callback_load.rs @@ -0,0 +1,234 @@ +//! Loading a module that takes host callbacks, from whichever source serves it. +//! +//! TinyBus binds host callbacks to a module before its `init` runs, and only +//! on paths that keep the admitted identity: an explicit local file (a `path` +//! override, a source build, or a staged `OPENHUMAN_MODULE_PATH` release) and +//! the pinned release chain itself. The release chain tries the installer +//! bundle, then the release cache, then a download into the cache when +//! `modules.allow_download` is on, and every candidate passes the registry's +//! archive SHA-256 and, where the record pins one, its library SHA-256 before +//! a binding is offered to it. +//! +//! `Resolver::ensure_loaded_with_host_callback_factory` drives both, preparing +//! a fresh binding for each candidate so a refused candidate revokes only its +//! own. `Resolver::ensure_loaded_with_host_callbacks` takes one binding and +//! accepts only a local-file source, so a release-served module (every shipped +//! host without `OPENHUMAN_MODULE_PATH`) could neither be loaded nor +//! re-checked through it. Every callback-bound load goes through [`load`] and +//! every later check through [`confirm`] instead, each run through +//! [`detached`] so a caller that stops waiting cannot strand the binding's +//! host state. + +use std::collections::HashMap; +use std::future::Future; +use std::sync::{Arc, Mutex, PoisonError}; + +use crate::config::Config; +use crate::core::runtime::CoreContext; +use crate::modules::types::ModuleState; +use tinybus::module::{HostCallbackBinding, Resolver}; + +/// Run a callback load on the module runtime and wait for it. +/// +/// TinyBus finishes a resolution it has started even when every caller stops +/// waiting (`ensure_loaded_within` bounds only the wait). A callback load also +/// publishes the admitted binding's host state, a storage endpoint or a +/// consequence host, into a process-wide cache, and that publication must not +/// be cancelled with the caller either: the next caller would find the module +/// serving with a binding nobody holds the state for. Dropping the returned +/// future drops only the wait; `work` runs to completion. +/// +/// The task is scheduled before anything is awaited, so a caller that gives +/// up while the module bus is still installing its host callbacks does not +/// lose the load either: the task awaits them itself. It runs under the +/// caller's [`CoreContext`], captured here: TinySecurity's first load prepares +/// its secret scope inside `work`, and that resolves the caller's profile, +/// agent and storage scope from the ambient context. +/// +/// # Errors +/// +/// `work`'s error, the module bus failing to start, or the task panicking. +pub(in crate::modules) async fn detached( + resolver: &Resolver, + work: Fut, +) -> Result +where + T: Send + 'static, + Fut: Future> + Send + 'static, +{ + let work = CoreContext::propagate(work); + let resolver = resolver.clone(); + let task = crate::modules::runtime::spawn_on_bus(async move { + resolver + .runtime() + .await + .map_err(|error| error.to_string())?; + work.await + }) + .map_err(|error| error.to_string())?; + task.await + .map_err(|error| format!("the module callback load did not finish: {error}"))? +} + +/// One prepared candidate: its binding, the module configuration naming it, +/// and the host state that must live as long as the binding is retained. +pub(in crate::modules) type Prepared = (Arc, serde_json::Value, S); + +/// Load `id` with host callbacks, preparing a fresh set with `prepare` for +/// each candidate artifact the resolver tries. +/// +/// Returns the binding the admitted module retains and the state prepared +/// with it; the state of refused candidates is dropped. +/// +/// # Errors +/// +/// The resolver's message when no candidate is admitted, `prepare`'s message +/// when a binding cannot be prepared, or a refusal when the module is already +/// serving with a binding this call did not prepare. +pub(in crate::modules) async fn load( + resolver: &Resolver, + id: &str, + prepare: F, +) -> Result<(Arc, S), String> +where + S: Send + 'static, + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future, String>> + Send + 'static, +{ + let candidates: Arc>> = Arc::new(Mutex::new(HashMap::new())); + let factory_candidates = Arc::clone(&candidates); + let prepare = Arc::new(prepare); + let binding = resolver + .ensure_loaded_with_host_callback_factory(id, move |_attempt| { + let candidates = Arc::clone(&factory_candidates); + let prepare = Arc::clone(&prepare); + async move { + let (binding, module_config, state) = prepare().await?; + candidates + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(binding.unique_name().to_string(), state); + Ok((binding, module_config)) + } + }) + .await?; + let state = candidates + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(binding.unique_name().as_str()) + .ok_or_else(|| { + format!("module '{id}' is serving with a callback binding this load did not prepare") + })?; + Ok((binding, state)) +} + +/// Confirm `id` is still admitted with `expected`, preparing nothing. +/// +/// For a module an earlier [`load`] admitted: the resolver answers from the +/// retained binding without touching any source, so this works whether the +/// module came from a local file or from the release chain. +/// +/// # Errors +/// +/// The resolver's cached failure, or a refusal when the module is not serving +/// with `expected`. +pub(in crate::modules) async fn confirm( + resolver: &Resolver, + id: &str, + expected: &HostCallbackBinding, +) -> Result<(), String> { + let module_id = id.to_owned(); + let retained = resolver + .ensure_loaded_with_host_callback_factory(id, move |_attempt| { + let id = module_id.clone(); + async move { + Err(format!( + "module '{id}' is not serving with its admitted callback binding" + )) + } + }) + .await?; + if retained.unique_name() != expected.unique_name() { + return Err(format!( + "module '{id}' is serving with a different callback binding" + )); + } + Ok(()) +} + +/// The body [`super::ensure_loaded_with_storage_callback`] runs detached. +pub(super) async fn load_with_storage_callback( + resolver: &Resolver, + config: &Config, + id: &str, +) -> Result<(), String> { + let mut loads = super::STORAGE_CALLBACK_LOADS.lock().await; + if let Some((binding, _served)) = loads.get(id) { + if matches!(super::state_of(id), ModuleState::Ready) { + // Already serving with this binding: a binding is spent once the + // module takes it, so re-presenting it would be refused. The module + // is process-wide, so every later workspace's client finds it + // already serving. + return Ok(()); + } + return confirm(resolver, id, binding).await; + } + + let runtime = resolver + .runtime() + .await + .map_err(|error| error.to_string())?; + let config = config.clone(); + let module_id = id.to_owned(); + let (binding, served) = load(resolver, id, move || { + let config = config.clone(); + let id = module_id.clone(); + async move { + let (binding, served) = crate::modules::storage_host::prepare_module_callbacks(runtime) + .await + .map_err(|error| error.to_string())?; + let mut module_config = super::module_config(&config, &id).await; + module_config["storage_callback"] = serde_json::json!(binding.unique_name().as_str()); + Ok((binding, module_config, served)) + } + }) + .await?; + loads.insert(id.to_owned(), (binding, served)); + Ok(()) +} + +#[cfg(feature = "security-module")] +/// The body [`super::ensure_loaded_with_computer_callback`] runs detached. +pub(super) async fn load_with_computer_callback( + resolver: &Resolver, + config: &Config, + id: &str, +) -> Result<(), String> { + let mut loads = super::COMPUTER_CALLBACK_LOADS.lock().await; + if let Some(binding) = loads.get(id) { + return confirm(resolver, id, binding).await; + } + + let runtime = resolver + .runtime() + .await + .map_err(|error| error.to_string())?; + let config = config.clone(); + let module_id = id.to_owned(); + let (binding, ()) = load(resolver, id, move || { + let config = config.clone(); + let id = module_id.clone(); + async move { + let binding = + crate::modules::browser_egress_host::prepare_consequence_callback(runtime) + .await + .map_err(|error| error.to_string())?; + let module_config = + super::module_config_with_consequence_callback(&config, &id, &binding).await; + Ok((binding, module_config, ())) + } + }) + .await?; + loads.insert(id.to_owned(), binding); + Ok(()) +} diff --git a/crates/openhuman-core/src/modules/ops_callback_release_tests.rs b/crates/openhuman-core/src/modules/ops_callback_release_tests.rs new file mode 100644 index 00000000000..f9cd9a0a436 --- /dev/null +++ b/crates/openhuman-core/src/modules/ops_callback_release_tests.rs @@ -0,0 +1,434 @@ +//! A callback-bound module served from its pinned release. +//! +//! TinyMCP, TinyEventQueues, TinyChannels, TinySecurity and TinyComputer take +//! host callbacks when they load. On a host without `OPENHUMAN_MODULE_PATH` +//! (every shipped build and embedder) their source is the pinned release, so +//! the callback load has to come from the installer bundle or the release +//! cache, digest-checked, and never require a local-file override. +//! +//! The module here is `tests/fixtures/storage_probe_module`, built by the +//! storage host's cdylib tests and staged as a release: the archive and its +//! extracted library side by side under `////`, +//! the layout TinyBus fills after a verified download. Its record pins the +//! archive and library digests, so a wrong digest is a refusal and not a load. +//! It is loaded through [`ensure_loaded_with_storage_callback`], the path the +//! storage-backed modules take; TinySecurity and TinyComputer share its +//! `callback_load` helpers. +//! +//! Each test runs in its own process: the module runtime, the resolution table +//! and the probe's bus name are process-wide, and TinyBus never unloads a +//! library. + +use std::collections::HashMap; +use std::io::Write as _; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use super::{ensure_loaded_with_storage_callback, report, runtime}; +use crate::config::Config; +use tinybus::module::{ + artifact_dir, platform::host_candidates, LinkedRegistry, ModuleSource as ResolverSource, + ReleaseAsset, ReleaseRecord, Resolver, ResolverInputs, +}; + +const ID: &str = "callbackprobe"; +const VERSION: &str = "0.0.1"; +const ARCHIVE: &str = "callbackprobe-module-0.0.1.tar.gz"; +const BUS_NAME: &str = "ai.tinyhumans.test.StorageProbe"; +/// The module SDK derives the object path from the bus name. +const OBJECT_PATH: &str = "/ai/tinyhumans/test/StorageProbe"; +const WRONG_DIGEST: &str = "0000000000000000000000000000000000000000000000000000000000000000"; + +/// Re-run `test` alone in a fresh process; `true` means this is that process. +fn isolated(test: &str) -> bool { + const ISOLATED: &str = "OPENHUMAN_TEST_ISOLATED_CALLBACK_RELEASE"; + if std::env::var_os(ISOLATED).is_some() { + return true; + } + // Build the fixture here so the child does not race its own cargo build. + let _ = probe_library(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + &format!("modules::ops::callback_release_tests::{test}"), + "--test-threads=1", + ]) + .env(ISOLATED, "1") + .output() + .expect("run the callback release test in a fresh process"); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success() && stdout.contains("1 passed"), + "isolated callback release test failed\nstdout:\n{stdout}\nstderr:\n{}", + String::from_utf8_lossy(&output.stderr) + ); + false +} + +fn probe_library() -> &'static Path { + crate::modules::storage_host::tests::cdylib::probe_artifact() +} + +fn library_name() -> String { + probe_library() + .file_name() + .and_then(|name| name.to_str()) + .expect("probe library name") + .to_owned() +} + +/// A private, canonical directory: TinyBus refuses module roots that others +/// can write to or that sit beneath a symlinked ancestor. +fn private_root() -> tempfile::TempDir { + let root = tempfile::tempdir_in(std::env::temp_dir().canonicalize().unwrap()).unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(root.path(), std::fs::Permissions::from_mode(0o700)).unwrap(); + } + root +} + +/// The probe library beside a `modules.toml` pinning it, in `directory`. +/// Returns the library's path and SHA-256. +fn stage_library(directory: &Path) -> (PathBuf, String) { + std::fs::create_dir_all(directory).unwrap(); + let library = directory.join(library_name()); + std::fs::copy(probe_library(), &library).unwrap(); + let digest = tinybus::module::sha256_file(&library).unwrap(); + std::fs::write( + directory.join("modules.toml"), + format!("{:?} = {digest:?}\n", library_name()), + ) + .unwrap(); + (library, digest) +} + +/// A verified release under `root`: the archive TinyBus checked and the +/// library it extracted. Returns the archive's and the library's SHA-256. +fn stage_release(root: &Path) -> (String, String) { + let directory = artifact_dir(root, ID, VERSION, &host_candidates()[0]).unwrap(); + let (library, library_digest) = stage_library(&directory); + let archive = directory.join(ARCHIVE); + let mut builder = tar::Builder::new(flate2::write::GzEncoder::new( + std::fs::File::create(&archive).unwrap(), + flate2::Compression::default(), + )); + builder + .append_path_with_name(&library, library_name()) + .unwrap(); + builder + .append_path_with_name(directory.join("modules.toml"), "modules.toml") + .unwrap(); + builder + .into_inner() + .unwrap() + .finish() + .unwrap() + .flush() + .unwrap(); + ( + tinybus::module::sha256_file(&archive).unwrap(), + library_digest, + ) +} + +/// The registry record pinning `archive_sha256` and `library_sha256`. +fn record(archive_sha256: &str, library_sha256: &str) -> &'static ReleaseRecord { + let host_key: &'static str = Box::leak(host_candidates()[0].clone().into_boxed_str()); + let sha256: &'static str = Box::leak(archive_sha256.to_owned().into_boxed_str()); + let library_sha256: &'static str = Box::leak(library_sha256.to_owned().into_boxed_str()); + let assets: &'static [ReleaseAsset<'static>] = Box::leak(Box::new([ReleaseAsset { + host_key, + archive: ARCHIVE, + sha256, + library_sha256: Some(library_sha256), + }])); + Box::leak(Box::new(ReleaseRecord { + id: ID, + bus_name: BUS_NAME, + object_path: OBJECT_PATH, + version: VERSION, + release_url: "https://github.com/tinyhumansai/openhuman/releases/tag/v0.0.1", + assets, + })) +} + +/// A resolver over `record` alone, offline, shaped like [`super::resolver`]'s. +fn resolver( + record: &'static ReleaseRecord, + install_dir: &Path, + bundled_dir: Option<&Path>, + sources: HashMap, +) -> Resolver { + Resolver::new(ResolverInputs { + enabled: true, + install_dir: Some(install_dir.to_path_buf()), + allow_download: false, + bundled_dir: bundled_dir.map(Path::to_path_buf), + sources, + linked: Arc::new(LinkedRegistry::new()), + search_paths: None, + search_skip: Vec::new(), + gate: None, + on_resolved: None, + records: vec![record], + terminal_marker: tinytools::status::MODULE_FAULT_MARKER.to_string(), + runtime: Arc::new(|| Box::pin(runtime::runtime())), + module_config: Arc::new(|_| serde_json::Value::Null), + async_module_config: None, + reporter: report::reporter(), + }) +} + +/// Whether the probe is admitted on the module bus and retains the storage +/// callback binding its load prepared. +async fn serving_with_callbacks(resolver: &Resolver) -> bool { + let Some(prepared) = super::STORAGE_CALLBACK_LOADS + .lock() + .await + .get(ID) + .map(|(binding, _served)| Arc::clone(binding)) + else { + return false; + }; + let host = runtime::runtime().await.unwrap().host(); + if !host + .list() + .iter() + .any(|info| info.manifest.bus_name.as_str() == BUS_NAME) + { + return false; + } + // A loaded module answers with its retained binding and prepares nothing. + resolver + .ensure_loaded_with_host_callback_factory(ID, |_attempt| async { + Err::<(Arc, serde_json::Value), _>( + "nothing to prepare for a loaded module".to_owned(), + ) + }) + .await + .is_ok_and(|retained| retained.unique_name() == prepared.unique_name()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_callback_module_in_the_installer_bundle_loads_with_callbacks() { + if !isolated("a_callback_module_in_the_installer_bundle_loads_with_callbacks") { + return; + } + let bundled = private_root(); + let cache = private_root(); + let (archive, library) = stage_release(bundled.path()); + let resolver = resolver( + record(&archive, &library), + cache.path(), + Some(bundled.path()), + HashMap::new(), + ); + + ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .expect("a bundled release loads with its storage callback"); + assert!(serving_with_callbacks(&resolver).await); + // A later load confirms the retained binding instead of re-presenting it. + ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .expect("a later load reuses the admitted binding"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_callback_module_in_the_release_cache_loads_with_callbacks() { + if !isolated("a_callback_module_in_the_release_cache_loads_with_callbacks") { + return; + } + let cache = private_root(); + let (archive, library) = stage_release(cache.path()); + let resolver = resolver( + record(&archive, &library), + cache.path(), + None, + HashMap::new(), + ); + + ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .expect("a cached release loads with its storage callback"); + assert!(serving_with_callbacks(&resolver).await); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_release_served_callback_module_is_confirmed_by_its_admitted_binding() { + if !isolated("a_release_served_callback_module_is_confirmed_by_its_admitted_binding") { + return; + } + // Every later TinySecurity and TinyComputer load re-checks the binding its + // first load admitted. On a release-served module that check must answer + // from the retained binding, never ask for a local-file source. + let cache = private_root(); + let (archive, library) = stage_release(cache.path()); + let resolver = resolver( + record(&archive, &library), + cache.path(), + None, + HashMap::new(), + ); + ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .expect("a cached release loads with its storage callback"); + let admitted = super::STORAGE_CALLBACK_LOADS + .lock() + .await + .get(ID) + .map(|(binding, _served)| Arc::clone(binding)) + .expect("the admitted binding is kept"); + + super::callback_load::confirm(&resolver, ID, &admitted) + .await + .expect("the admitted binding is confirmed"); + let runtime = runtime::runtime().await.unwrap(); + let (other, _served) = crate::modules::storage_host::prepare_module_callbacks(runtime) + .await + .unwrap(); + let error = super::callback_load::confirm(&resolver, ID, &other) + .await + .unwrap_err(); + assert!(error.contains("different callback binding"), "{error}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_callback_load_whose_caller_gave_up_is_found_by_the_next_caller() { + if !isolated("a_callback_load_whose_caller_gave_up_is_found_by_the_next_caller") { + return; + } + // `ensure_loaded_within` bounds only the wait. A caller that gives up + // must not strand the admitted binding's host state: TinyBus keeps + // resolving, and the next caller has to find the module serving. + let cache = private_root(); + let (archive, library) = stage_release(cache.path()); + let resolver = resolver( + record(&archive, &library), + cache.path(), + None, + HashMap::new(), + ); + // The bus is not started yet: the caller gives up before its host + // callbacks are installed, which must not cancel the load either. + + let gave_up = tokio::time::timeout( + std::time::Duration::ZERO, + ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID), + ) + .await; + assert!(gave_up.is_err(), "the first caller stops waiting"); + + ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .expect("the next caller finds the load the first one started"); + assert!(serving_with_callbacks(&resolver).await); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_bundled_archive_that_does_not_match_its_pin_is_refused() { + if !isolated("a_bundled_archive_that_does_not_match_its_pin_is_refused") { + return; + } + let bundled = private_root(); + let cache = private_root(); + let (_archive, library) = stage_release(bundled.path()); + let resolver = resolver( + record(WRONG_DIGEST, &library), + cache.path(), + Some(bundled.path()), + HashMap::new(), + ); + + let error = ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .unwrap_err(); + // A bundled archive that fails its pin is a broken installation: refused, + // never replaced by the release cache or a download. + assert!(error.contains("installer bundle"), "{error}"); + assert!(error.contains("not cached"), "{error}"); + assert!(!serving_with_callbacks(&resolver).await); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_cached_library_that_does_not_match_its_pin_is_refused() { + if !isolated("a_cached_library_that_does_not_match_its_pin_is_refused") { + return; + } + let cache = private_root(); + let (archive, _library) = stage_release(cache.path()); + let resolver = resolver( + record(&archive, WRONG_DIGEST), + cache.path(), + None, + HashMap::new(), + ); + + let error = ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .unwrap_err(); + assert!( + error.contains("extracted library hash does not match its release checksum"), + "{error}" + ); + assert!(!serving_with_callbacks(&resolver).await); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_local_file_override_wins_over_the_release() { + if !isolated("a_local_file_override_wins_over_the_release") { + return; + } + // The cached release does not match its pin, so a load can only succeed + // from the override. `ops::resolver` turns a `path` override, a source + // build and an `OPENHUMAN_MODULE_PATH` artifact into this source. + let cache = private_root(); + let (_archive, library) = stage_release(cache.path()); + let local = private_root(); + let (path, _) = stage_library(local.path()); + let sources = HashMap::from([( + ID.to_owned(), + ResolverSource::LocalFile { + path: path.to_string_lossy().into_owned(), + }, + )]); + let resolver = resolver(record(WRONG_DIGEST, &library), cache.path(), None, sources); + + ensure_loaded_with_storage_callback(&resolver, &Config::default(), ID) + .await + .expect("the override loads with its storage callback"); + assert!(serving_with_callbacks(&resolver).await); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_detached_callback_load_runs_under_the_callers_context() { + use crate::core::runtime::{ContextOverlay, CoreContext, DomainSet}; + + // TinySecurity's first load prepares its secret scope inside the detached + // work, and that reads the ambient context (profile, agent, storage + // scope). The task on the module runtime must act for its caller, not + // for the process default. + let caller = CoreContext::for_test(DomainSet::full(), None).derive_with( + ContextOverlay::new(Config::default(), DomainSet::full(), Default::default()) + .session_agent("callback-load-caller"), + ); + let cache = private_root(); + let resolver = resolver( + record(WRONG_DIGEST, WRONG_DIGEST), + cache.path(), + None, + HashMap::new(), + ); + + let seen = CoreContext::scope( + caller, + super::callback_load::detached(&resolver, async { + Ok(CoreContext::current().and_then(|ctx| ctx.session_agent().map(str::to_owned))) + }), + ) + .await + .expect("the detached work runs"); + assert_eq!(seen.as_deref(), Some("callback-load-caller")); +} diff --git a/crates/openhuman-core/src/modules/runtime.rs b/crates/openhuman-core/src/modules/runtime.rs index 0118e9b1d5d..a8cf153261a 100644 --- a/crates/openhuman-core/src/modules/runtime.rs +++ b/crates/openhuman-core/src/modules/runtime.rs @@ -80,6 +80,17 @@ where Ok(runtime_now()?.spawn(future)) } +/// Run `future` on the module bus's own executor, starting the bus if needed +/// but without awaiting its host callbacks. The task outlives whichever caller +/// spawned it; work that needs the callbacks awaits [`runtime`] inside it. +pub(crate) fn spawn_on_bus(future: F) -> tinybus::Result> +where + F: std::future::Future + Send + 'static, + F::Output: Send + 'static, +{ + Ok(runtime_now()?.spawn(future)) +} + fn runtime_now() -> tinybus::Result<&'static ModuleRuntime> { let started = RUNTIME.get_or_init(|| { let names = RuntimeNames { diff --git a/crates/openhuman-core/src/modules/security/host_callbacks.rs b/crates/openhuman-core/src/modules/security/host_callbacks.rs index 27538a73f85..3b7c234b6ae 100644 --- a/crates/openhuman-core/src/modules/security/host_callbacks.rs +++ b/crates/openhuman-core/src/modules/security/host_callbacks.rs @@ -193,14 +193,11 @@ pub(in crate::modules) async fn ensure_loaded( if state.key_path != key_path { return Err("security secret scope changed after module admission".into()); } - let module_config = state.module_config(config)?; - resolver - .ensure_loaded_with_host_callbacks( - id, - std::sync::Arc::clone(&state.binding), - module_config, - ) - .await?; + // Confirm the admitted module still retains this binding. Re-presenting + // it would be refused (a binding is spent once a module takes it), and + // the single-binding path accepts only a local-file source, so it + // failed every later call on a host served from the release. + super::super::ops::callback_load::confirm(resolver, id, &state.binding).await?; register_scope(config, state).await?; return Ok(()); } diff --git a/crates/openhuman-core/src/modules/storage_host.rs b/crates/openhuman-core/src/modules/storage_host.rs index 5883da4ccbf..ef28dee4885 100644 --- a/crates/openhuman-core/src/modules/storage_host.rs +++ b/crates/openhuman-core/src/modules/storage_host.rs @@ -740,7 +740,7 @@ mod endpoint; #[cfg(test)] #[path = "storage_host_tests.rs"] -mod tests; +pub(in crate::modules) mod tests; #[cfg(test)] #[path = "storage_host_ctx_tests.rs"] diff --git a/crates/openhuman-core/src/modules/storage_host_cdylib_tests.rs b/crates/openhuman-core/src/modules/storage_host_cdylib_tests.rs index 8a7c7bc2f7d..4b54623ac41 100644 --- a/crates/openhuman-core/src/modules/storage_host_cdylib_tests.rs +++ b/crates/openhuman-core/src/modules/storage_host_cdylib_tests.rs @@ -55,7 +55,7 @@ fn repo_root() -> PathBuf { /// The built probe library: `OPENHUMAN_STORAGE_PROBE_MODULE` when set, else a /// build of the fixture under the checkout's own `target/storage-probe`. -fn probe_artifact() -> &'static Path { +pub(in crate::modules) fn probe_artifact() -> &'static Path { static BUILT: OnceLock = OnceLock::new(); BUILT.get_or_init(|| { if let Some(path) = std::env::var_os("OPENHUMAN_STORAGE_PROBE_MODULE") { diff --git a/crates/openhuman-core/src/modules/storage_host_tests.rs b/crates/openhuman-core/src/modules/storage_host_tests.rs index 0bfda4afe0c..593e1fd0281 100644 --- a/crates/openhuman-core/src/modules/storage_host_tests.rs +++ b/crates/openhuman-core/src/modules/storage_host_tests.rs @@ -699,7 +699,7 @@ mod bus; mod queues; #[path = "storage_host_cdylib_tests.rs"] -mod cdylib; +pub(in crate::modules) mod cdylib; #[cfg(test)] #[path = "storage_host_stream_tests.rs"] mod streams;