Skip to content
Merged
Show file tree
Hide file tree
Changes from 20 commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
12fa49b
chore(vendor): bump tiny* submodules to latest commits
senamakel Nov 3, 2026
da5dc43
chore(vendor): update tinyconnectors submodule
senamakel Nov 3, 2026
9020222
chore(vendor): bump tinydocs submodule to 14f8240
senamakel Nov 3, 2026
cc52796
chore(vendor): bump tinyflows submodule
senamakel Nov 3, 2026
5edaecf
chore(vendor): bump tinyhosts submodule
senamakel Nov 3, 2026
7c1548d
chore(vendor): bump tinyhumans-sdk and tinyjuice submodules
senamakel Nov 3, 2026
5b628db
test(modules): add release tests for ops callback
senamakel Nov 3, 2026
e9511e9
test(modules): wire up callback release tests and share storage probe…
senamakel Nov 3, 2026
9e596d9
chore(vendor): bump tinymcp and tinymemory submodules
senamakel Nov 3, 2026
5b5ad69
chore(vendor): bump tiny* submodules to latest commits
senamakel Nov 3, 2026
e9e1801
test(ops): verify retained callback binding on module reload
senamakel Nov 3, 2026
62ee083
test(ops_callback_release): align probe object path with sdk derivation
senamakel Nov 3, 2026
8e65efe
refactor(modules): extract shared callback load path into callback_load
senamakel Nov 3, 2026
ac8a980
fix(modules): confirm admitted host callback binding instead of re-pr…
senamakel Nov 3, 2026
1288f97
test(modules): cover callback binding confirmation for release-served…
senamakel Nov 3, 2026
f91de13
docs(modules): document callback module load path in ops table
senamakel Nov 3, 2026
33a360c
test(modules): cover callback load surviving a caller timeout
senamakel Nov 3, 2026
01f4cf4
fix(modules): keep callback loads running when the caller stops waiting
senamakel Nov 3, 2026
bdc0dde
test(ops): start runtime before simulating an abandoned callback load
senamakel Nov 3, 2026
de764d2
docs(modules): document callback module load paths in the README
senamakel Nov 3, 2026
4a4daeb
test(modules): cover detached callback load inheriting caller context
senamakel Nov 4, 2026
fa76601
fix(modules): keep callback loads alive when the caller gives up early
senamakel Nov 4, 2026
dab1823
refactor(modules): move callback load helpers into callback_load module
senamakel Nov 4, 2026
be853a3
chore(core): reorder imports and reformat callback binding
senamakel Nov 4, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/openhuman-core/src/modules/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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. |

Expand Down
105 changes: 76 additions & 29 deletions crates/openhuman-core/src/modules/ops.rs
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ use tinybus::module::{
};
use tinybus::Proxy;

pub(super) mod callback_load;
mod list;
pub use list::list;

Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -251,10 +255,24 @@ type StorageCallbackLoads = tokio::sync::Mutex<
static STORAGE_CALLBACK_LOADS: std::sync::LazyLock<StorageCallbackLoads> =
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 (task_resolver, config, id) = (resolver.clone(), config.clone(), id.to_owned());
callback_load::detached(resolver, async move {
load_with_storage_callback(&task_resolver, &config, &id).await
})
.await
}

async fn load_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) {
Expand All @@ -265,25 +283,28 @@ async fn ensure_loaded_with_storage_callback(
// 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;
return callback_load::confirm(resolver, id, binding).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?;
let config = config.clone();
let module_id = id.to_owned();
let (binding, served) = callback_load::load(resolver, id, move || {
let config = config.clone();
let id = module_id.clone();
async move {
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());
Ok((binding, module_config, served))
}
})
.await?;
loads.insert(id.to_owned(), (binding, served));
Ok(())
}
Expand Down Expand Up @@ -342,34 +363,56 @@ 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")]
async fn ensure_loaded_with_computer_callback(
resolver: &Resolver,
config: &Config,
id: &str,
) -> Result<(), String> {
let (task_resolver, config, id) = (resolver.clone(), config.clone(), id.to_owned());
callback_load::detached(resolver, async move {
load_with_computer_callback(&task_resolver, &config, &id).await
})
.await
}

#[cfg(feature = "security-module")]
async fn load_with_computer_callback(
resolver: &Resolver,
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;
return callback_load::confirm(resolver, id, binding).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?;
let config = config.clone();
let module_id = id.to_owned();
let (binding, ()) = callback_load::load(resolver, id, move || {
let config = config.clone();
let id = module_id.clone();
async move {
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;
Ok((binding, module_config, ()))
}
})
.await?;
loads.insert(id.to_owned(), binding);
Ok(())
}
Expand Down Expand Up @@ -712,3 +755,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;
143 changes: 143 additions & 0 deletions crates/openhuman-core/src/modules/ops/callback_load.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
//! 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 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.
///
/// # Errors
///
/// `work`'s error, the module bus failing to start, or the task panicking.
pub(in crate::modules) async fn detached<T, Fut>(
resolver: &Resolver,
work: Fut,
) -> Result<T, String>
where
T: Send + 'static,
Fut: Future<Output = Result<T, String>> + Send + 'static,
{
let runtime = resolver
.runtime()
.await
.map_err(|error| error.to_string())?;
Comment thread
senamakel marked this conversation as resolved.
Resolved
runtime
.spawn(work)
Comment thread
senamakel marked this conversation as resolved.
Resolved
.await
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
.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<S> = (Arc<HostCallbackBinding>, 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<S, F, Fut>(
resolver: &Resolver,
id: &str,
prepare: F,
) -> Result<(Arc<HostCallbackBinding>, S), String>
where
S: Send + 'static,
F: Fn() -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Prepared<S>, String>> + Send + 'static,
{
let candidates: Arc<Mutex<HashMap<String, S>>> = 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")
})?;
Comment thread
senamakel marked this conversation as resolved.
Resolved
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(())
}
Loading
Loading