From 7d88d31462c50a2d8ca8bbd28c447ad270d1e4f6 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 9 Aug 2026 11:23:41 +0300 Subject: [PATCH 1/2] fix(core): repair main after the tinybus and memory-subsystem PRs collided MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #5459 deleted `src/core/event_bus/` and moved the bus onto the vendored tinybus crate; #5446 added new references to `crate::core::event_bus`. Both merged, and neither Cargo world has compiled since. No feature selection avoids it — every failing site is ungated code. Four call sites, mechanically retargeted at the post-tinybus paths: - `memory/guard/audit.rs`, `memory/binding.rs` — `publish_global(..)` is `BUS.publish(..)`, and `event_bus::DomainEvent` is `core::events::DomainEvent`. - `app/src-tauri/src/whatsapp_data/mod.rs` — the free function `register_native_global` is `BUS.native().register`, one-for-one. The shell is a separate Cargo world, so this one is invisible to `cargo check` at the root and surfaces only in the Tauri clippy lane (or the pre-push hook). - `memory/guard/provider_tests.rs` — the two step-7 tests observed the old synchronous `init_global(..).raw_receiver()` broadcast channel, which has no tinybus equivalent. They now subscribe a recording `EventHandler` to the global bus (the idiom `web_chat::event_bus`'s own tests use) and poll, because delivery routes through a broker and is asynchronous now. The refusal test polls with a 2s bound so a regression fails rather than hangs; the success-path test sleeps briefly before asserting absence, since asserting "nothing arrived" the instant after the call would pass even if something had. Two more artefacts of the same collision: - `app/src-tauri/Cargo.lock` never learned about `tinybus`/`tinybus-macros`. - `cargo fmt --all` and prettier both had work to do: #5459 landed ~72 Rust files unformatted and one `.tsx` reflow went with it, so the fmt lane is red on main too. That is the bulk of the diff and it is entirely mechanical. NOT fixed here, because it is a behavioural question rather than a mechanical one: `memory::guard::families::tests::guard_explicit_scope_argument_wins_over_the_ambient_one` fails deterministically on main, in isolation, in a file this PR does not touch — an explicit `SourceScope` argument is widened back to the ambient scope. Filed separately. Co-authored-by: Medulla --- app/src-tauri/Cargo.lock | 23 ++++ app/src-tauri/src/whatsapp_data/mod.rs | 14 +- .../components/settings/panels/VoicePanel.tsx | 10 +- src/core/bus.rs | 18 +-- src/core/jsonrpc.rs | 56 ++++---- src/core/mod.rs | 4 +- src/openhuman/agent/artifacts/store_tests.rs | 4 +- src/openhuman/agent/bus.rs | 28 ++-- .../agent/harness/session/runtime_tests.rs | 17 ++- .../agent/harness/session/turn/tools.rs | 4 +- .../agent/learning/extract/signature.rs | 4 +- .../agent/learning/profile_md_renderer.rs | 4 +- src/openhuman/agent/learning/startup.rs | 4 +- .../orchestration/run_ledger_finalize.rs | 2 +- .../run_ledger_finalize_tests.rs | 2 +- src/openhuman/agent/tinyagents/tools.rs | 24 ++-- src/openhuman/agent/triage/escalation.rs | 2 +- src/openhuman/agent/triage/evaluator.rs | 11 +- src/openhuman/agent/triage/events.rs | 19 +-- src/openhuman/channels/bus.rs | 2 +- src/openhuman/channels/host/adapters.rs | 2 +- src/openhuman/channels/proactive.rs | 8 +- .../providers/telegram/approval_surface.rs | 2 +- .../telegram/approval_surface_tests.rs | 2 +- .../channels/providers/telegram/bus.rs | 2 +- .../channels/providers/telegram/bus_tests.rs | 2 +- src/openhuman/channels/routes_tests.rs | 2 +- .../channels/runtime/dispatch/processor.rs | 40 +++--- .../channels/runtime/test_support.rs | 124 +++++++++--------- src/openhuman/channels/tests/health.rs | 9 +- .../channels/tests/runtime_dispatch.rs | 3 +- src/openhuman/config/ops/agent.rs | 8 +- src/openhuman/cron/bus.rs | 2 +- src/openhuman/cron/scheduler_tests.rs | 6 +- src/openhuman/desktop/notifications/bus.rs | 8 +- src/openhuman/flows/bus.rs | 2 +- src/openhuman/flows/ops.rs | 24 ++-- src/openhuman/flows/ops_tests.rs | 8 +- .../inference/provider/factory_tests.rs | 12 +- .../provider/openhuman_backend_model.rs | 18 ++- .../inference/provider/ops/http_error.rs | 16 +-- src/openhuman/inference/provider/ops_tests.rs | 7 +- .../integrations/composio/ops/direct_mode.rs | 20 ++- .../integrations/task_sources/bus.rs | 2 +- src/openhuman/meet/backend_bot/calendar.rs | 4 +- src/openhuman/memory/binding.rs | 4 +- src/openhuman/memory/conversations/bus.rs | 6 +- src/openhuman/memory/diff/ops.rs | 26 ++-- src/openhuman/memory/global.rs | 5 +- src/openhuman/memory/guard/audit.rs | 7 +- src/openhuman/memory/guard/provider_tests.rs | 113 ++++++++++++---- src/openhuman/memory/ops/sync.rs | 23 ++-- src/openhuman/memory/store/client.rs | 30 ++--- src/openhuman/memory/sync/composio/bus.rs | 12 +- src/openhuman/memory/sync_events.rs | 30 +++-- .../memory/sync_pipeline_e2e_tests.rs | 4 +- src/openhuman/memory/tinycortex/sync.rs | 18 ++- src/openhuman/memory/tree/tree_runtime/bus.rs | 2 +- src/openhuman/security/approval/gate.rs | 5 +- src/openhuman/security/credentials/bus.rs | 2 +- .../security/credentials/session_support.rs | 13 +- src/openhuman/security/devices/bus.rs | 4 +- src/openhuman/security/egress/emit_tests.rs | 8 +- .../security/keyring_consent/policy.rs | 19 +-- src/openhuman/skills/bus.rs | 4 +- src/openhuman/skills/ops_create.rs | 4 +- src/openhuman/skills/webhooks/bus.rs | 2 +- src/openhuman/voice/bus.rs | 4 +- src/openhuman/web_chat/event_bus.rs | 2 +- tests/agent_harness_e2e.rs | 4 +- tests/calendar_grounding_e2e.rs | 2 +- ...io_list_tools_stack_overflow_regression.rs | 2 +- .../config_auth_app_state_connectivity_e2e.rs | 2 +- tests/json_rpc_e2e.rs | 39 +++--- tests/monitor_agent_e2e.rs | 2 +- tests/subconscious_conversation_e2e.rs | 33 ++--- tests/subconscious_fullstack_e2e.rs | 2 +- tests/subconscious_triggers_e2e.rs | 4 +- 78 files changed, 536 insertions(+), 486 deletions(-) diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 1eef37c81e..21a1a1073b 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -5568,6 +5568,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tinyagents", + "tinybus", "tinychannels", "tinycortex", "tinycortex-api", @@ -8904,6 +8905,28 @@ dependencies = [ "tracing", ] +[[package]] +name = "tinybus" +version = "0.1.0" +dependencies = [ + "async-trait", + "serde", + "serde_json", + "thiserror 2.0.18", + "tinybus-macros", + "tokio", + "tracing", +] + +[[package]] +name = "tinybus-macros" +version = "0.1.0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "tinychannels" version = "0.1.0" diff --git a/app/src-tauri/src/whatsapp_data/mod.rs b/app/src-tauri/src/whatsapp_data/mod.rs index c43fbc23c2..af5d437451 100644 --- a/app/src-tauri/src/whatsapp_data/mod.rs +++ b/app/src-tauri/src/whatsapp_data/mod.rs @@ -60,30 +60,34 @@ pub async fn ensure_store() -> Result, String> { /// (`openhuman_core::openhuman::channels::whatsapp_data::methods`) so the two sides never /// drift on the string key. pub fn register_native_handlers() { - use openhuman_core::core::event_bus::register_native_global; + // Post-#5459 the native registry is tinybus's, reached through the core's + // global bus handle. `NativeRegistry::register` replaces the old free + // function `register_native_global` one-for-one; the method-name constants + // and handler signatures are unchanged. + let native = openhuman_core::core::bus::BUS.native(); - register_native_global::, _, _>( + native.register::, _, _>( methods::LIST_CHATS, |req| async move { let store = ensure_store().await?; ops::list_chats(&store, req).map_err(|e| format!("{e:#}")) }, ); - register_native_global::, _, _>( + native.register::, _, _>( methods::LIST_MESSAGES, |req| async move { let store = ensure_store().await?; ops::list_messages(&store, req).map_err(|e| format!("{e:#}")) }, ); - register_native_global::, _, _>( + native.register::, _, _>( methods::SEARCH_MESSAGES, |req| async move { let store = ensure_store().await?; ops::search_messages(&store, req).map_err(|e| format!("{e:#}")) }, ); - register_native_global::( + native.register::( methods::INGEST, |req| async move { let store = ensure_store().await?; diff --git a/app/src/components/settings/panels/VoicePanel.tsx b/app/src/components/settings/panels/VoicePanel.tsx index 4a77989bd1..4c77c01a60 100644 --- a/app/src/components/settings/panels/VoicePanel.tsx +++ b/app/src/components/settings/panels/VoicePanel.tsx @@ -203,11 +203,11 @@ const VoicePanel = ({ embedded = false }: VoicePanelProps = {}) => { const slugs = new Set(vs.voiceProviders.map(p => p.slug)); const sttStr = vs.sttProvider.kind === 'cloud' - // `cloud` is a routing sentinel: it delegates to the configured - // engine, which voice_status reports after resolving it. Seed the - // selector with that effective engine so Settings does not claim - // the backend proxy is in use when a hosted BYOK engine is. - ? voiceResponse.stt_engine || 'cloud' + ? // `cloud` is a routing sentinel: it delegates to the configured + // engine, which voice_status reports after resolving it. Seed the + // selector with that effective engine so Settings does not claim + // the backend proxy is in use when a hosted BYOK engine is. + voiceResponse.stt_engine || 'cloud' : vs.sttProvider.kind === 'local' ? vs.sttProvider.engine : slugs.has(vs.sttProvider.providerSlug) diff --git a/src/core/bus.rs b/src/core/bus.rs index cd3057eb41..d03e9c47fa 100644 --- a/src/core/bus.rs +++ b/src/core/bus.rs @@ -80,13 +80,9 @@ pub fn manifest() -> PeerManifest { .expect("the events interface constant is valid"); PeerManifest::new("openhuman") .version( - Version::parse(env!("CARGO_PKG_VERSION")) - .unwrap_or_else(|_| Version::new(0, 0, 0)), + Version::parse(env!("CARGO_PKG_VERSION")).unwrap_or_else(|_| Version::new(0, 0, 0)), ) - .provides(InterfaceVersion::provided( - interface, - EVENTS_VERSION, - )) + .provides(InterfaceVersion::provided(interface, EVENTS_VERSION)) .consumes(InterfaceVersion::consumed( EVENTS_INTERFACE .try_into() @@ -183,8 +179,14 @@ mod tests { fn the_manifest_declares_the_catalog_in_both_directions() { let manifest = manifest(); let interface = EVENTS_INTERFACE.try_into().unwrap(); - assert!(manifest.provided(&interface).is_some(), "openhuman publishes"); - assert!(manifest.consumed(&interface).is_some(), "openhuman subscribes"); + assert!( + manifest.provided(&interface).is_some(), + "openhuman publishes" + ); + assert!( + manifest.consumed(&interface).is_some(), + "openhuman subscribes" + ); } #[tokio::test] diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index bdd32e22dc..665775a631 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -298,12 +298,10 @@ pub async fn invoke_method(state: AppState, method: &str, params: Value) -> Resu // `scheduler_gate::set_signed_out(false)`. Duplicating that check // here would pull a domain concern into the transport layer and would // add an extra config-load round-trip on every 401. - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::SessionExpired { - source: format!("jsonrpc.invoke_method:{method}"), - reason: sanitized_reason, - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::SessionExpired { + source: format!("jsonrpc.invoke_method:{method}"), + reason: sanitized_reason, + }); } else if is_unconfirmed_unauthorized_error(msg) { log::info!( "[jsonrpc] unconfirmed unauthorized error for method='{}' (not session expiry) — leaving session intact: {}", @@ -1663,25 +1661,23 @@ async fn domain_events_handler(headers: axum::http::HeaderMap) -> Response { .await .map(|event| (Ok::<_, std::convert::Infallible>(event), rx)) }) - .filter_map( - |item| -> Option> { - let event = match item { - Ok(ev) => ev, - Err(_) => return None, - }; - let domain = event.domain().to_string(); - let event_name = event.variant_name(); - let agent = event.agent_hint().unwrap_or("").to_string(); - let data = json!({ - "domain": domain, - "event": event_name, - "agent": agent, - "timestamp": chrono::Utc::now().format("%H:%M:%S").to_string(), - }); - let data_str = serde_json::to_string(&data).ok()?; - Some(Ok(Event::default().event(domain).data(data_str))) - }, - ); + .filter_map(|item| -> Option> { + let event = match item { + Ok(ev) => ev, + Err(_) => return None, + }; + let domain = event.domain().to_string(); + let event_name = event.variant_name(); + let agent = event.agent_hint().unwrap_or("").to_string(); + let data = json!({ + "domain": domain, + "event": event_name, + "agent": agent, + "timestamp": chrono::Utc::now().format("%H:%M:%S").to_string(), + }); + let data_str = serde_json::to_string(&data).ok()?; + Some(Ok(Event::default().event(domain).data(data_str))) + }); let config_stream = futures::stream::once(async move { Ok::<_, std::convert::Infallible>(config_event) }); @@ -2660,12 +2656,10 @@ pub async fn bootstrap_core_runtime( Prompt-class external-effect tool calls run unprompted", host_kind.tag() ); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::ApprovalGateDisabled { - host: host_kind.tag().to_string(), - reason: "env-override".to_string(), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::ApprovalGateDisabled { + host: host_kind.tag().to_string(), + reason: "env-override".to_string(), + }); } // Artifact surface bridges DomainEvent::ArtifactReady/Failed onto the web // channel ("Files in this chat" panel + ArtifactCard updates). This is diff --git a/src/core/mod.rs b/src/core/mod.rs index f74fa67983..baff34bf02 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -9,12 +9,12 @@ use serde::Serialize; pub mod agent_cli; pub mod all; pub mod auth; +pub mod bus; +pub mod bus_testing; pub mod cli; pub mod cli_capability; pub mod dispatch; pub mod event_bind_tokens; -pub mod bus; -pub mod bus_testing; pub mod events; // Ungated compile-time marker for the `http-server` gate (#5048) — the desktop // shell asserts `HTTP_SERVER_COMPILED_IN` so a listener-less core fails the diff --git a/src/openhuman/agent/artifacts/store_tests.rs b/src/openhuman/agent/artifacts/store_tests.rs index 55b565c1bf..26d4bbb43a 100644 --- a/src/openhuman/agent/artifacts/store_tests.rs +++ b/src/openhuman/agent/artifacts/store_tests.rs @@ -196,10 +196,10 @@ async fn validate_artifact_id_rejects_slashes() { use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; -use tinybus::SubscriptionHandle; use async_trait::async_trait; use std::sync::{Arc, Mutex as StdMutex}; +use tinybus::EventHandler; +use tinybus::SubscriptionHandle; #[derive(Clone)] struct PendingCollector { diff --git a/src/openhuman/agent/bus.rs b/src/openhuman/agent/bus.rs index 4980db30e5..6c2465afcc 100644 --- a/src/openhuman/agent/bus.rs +++ b/src/openhuman/agent/bus.rs @@ -373,9 +373,8 @@ async fn handle_agent_run_turn_on_large_stack( /// allowing any part of the system to request an agentic turn without /// depending directly on the agent harness. pub fn register_agent_handlers() { - BUS.native().register::( - AGENT_RUN_TURN_METHOD, - |req| { + BUS.native() + .register::(AGENT_RUN_TURN_METHOD, |req| { #[cfg(test)] { handle_agent_run_turn_on_large_stack(req) @@ -384,8 +383,7 @@ pub fn register_agent_handlers() { { handle_agent_run_turn(req) } - }, - ); + }); tracing::debug!("[agent::bus] registered native handler `{AGENT_RUN_TURN_METHOD}`"); } @@ -434,20 +432,16 @@ pub fn register_agent_handlers() { /// } /// ``` #[cfg(test)] -pub async fn mock_agent_run_turn( - handler: F, -) -> crate::core::bus_testing::MockBusGuard +pub async fn mock_agent_run_turn(handler: F) -> crate::core::bus_testing::MockBusGuard where F: Fn(AgentTurnRequest) -> Fut + Send + Sync + 'static, Fut: std::future::Future> + Send + 'static, { - crate::core::bus_testing::mock_bus_stub::< - AgentTurnRequest, - AgentTurnResponse, - F, - Fut, - _, - >(AGENT_RUN_TURN_METHOD, handler, || register_agent_handlers()) + crate::core::bus_testing::mock_bus_stub::( + AGENT_RUN_TURN_METHOD, + handler, + || register_agent_handlers(), + ) .await } @@ -461,9 +455,7 @@ where /// handler with a stub, use [`mock_agent_run_turn`] instead. #[cfg(test)] pub async fn use_real_agent_handler() -> tokio::sync::MutexGuard<'static, ()> { - let guard = crate::core::bus_testing::BUS_HANDLER_LOCK - .lock() - .await; + let guard = crate::core::bus_testing::BUS_HANDLER_LOCK.lock().await; register_agent_handlers(); guard } diff --git a/src/openhuman/agent/harness/session/runtime_tests.rs b/src/openhuman/agent/harness/session/runtime_tests.rs index d3dd0c0ae4..f0773579c8 100644 --- a/src/openhuman/agent/harness/session/runtime_tests.rs +++ b/src/openhuman/agent/harness/session/runtime_tests.rs @@ -258,13 +258,16 @@ async fn run_single_publishes_completed_and_error_events() { crate::core::bus::init().await.expect("bus init"); let events = Arc::new(AsyncMutex::new(Vec::::new())); let events_handler = Arc::clone(&events); - let _handle = crate::core::bus::BUS.get().unwrap().on("runtime-events-test", move |event| { - let events = Arc::clone(&events_handler); - let cloned = event.clone(); - Box::pin(async move { - events.lock().await.push(cloned); - }) - }); + let _handle = crate::core::bus::BUS + .get() + .unwrap() + .on("runtime-events-test", move |event| { + let events = Arc::clone(&events_handler); + let cloned = event.clone(); + Box::pin(async move { + events.lock().await.push(cloned); + }) + }); let ok_provider: Arc> = Arc::new(StaticModel { response: Mutex::new(Some(Ok(ChatResponse { diff --git a/src/openhuman/agent/harness/session/turn/tools.rs b/src/openhuman/agent/harness/session/turn/tools.rs index fd325370a7..ef219726f9 100644 --- a/src/openhuman/agent/harness/session/turn/tools.rs +++ b/src/openhuman/agent/harness/session/turn/tools.rs @@ -165,9 +165,7 @@ impl Agent { let mut closed = false; loop { match rx.try_recv() { - Ok(crate::core::events::DomainEvent::ComposioIntegrationsChanged { - toolkits, - }) => { + Ok(crate::core::events::DomainEvent::ComposioIntegrationsChanged { toolkits }) => { saw_signal = true; log::info!( "[agent_loop] received composio integrations changed event (active_toolkits={:?})", diff --git a/src/openhuman/agent/learning/extract/signature.rs b/src/openhuman/agent/learning/extract/signature.rs index 73d9831248..2277386457 100644 --- a/src/openhuman/agent/learning/extract/signature.rs +++ b/src/openhuman/agent/learning/extract/signature.rs @@ -24,11 +24,11 @@ use async_trait::async_trait; use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; -use tinybus::SubscriptionHandle; use crate::openhuman::agent::learning::candidate::{ self, Buffer, CueFamily, EvidenceRef, FacetClass, LearningCandidate, }; +use tinybus::EventHandler; +use tinybus::SubscriptionHandle; // ── Constants ──────────────────────────────────────────────────────────────── diff --git a/src/openhuman/agent/learning/profile_md_renderer.rs b/src/openhuman/agent/learning/profile_md_renderer.rs index 57cf2699a4..a6cb02f796 100644 --- a/src/openhuman/agent/learning/profile_md_renderer.rs +++ b/src/openhuman/agent/learning/profile_md_renderer.rs @@ -41,11 +41,11 @@ use async_trait::async_trait; use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; -use tinybus::SubscriptionHandle; use crate::openhuman::agent::learning::cache::FacetCache; use crate::openhuman::integrations::composio::providers::profile_md::replace_managed_block; use crate::openhuman::memory::store::profile::UserState; +use tinybus::EventHandler; +use tinybus::SubscriptionHandle; // ── Class → block metadata ──────────────────────────────────────────────────── diff --git a/src/openhuman/agent/learning/startup.rs b/src/openhuman/agent/learning/startup.rs index 4c76bfcdbc..3b22186451 100644 --- a/src/openhuman/agent/learning/startup.rs +++ b/src/openhuman/agent/learning/startup.rs @@ -28,9 +28,9 @@ use std::path::Path; use std::sync::OnceLock; -use tinybus::SubscriptionHandle; use crate::openhuman::memory::global::client_if_ready; use crate::openhuman::memory::store::MemoryClientRef; +use tinybus::SubscriptionHandle; static EMAIL_SIG_HANDLE: OnceLock> = OnceLock::new(); @@ -167,7 +167,6 @@ fn register_with_client( mod tests { use super::*; use crate::core::events::DomainEvent; -use tinybus::EventBus; use crate::openhuman::agent::learning::candidate::Buffer; use crate::openhuman::agent::learning::extract::signature::{ parse_signature, register_email_signature_subscriber_on, @@ -176,6 +175,7 @@ use tinybus::EventBus; use std::sync::Arc; use std::time::Duration; use tempfile::TempDir; + use tinybus::EventBus; /// Build a real `MemoryClient` against a fresh temp workspace. The temp dir /// is returned so callers keep it alive for the client's lifetime. diff --git a/src/openhuman/agent/orchestration/run_ledger_finalize.rs b/src/openhuman/agent/orchestration/run_ledger_finalize.rs index 334496c22c..2a85be89cc 100644 --- a/src/openhuman/agent/orchestration/run_ledger_finalize.rs +++ b/src/openhuman/agent/orchestration/run_ledger_finalize.rs @@ -31,9 +31,9 @@ use async_trait::async_trait; use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::config::Config; use tinyagents::session::run_ledger::{transition_agent_run_status, AgentRunStatus}; +use tinybus::EventHandler; const LOG_PREFIX: &str = "[run_ledger][finalize]"; diff --git a/src/openhuman/agent/orchestration/run_ledger_finalize_tests.rs b/src/openhuman/agent/orchestration/run_ledger_finalize_tests.rs index 16eec16dba..fe961da5a2 100644 --- a/src/openhuman/agent/orchestration/run_ledger_finalize_tests.rs +++ b/src/openhuman/agent/orchestration/run_ledger_finalize_tests.rs @@ -5,10 +5,10 @@ use super::*; use serde_json::json; use tempfile::TempDir; -use tinybus::EventHandler; use tinyagents::session::run_ledger::{ get_agent_run, upsert_agent_run, AgentRunKind, AgentRunStatus, AgentRunUpsert, }; +use tinybus::EventHandler; fn test_config(dir: &TempDir) -> Config { let mut config = Config::default(); diff --git a/src/openhuman/agent/tinyagents/tools.rs b/src/openhuman/agent/tinyagents/tools.rs index bae233ad29..cb795d701f 100644 --- a/src/openhuman/agent/tinyagents/tools.rs +++ b/src/openhuman/agent/tinyagents/tools.rs @@ -210,12 +210,10 @@ pub(crate) async fn execute_openhuman_tool( // the `TaToolResult` below. let started = std::time::Instant::now(); let tool_name = call.name.clone(); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::ToolExecutionStarted { - tool_name: tool_name.clone(), - session_id: TINYAGENTS_TOOL_SESSION.to_string(), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::ToolExecutionStarted { + tool_name: tool_name.clone(), + session_id: TINYAGENTS_TOOL_SESSION.to_string(), + }); // Approval (HITL) now runs in `ApprovalSecurityMiddleware` // (`tinyagents/middleware.rs`, a `wrap_tool` middleware) so a denial @@ -304,14 +302,12 @@ pub(crate) async fn execute_openhuman_tool( // Terminal per-tool telemetry (#4467, item 5): success is derived from the // rendered result's error channel so a tool-reported error surfaces as a // failed completion, mirroring the node-runtime bridge. - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::ToolExecutionCompleted { - tool_name, - session_id: TINYAGENTS_TOOL_SESSION.to_string(), - success: result.error.is_none(), - elapsed_ms, - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::ToolExecutionCompleted { + tool_name, + session_id: TINYAGENTS_TOOL_SESSION.to_string(), + success: result.error.is_none(), + elapsed_ms, + }); result } diff --git a/src/openhuman/agent/triage/escalation.rs b/src/openhuman/agent/triage/escalation.rs index dca3faf97e..ef4234c432 100644 --- a/src/openhuman/agent/triage/escalation.rs +++ b/src/openhuman/agent/triage/escalation.rs @@ -385,7 +385,7 @@ async fn gate_linked_card_terminal(envelope: &TriggerEnvelope, decision: &str) { mod tests { use super::*; use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; use serde_json::json; use tokio::time::{sleep, timeout, Duration}; diff --git a/src/openhuman/agent/triage/evaluator.rs b/src/openhuman/agent/triage/evaluator.rs index 255cd52796..43d9210928 100644 --- a/src/openhuman/agent/triage/evaluator.rs +++ b/src/openhuman/agent/triage/evaluator.rs @@ -35,7 +35,6 @@ use std::time::{Duration, Instant}; use anyhow::{anyhow, Context}; use crate::core::bus::BUS; -use tinybus::NativeRequestError; use crate::openhuman::agent::bus::{AgentTurnRequest, AgentTurnResponse, AGENT_RUN_TURN_METHOD}; use crate::openhuman::agent::harness::definition::{AgentDefinition, PromptSource}; use crate::openhuman::agent::harness::AgentDefinitionRegistry; @@ -46,6 +45,7 @@ use crate::openhuman::cron::scheduler_gate::LlmPermit; use crate::openhuman::inference::provider::error_classify::{ is_rate_limited, is_upstream_unhealthy, parse_retry_after_ms, }; +use tinybus::NativeRequestError; use super::decision::{parse_triage_decision, ParseError, TriageDecision}; use super::envelope::TriggerEnvelope; @@ -508,11 +508,10 @@ async fn try_arm( }, }; - let response = match BUS.native().request::( - AGENT_RUN_TURN_METHOD, - request, - ) - .await + let response = match BUS + .native() + .request::(AGENT_RUN_TURN_METHOD, request) + .await { Ok(r) => r, Err(err) => { diff --git a/src/openhuman/agent/triage/events.rs b/src/openhuman/agent/triage/events.rs index 5de795b0dc..fbfb0cf32f 100644 --- a/src/openhuman/agent/triage/events.rs +++ b/src/openhuman/agent/triage/events.rs @@ -109,7 +109,7 @@ pub fn publish_failed(envelope: &TriggerEnvelope, reason: &str) { mod tests { use super::*; use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; use crate::openhuman::agent::triage::TriggerEnvelope; use serde_json::json; use std::sync::Arc; @@ -121,13 +121,16 @@ use crate::core::events::DomainEvent; crate::core::bus::init().await.expect("bus init"); let seen = Arc::new(Mutex::new(Vec::::new())); let seen_handler = Arc::clone(&seen); - let _handle = crate::core::bus::BUS.get().unwrap().on("triage-events-test", move |event| { - let seen = Arc::clone(&seen_handler); - let cloned = event.clone(); - Box::pin(async move { - seen.lock().await.push(cloned); - }) - }); + let _handle = crate::core::bus::BUS + .get() + .unwrap() + .on("triage-events-test", move |event| { + let seen = Arc::clone(&seen_handler); + let cloned = event.clone(); + Box::pin(async move { + seen.lock().await.push(cloned); + }) + }); let envelope = TriggerEnvelope::from_composio( "gmail", diff --git a/src/openhuman/channels/bus.rs b/src/openhuman/channels/bus.rs index 65d390492a..a3756b9fa3 100644 --- a/src/openhuman/channels/bus.rs +++ b/src/openhuman/channels/bus.rs @@ -5,9 +5,9 @@ //! channel provider and sends the reply back through the REST API. use crate::core::events::DomainEvent; -use tinybus::EventHandler; use async_trait::async_trait; use serde_json::{json, Value}; +use tinybus::EventHandler; /// Subscribes to `ChannelInboundMessage` events and runs the agent loop, /// sending replies back to the originating channel via the backend REST API. diff --git a/src/openhuman/channels/host/adapters.rs b/src/openhuman/channels/host/adapters.rs index 381018ed50..4b5b7f5600 100644 --- a/src/openhuman/channels/host/adapters.rs +++ b/src/openhuman/channels/host/adapters.rs @@ -357,7 +357,7 @@ impl EventSink for OpenHumanEventSink { } "channel" => { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; let event = match kind { "reaction_received" => DomainEvent::ChannelReactionReceived { channel: json_str(&payload, "channel"), diff --git a/src/openhuman/channels/proactive.rs b/src/openhuman/channels/proactive.rs index 62854e1225..b13b95fe44 100644 --- a/src/openhuman/channels/proactive.rs +++ b/src/openhuman/channels/proactive.rs @@ -20,13 +20,13 @@ //! exist. use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::core::socketio::WebChannelEvent; use crate::openhuman::channels::{Channel, ChannelSendExt, SendMessage}; use crate::openhuman::web_chat::publish_web_channel_event; use async_trait::async_trait; use std::collections::HashMap; use std::sync::{Arc, RwLock}; +use tinybus::EventHandler; #[cfg(not(test))] fn proactive_approval_gate() -> Option> { @@ -46,9 +46,9 @@ pub fn register_web_only_proactive_subscriber() { use std::sync::Once; static REGISTERED: Once = Once::new(); REGISTERED.call_once(|| { - if let Some(handle) = crate::core::bus::BUS.subscribe(Arc::new( - ProactiveMessageSubscriber::web_only(), - )) { + if let Some(handle) = + crate::core::bus::BUS.subscribe(Arc::new(ProactiveMessageSubscriber::web_only())) + { std::mem::forget(handle); tracing::debug!("[proactive] web-only subscriber registered"); } else { diff --git a/src/openhuman/channels/providers/telegram/approval_surface.rs b/src/openhuman/channels/providers/telegram/approval_surface.rs index e0539bcb99..053f7cfd1d 100644 --- a/src/openhuman/channels/providers/telegram/approval_surface.rs +++ b/src/openhuman/channels/providers/telegram/approval_surface.rs @@ -36,12 +36,12 @@ //! [`parse_approval_reply`]: crate::openhuman::security::approval::parse_approval_reply use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::channels::traits::{ChannelSendExt, SendMessage}; use crate::openhuman::channels::Channel; use async_trait::async_trait; use std::collections::HashMap; use std::sync::{Arc, Mutex}; +use tinybus::EventHandler; const LOG_PREFIX: &str = "[telegram-approval]"; diff --git a/src/openhuman/channels/providers/telegram/approval_surface_tests.rs b/src/openhuman/channels/providers/telegram/approval_surface_tests.rs index 81e211fcb0..2a2cc9cf4d 100644 --- a/src/openhuman/channels/providers/telegram/approval_surface_tests.rs +++ b/src/openhuman/channels/providers/telegram/approval_surface_tests.rs @@ -1,13 +1,13 @@ //! Tests for the Telegram approval-surface subscriber. use super::*; -use tinybus::EventHandler; use crate::openhuman::channels::traits::{ChannelMessage, SendMessage}; use crate::openhuman::channels::Channel; use async_trait::async_trait; use std::collections::HashMap; use std::path::PathBuf; use std::sync::Mutex as StdMutex; +use tinybus::EventHandler; /// Mock channel that records every outbound `send()` call. Used in place /// of the real `TelegramChannel` so tests can assert what the subscriber diff --git a/src/openhuman/channels/providers/telegram/bus.rs b/src/openhuman/channels/providers/telegram/bus.rs index bde2fc6b45..d2b767fc14 100644 --- a/src/openhuman/channels/providers/telegram/bus.rs +++ b/src/openhuman/channels/providers/telegram/bus.rs @@ -1,10 +1,10 @@ //! Event-bus subscriber for Telegram remote-control lifecycle signals. use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::channels::providers::telegram::session_store::with_store; use async_trait::async_trait; use std::path::PathBuf; +use tinybus::EventHandler; const LOG_PREFIX: &str = "[telegram-remote]"; diff --git a/src/openhuman/channels/providers/telegram/bus_tests.rs b/src/openhuman/channels/providers/telegram/bus_tests.rs index 32b774f6e9..e2802b3923 100644 --- a/src/openhuman/channels/providers/telegram/bus_tests.rs +++ b/src/openhuman/channels/providers/telegram/bus_tests.rs @@ -1,7 +1,7 @@ use super::bus::TelegramRemoteSubscriber; use crate::core::events::DomainEvent; -use tinybus::EventHandler; use tempfile::tempdir; +use tinybus::EventHandler; #[tokio::test] async fn subscriber_marks_busy_on_received_and_clears_on_processed() { diff --git a/src/openhuman/channels/routes_tests.rs b/src/openhuman/channels/routes_tests.rs index a66e322def..288202e980 100644 --- a/src/openhuman/channels/routes_tests.rs +++ b/src/openhuman/channels/routes_tests.rs @@ -1,6 +1,5 @@ use super::*; use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::agent::messages::ChatMessage; use crate::openhuman::channels::context::{ ChannelRuntimeContext, RouteSelectionMap, TurnModelSourceCacheMap, @@ -13,6 +12,7 @@ use async_trait::async_trait; use std::collections::HashMap; use std::path::PathBuf; use std::sync::{Arc, Mutex}; +use tinybus::EventHandler; struct DummyMemory; diff --git a/src/openhuman/channels/runtime/dispatch/processor.rs b/src/openhuman/channels/runtime/dispatch/processor.rs index 84ec82a75e..370b490976 100644 --- a/src/openhuman/channels/runtime/dispatch/processor.rs +++ b/src/openhuman/channels/runtime/dispatch/processor.rs @@ -12,7 +12,6 @@ use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::NativeRequestError; use crate::openhuman::agent::bus::{AgentTurnRequest, AgentTurnResponse, AGENT_RUN_TURN_METHOD}; use crate::openhuman::agent::messages::ChatMessage; use crate::openhuman::agent::progress::AgentProgress; @@ -30,6 +29,7 @@ use crate::openhuman::inference::provider; use crate::openhuman::util::truncate_with_ellipsis; use std::sync::Arc; use std::time::{Duration, Instant}; +use tinybus::NativeRequestError; use tokio_util::sync::CancellationToken; use super::helpers::{ @@ -519,26 +519,24 @@ pub(crate) async fn process_channel_runtime_message( "[channels::dispatch] dispatching {AGENT_RUN_TURN_METHOD} via native bus" ); let agent_call = async { - BUS.native().request::( - AGENT_RUN_TURN_METHOD, - turn_request, - ) - .await - .map_err(|err| match err { - // Unwrap handler-returned errors so the underlying - // message (e.g. "Agent exceeded maximum tool iterations") - // flows through without being wrapped in bus-transport - // layer prose. The error-formatting path downstream - // treats this `anyhow::Error` the same way it did before - // the bus migration. - NativeRequestError::HandlerFailed { message, .. } => { - anyhow::anyhow!(message) - } - // Bus-level errors (UnregisteredHandler / TypeMismatch / - // NotInitialized) surface with their full Display so - // startup wiring bugs are immediately obvious in logs. - other => anyhow::anyhow!("[agent.run_turn dispatch] {other}"), - }) + BUS.native() + .request::(AGENT_RUN_TURN_METHOD, turn_request) + .await + .map_err(|err| match err { + // Unwrap handler-returned errors so the underlying + // message (e.g. "Agent exceeded maximum tool iterations") + // flows through without being wrapped in bus-transport + // layer prose. The error-formatting path downstream + // treats this `anyhow::Error` the same way it did before + // the bus migration. + NativeRequestError::HandlerFailed { message, .. } => { + anyhow::anyhow!(message) + } + // Bus-level errors (UnregisteredHandler / TypeMismatch / + // NotInitialized) surface with their full Display so + // startup wiring bugs are immediately obvious in logs. + other => anyhow::anyhow!("[agent.run_turn dispatch] {other}"), + }) }; // Sub-issue 2 of #3098: scope the agent turn in an `ApprovalChatContext` // for channels that have a registered approval surface — currently diff --git a/src/openhuman/channels/runtime/test_support.rs b/src/openhuman/channels/runtime/test_support.rs index 05279307a5..568179f04a 100644 --- a/src/openhuman/channels/runtime/test_support.rs +++ b/src/openhuman/channels/runtime/test_support.rs @@ -298,9 +298,7 @@ fn memory_entry(input: TestMemoryEntry) -> MemoryEntry { /// `start_channels`) so concurrent registrations cannot race in the same /// process. pub async fn lock_agent_handler() -> tokio::sync::MutexGuard<'static, ()> { - crate::core::bus_testing::BUS_HANDLER_LOCK - .lock() - .await + crate::core::bus_testing::BUS_HANDLER_LOCK.lock().await } pub async fn run_dispatch_harness(options: DispatchHarnessOptions) -> DispatchHarnessObservation { @@ -312,7 +310,10 @@ pub async fn run_dispatch_harness(options: DispatchHarnessOptions) -> DispatchHa let _harness_guard = lock_agent_handler().await; crate::core::bus::init().await.expect("bus init"); - let mut event_rx = crate::core::bus::BUS.get().expect("bus initialised").receiver(); + let mut event_rx = crate::core::bus::BUS + .get() + .expect("bus initialised") + .receiver(); let _ = crate::openhuman::agent::harness::definition::AgentDefinitionRegistry::init_global_builtins( ); @@ -329,70 +330,71 @@ pub async fn run_dispatch_harness(options: DispatchHarnessOptions) -> DispatchHa let handler_error = options.handler_error.clone(); let handler_delay = Duration::from_millis(options.handler_delay_ms); - BUS.native().register::(AGENT_RUN_TURN_METHOD, { - let handler_roles = Arc::clone(&handler_roles); - let handler_text = Arc::clone(&handler_text); - let handler_provider = Arc::clone(&handler_provider); - let handler_channel = Arc::clone(&handler_channel); - let handler_progress = Arc::clone(&handler_progress); - move |req| { + BUS.native() + .register::(AGENT_RUN_TURN_METHOD, { let handler_roles = Arc::clone(&handler_roles); let handler_text = Arc::clone(&handler_text); let handler_provider = Arc::clone(&handler_provider); let handler_channel = Arc::clone(&handler_channel); let handler_progress = Arc::clone(&handler_progress); - let response_text = response_text.clone(); - let handler_error = handler_error.clone(); - async move { - *handler_roles.lock().expect("roles lock") = - req.history.iter().map(|msg| msg.role.clone()).collect(); - *handler_text.lock().expect("text lock") = req - .history - .iter() - .map(|msg| msg.content.as_str()) - .collect::>() - .join("\n---\n"); - *handler_provider.lock().expect("provider lock") = req.provider_name; - *handler_channel.lock().expect("channel lock") = req.channel_name; - - if let Some(tx) = req.on_progress { - handler_progress.fetch_add(1, Ordering::SeqCst); - let _ = tx.send(AgentProgress::TurnStarted).await; - let _ = tx - .send(AgentProgress::ThinkingDelta { - delta: "thinking".to_string(), - iteration: 1, - }) - .await; - let _ = tx - .send(AgentProgress::TextDelta { - delta: "partial ".to_string(), - iteration: 1, - }) - .await; - let _ = tx - .send(AgentProgress::ToolCallStarted { - call_id: "call-1".to_string(), - tool_name: "harness_tool".to_string(), - arguments: serde_json::json!({}), - iteration: 1, - display_label: None, - display_detail: None, - }) - .await; - } - - if !handler_delay.is_zero() { - tokio::time::sleep(handler_delay).await; - } - - match handler_error { - Some(message) => Err(message), - None => Ok(AgentTurnResponse::new(response_text)), + move |req| { + let handler_roles = Arc::clone(&handler_roles); + let handler_text = Arc::clone(&handler_text); + let handler_provider = Arc::clone(&handler_provider); + let handler_channel = Arc::clone(&handler_channel); + let handler_progress = Arc::clone(&handler_progress); + let response_text = response_text.clone(); + let handler_error = handler_error.clone(); + async move { + *handler_roles.lock().expect("roles lock") = + req.history.iter().map(|msg| msg.role.clone()).collect(); + *handler_text.lock().expect("text lock") = req + .history + .iter() + .map(|msg| msg.content.as_str()) + .collect::>() + .join("\n---\n"); + *handler_provider.lock().expect("provider lock") = req.provider_name; + *handler_channel.lock().expect("channel lock") = req.channel_name; + + if let Some(tx) = req.on_progress { + handler_progress.fetch_add(1, Ordering::SeqCst); + let _ = tx.send(AgentProgress::TurnStarted).await; + let _ = tx + .send(AgentProgress::ThinkingDelta { + delta: "thinking".to_string(), + iteration: 1, + }) + .await; + let _ = tx + .send(AgentProgress::TextDelta { + delta: "partial ".to_string(), + iteration: 1, + }) + .await; + let _ = tx + .send(AgentProgress::ToolCallStarted { + call_id: "call-1".to_string(), + tool_name: "harness_tool".to_string(), + arguments: serde_json::json!({}), + iteration: 1, + display_label: None, + display_detail: None, + }) + .await; + } + + if !handler_delay.is_zero() { + tokio::time::sleep(handler_delay).await; + } + + match handler_error { + Some(message) => Err(message), + None => Ok(AgentTurnResponse::new(response_text)), + } } } - } - }); + }); let state = Arc::new(HarnessState::default()); let channel_impl = Arc::new(HarnessChannel { diff --git a/src/openhuman/channels/tests/health.rs b/src/openhuman/channels/tests/health.rs index fd24281ee5..48d5142aab 100644 --- a/src/openhuman/channels/tests/health.rs +++ b/src/openhuman/channels/tests/health.rs @@ -42,10 +42,11 @@ async fn supervised_listener_marks_error_and_restarts_on_failures() { // The global health subscriber may have been registered by another test // runtime; keep a fresh subscriber alive for this test's runtime too. crate::core::bus::init().await.expect("bus init"); - let _health_handle = crate::core::bus::BUS.subscribe(Arc::new( - crate::openhuman::platform::health::bus::HealthSubscriber, - )) - .expect("event bus should be initialized for channel health test"); + let _health_handle = crate::core::bus::BUS + .subscribe(Arc::new( + crate::openhuman::platform::health::bus::HealthSubscriber, + )) + .expect("event bus should be initialized for channel health test"); tokio::task::yield_now().await; let handle = spawn_supervised_listener(channel, tx, 1, 1); diff --git a/src/openhuman/channels/tests/runtime_dispatch.rs b/src/openhuman/channels/tests/runtime_dispatch.rs index 11cb708f04..cb607c967d 100644 --- a/src/openhuman/channels/tests/runtime_dispatch.rs +++ b/src/openhuman/channels/tests/runtime_dispatch.rs @@ -352,7 +352,8 @@ async fn dispatch_routes_through_agent_run_turn_bus_handler() { #[tokio::test] async fn channel_processed_event_records_resolved_agent_route() { crate::core::bus::init().await.expect("bus init"); - let mut events = crate::core::bus::BUS.get() + let mut events = crate::core::bus::BUS + .get() .expect("event bus should be initialized") .receiver(); diff --git a/src/openhuman/config/ops/agent.rs b/src/openhuman/config/ops/agent.rs index f89078f740..b14db1631d 100644 --- a/src/openhuman/config/ops/agent.rs +++ b/src/openhuman/config/ops/agent.rs @@ -133,9 +133,7 @@ pub async fn apply_autonomy_settings( config.save().await.map_err(|e| e.to_string())?; crate::openhuman::security::live_policy::reload_from(&config.autonomy); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::AutonomyConfigChanged, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::AutonomyConfigChanged); let snapshot = snapshot_config_json(config)?; Ok(RpcOutcome::new( @@ -472,9 +470,7 @@ pub async fn apply_agent_paths_settings( config.save().await.map_err(|e| e.to_string())?; crate::openhuman::security::live_policy::set_action_dir(config.action_dir.clone()); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::AgentPathsChanged, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::AgentPathsChanged); log::debug!( "[config][agent_paths] action_dir now '{}' (source={})", diff --git a/src/openhuman/cron/bus.rs b/src/openhuman/cron/bus.rs index 90bd5447d0..1d62e24558 100644 --- a/src/openhuman/cron/bus.rs +++ b/src/openhuman/cron/bus.rs @@ -7,10 +7,10 @@ //! channel construction out of the scheduler. use crate::core::events::DomainEvent; -use tinybus::EventHandler; use async_trait::async_trait; use std::collections::HashMap; use std::sync::Arc; +use tinybus::EventHandler; use tinychannels::{Channel, SendMessage}; /// Subscribes to `CronDeliveryRequested` events and dispatches diff --git a/src/openhuman/cron/scheduler_tests.rs b/src/openhuman/cron/scheduler_tests.rs index 365298af89..ff71f5e427 100644 --- a/src/openhuman/cron/scheduler_tests.rs +++ b/src/openhuman/cron/scheduler_tests.rs @@ -1201,8 +1201,8 @@ async fn deliver_if_configured_skips_non_announce_mode() { #[tokio::test] async fn deliver_if_configured_publishes_event_for_announce_mode() { use crate::core::events::DomainEvent; -use tinybus::EventHandler; use std::sync::atomic::{AtomicUsize, Ordering}; + use tinybus::EventHandler; // Create an isolated bus for this test. let bus = crate::core::bus_testing::isolated_bus().await; @@ -1741,10 +1741,10 @@ fn classify_agent_anyhow_does_not_leak_when_downcast_succeeds() { #[tokio::test] async fn scheduler_tick_once_publishes_health_recovery_signal_on_empty_queue() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; -use tinybus::EventHandler; + use crate::core::events::DomainEvent; use async_trait::async_trait; use std::sync::Mutex as StdMutex; + use tinybus::EventHandler; #[derive(Default)] struct HealthEventCollector { diff --git a/src/openhuman/desktop/notifications/bus.rs b/src/openhuman/desktop/notifications/bus.rs index 026688a5ac..f8ec81efd9 100644 --- a/src/openhuman/desktop/notifications/bus.rs +++ b/src/openhuman/desktop/notifications/bus.rs @@ -14,9 +14,9 @@ use std::time::{SystemTime, UNIX_EPOCH}; use tokio::sync::broadcast; use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::config::Config; use async_trait::async_trait; +use tinybus::EventHandler; use super::types::{CoreNotificationCategory, CoreNotificationEvent}; @@ -253,9 +253,9 @@ impl EventHandler for NotificationBridgeSubscriber { /// but the caller (`register_domain_subscribers`) is Once-guarded. pub fn register_notification_bridge_subscriber(config: Config) { use std::sync::Arc; - if let Some(handle) = crate::core::bus::BUS.subscribe(Arc::new( - NotificationBridgeSubscriber::new(config), - )) { + if let Some(handle) = + crate::core::bus::BUS.subscribe(Arc::new(NotificationBridgeSubscriber::new(config))) + { // SAFETY: intentional leak; handle's Drop would cancel the subscriber. std::mem::forget(handle); log::info!("{LOG_PREFIX} notification bridge subscriber registered"); diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index 6e0068be11..39d58f3f57 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -11,7 +11,6 @@ //! dispatch on enable/disable. use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::config::Config; use crate::openhuman::flows::store; use crate::openhuman::flows::{flow_namespace, Flow, FlowRun}; @@ -20,6 +19,7 @@ use async_trait::async_trait; use serde_json::Value; use std::collections::{HashMap, HashSet}; use std::sync::{Arc, LazyLock, Mutex}; +use tinybus::EventHandler; use tinyflows::model::{NodeKind, TriggerKind}; use tinyflows::nodes::control_flow::dedup as dedup_node; diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 5198eadb61..27b913acd7 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -5731,13 +5731,11 @@ pub async fn sweep_expired_parked_runs(config: &Config) -> usize { flow_id, "[flows] TTL sweep: publishing FlowRunFinished for expired parked run" ); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::FlowRunFinished { - flow_id: flow_id.to_string(), - run_id: run_id.to_string(), - status: "cancelled".to_string(), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::FlowRunFinished { + flow_id: flow_id.to_string(), + run_id: run_id.to_string(), + status: "cancelled".to_string(), + }); drop_checkpoint(config, run_id).await; } if !swept.is_empty() { @@ -5806,13 +5804,11 @@ pub async fn sweep_orphaned_running_runs_on_boot(config: &Config) -> usize { if let Err(e) = store::record_run(config, &flow_id, "interrupted") { tracing::warn!(target: "flows", run_id = %run_id, flow_id = %flow_id, error = %e, "[flows] boot sweep: failed to update flow summary for reconciled run"); } - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::FlowRunFinished { - flow_id: flow_id.clone(), - run_id: run_id.clone(), - status: "interrupted".to_string(), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::FlowRunFinished { + flow_id: flow_id.clone(), + run_id: run_id.clone(), + status: "interrupted".to_string(), + }); drop_checkpoint(config, &run_id).await; tracing::info!(target: "flows", run_id = %run_id, flow_id = %flow_id, "[flows] boot sweep: reconciled orphaned running run to 'interrupted'"); } diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 10b60c78f8..affba8812a 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -2955,10 +2955,10 @@ async fn flows_run_does_not_notify_when_run_completes_without_pending_approvals( #[tokio::test] async fn flows_run_publishes_flow_run_started_with_flow_and_run_id() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; -use tinybus::EventHandler; + use crate::core::events::DomainEvent; use async_trait::async_trait; use std::sync::Mutex as StdMutex; + use tinybus::EventHandler; #[derive(Default)] struct Collector { @@ -3043,10 +3043,10 @@ use tinybus::EventHandler; #[tokio::test] async fn flows_run_finished_event_skips_pending_approval_and_fires_once_on_resume() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; -use tinybus::EventHandler; + use crate::core::events::DomainEvent; use async_trait::async_trait; use std::sync::Mutex as StdMutex; + use tinybus::EventHandler; #[derive(Default)] struct Collector { diff --git a/src/openhuman/inference/provider/factory_tests.rs b/src/openhuman/inference/provider/factory_tests.rs index c2ed3a606b..010c828963 100644 --- a/src/openhuman/inference/provider/factory_tests.rs +++ b/src/openhuman/inference/provider/factory_tests.rs @@ -1092,7 +1092,7 @@ fn configured_openhuman_jwt_slug_routes_to_managed_chat_model() { #[tokio::test] async fn openhuman_jwt_slug_discloses_pinned_model() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; use crate::openhuman::security::egress::{EgressDescriptor, EgressReason}; use std::time::Duration; @@ -1141,7 +1141,7 @@ use crate::core::events::DomainEvent; #[tokio::test] async fn native_claude_turn_routes_disclose_pinned_models() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; use crate::openhuman::security::egress::EgressDescriptor; use std::time::Duration; @@ -1327,7 +1327,7 @@ fn crate_native_chat_model_factory_preserves_invalid_route_diagnostics() { #[tokio::test] async fn from_string_external_provider_emits_egress_realpath() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; use crate::openhuman::security::egress::EgressReason; crate::core::bus::init().await.expect("bus init"); @@ -1350,7 +1350,7 @@ use crate::core::events::DomainEvent; } Some(_) => continue, None => panic!("the bus closed before the expected event arrived"), -} + } } }) .await; @@ -1369,7 +1369,7 @@ use crate::core::events::DomainEvent; #[tokio::test] async fn create_chat_model_managed_emits_exactly_one_egress_realpath() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; use crate::openhuman::security::egress::{EgressDescriptor, EgressReason}; use std::time::Duration; @@ -1422,7 +1422,7 @@ use crate::core::events::DomainEvent; #[tokio::test] async fn create_chat_model_local_runtime_does_not_emit_egress_realpath() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; use crate::openhuman::security::egress::EgressDescriptor; use std::time::Duration; diff --git a/src/openhuman/inference/provider/openhuman_backend_model.rs b/src/openhuman/inference/provider/openhuman_backend_model.rs index ef88611191..18502f02e0 100644 --- a/src/openhuman/inference/provider/openhuman_backend_model.rs +++ b/src/openhuman/inference/provider/openhuman_backend_model.rs @@ -390,16 +390,14 @@ fn maybe_publish_session_expired(err: &TinyAgentsError, operation: &str) { if pe.provider.as_str() == "OpenHuman" && matches!(pe.status, Some(401 | 403)) { let reason = crate::openhuman::inference::provider::ops::sanitize_api_error(&pe.message); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::SessionExpired { - source: format!( - "openhuman_backend_model.{}({})", - operation, - pe.status.unwrap_or(0) - ), - reason, - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::SessionExpired { + source: format!( + "openhuman_backend_model.{}({})", + operation, + pe.status.unwrap_or(0) + ), + reason, + }); } } } diff --git a/src/openhuman/inference/provider/ops/http_error.rs b/src/openhuman/inference/provider/ops/http_error.rs index 166410948b..c407ed59ac 100644 --- a/src/openhuman/inference/provider/ops/http_error.rs +++ b/src/openhuman/inference/provider/ops/http_error.rs @@ -837,15 +837,13 @@ pub fn log_byo_provider_auth_failure( // re-flooding the notification center the way the raw error flooded Sentry. let status_code = status.as_u16(); if crate::openhuman::inference::auth_error_registry::record(provider, status_code) { - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::ProviderApiKeyRejected { - provider: provider.to_string(), - message: crate::openhuman::inference::auth_error_registry::auth_error_message( - provider, - status_code, - ), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::ProviderApiKeyRejected { + provider: provider.to_string(), + message: crate::openhuman::inference::auth_error_registry::auth_error_message( + provider, + status_code, + ), + }); } } diff --git a/src/openhuman/inference/provider/ops_tests.rs b/src/openhuman/inference/provider/ops_tests.rs index f3714a6bde..6b69ee55b3 100644 --- a/src/openhuman/inference/provider/ops_tests.rs +++ b/src/openhuman/inference/provider/ops_tests.rs @@ -1332,10 +1332,13 @@ async fn api_error_monthly_quota_returns_message_via_demoted_branch() { #[tokio::test] async fn publish_backend_session_expired_emits_sanitized_session_expired() { use crate::core::bus::BUS; -use crate::core::events::DomainEvent; + use crate::core::events::DomainEvent; crate::core::bus::init().await.expect("bus init"); - let mut rx = crate::core::bus::BUS.get().expect("event bus initialized").receiver(); + let mut rx = crate::core::bus::BUS + .get() + .expect("event bus initialized") + .receiver(); // `TEST_MARKER_A` makes this event distinguishable from the sibling // `chat_completions_backend_401_*` test's event on the shared global diff --git a/src/openhuman/integrations/composio/ops/direct_mode.rs b/src/openhuman/integrations/composio/ops/direct_mode.rs index 029d0079ce..7e8b0ac71c 100644 --- a/src/openhuman/integrations/composio/ops/direct_mode.rs +++ b/src/openhuman/integrations/composio/ops/direct_mode.rs @@ -111,12 +111,10 @@ pub async fn composio_set_api_key( config.composio.mode.clone() }; - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::ComposioConfigChanged { - mode: effective_mode.clone(), - api_key_set: true, - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::ComposioConfigChanged { + mode: effective_mode.clone(), + api_key_set: true, + }); tracing::debug!( mode = %effective_mode, "[composio-cache] published ComposioConfigChanged after set_api_key" @@ -149,12 +147,10 @@ pub async fn composio_clear_api_key(config: &Config) -> OpResult MemoryBinding { ); // Sync, and a no-op when the bus is not yet initialized, so this is // safe to call pre-boot with no `#[cfg(test)]` guard. - crate::core::event_bus::publish_global( - crate::core::event_bus::DomainEvent::MemoryDriverBindFailed { + crate::core::bus::BUS.publish( + crate::core::events::DomainEvent::MemoryDriverBindFailed { configured_driver: fallback.configured_driver.clone(), bound_driver: NULL_DRIVER_ID.to_string(), reason: fallback.reason.clone(), diff --git a/src/openhuman/memory/conversations/bus.rs b/src/openhuman/memory/conversations/bus.rs index 5dd01e59e4..864246869f 100644 --- a/src/openhuman/memory/conversations/bus.rs +++ b/src/openhuman/memory/conversations/bus.rs @@ -44,9 +44,9 @@ pub fn register_conversation_persistence_subscriber(workspace_dir: PathBuf) { return; } - match crate::core::bus::BUS.subscribe(Arc::new( - ConversationPersistenceSubscriber::new_shared(Arc::clone(workspace)), - )) { + match crate::core::bus::BUS.subscribe(Arc::new(ConversationPersistenceSubscriber::new_shared( + Arc::clone(workspace), + ))) { Some(handle) => { let _ = CONVERSATION_PERSISTENCE_HANDLE.set(handle); } diff --git a/src/openhuman/memory/diff/ops.rs b/src/openhuman/memory/diff/ops.rs index 02c0c1d024..3236e38c5b 100644 --- a/src/openhuman/memory/diff/ops.rs +++ b/src/openhuman/memory/diff/ops.rs @@ -59,15 +59,13 @@ pub async fn take_snapshot( "[memory_diff] snapshot taken" ); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::MemoryDiffSnapshotTaken { - snapshot_id: snapshot.id.clone(), - source_id: source.id.clone(), - source_kind: source.kind.as_str().to_string(), - item_count: snapshot.item_count as usize, - trigger: snapshot.trigger.as_str().to_string(), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::MemoryDiffSnapshotTaken { + snapshot_id: snapshot.id.clone(), + source_id: source.id.clone(), + source_kind: source.kind.as_str().to_string(), + item_count: snapshot.item_count as usize, + trigger: snapshot.trigger.as_str().to_string(), + }); Ok(snapshot) } @@ -230,12 +228,10 @@ pub async fn mark_read(config: &Config, source_ids: Option>) -> Resu "[memory_diff] mark_read committed read markers" ); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::MemoryDiffMarkedRead { - source_ids: target_ids, - snapshot_ids, - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::MemoryDiffMarkedRead { + source_ids: target_ids, + snapshot_ids, + }); Ok(marked) } diff --git a/src/openhuman/memory/global.rs b/src/openhuman/memory/global.rs index e5c415cec1..b94d35ff7d 100644 --- a/src/openhuman/memory/global.rs +++ b/src/openhuman/memory/global.rs @@ -221,10 +221,7 @@ fn cached_client(workspace_dir: &Path) -> Result, String /// A racing caller may have inserted first; theirs wins, so the "one ingestion /// worker per workspace" property holds even when two paths construct /// concurrently. Callers must use the returned handle, not the one they passed. -fn cache_client( - workspace_dir: &Path, - client: &MemoryClientRef, -) -> Result { +fn cache_client(workspace_dir: &Path, client: &MemoryClientRef) -> Result { let mut guard = WORKSPACE_CLIENTS .get_or_init(Default::default) .write() diff --git a/src/openhuman/memory/guard/audit.rs b/src/openhuman/memory/guard/audit.rs index 92c105bc23..4b1f80fd40 100644 --- a/src/openhuman/memory/guard/audit.rs +++ b/src/openhuman/memory/guard/audit.rs @@ -30,7 +30,8 @@ use tinycortex_api::capabilities::Capability; -use crate::core::event_bus::{publish_global, DomainEvent}; +use crate::core::bus::BUS; +use crate::core::events::DomainEvent; use crate::openhuman::memory::util::redact::redact; use super::policy::GuardPolicy; @@ -91,7 +92,7 @@ pub fn trace_budget(policy: &GuardPolicy, method: &str, dropped: usize, trimmed_ /// /// Called from [`GuardPolicy::denied`](super::GuardPolicy::denied), so every /// deny path audits by construction rather than by each call site remembering -/// to. `publish_global` is synchronous and a no-op before the bus is +/// to. `BUS.publish` is synchronous and a no-op before the bus is /// initialised, so this is safe pre-boot with no `#[cfg(test)]` guard — the /// same property `binding::build` relies on. pub fn publish_guard_denied(policy: &GuardPolicy, method: &str, reason: &str) { @@ -100,7 +101,7 @@ pub fn publish_guard_denied(policy: &GuardPolicy, method: &str, reason: &str) { policy.driver_id(), policy.class(), ); - publish_global(DomainEvent::MemoryGuardDenied { + BUS.publish(DomainEvent::MemoryGuardDenied { driver_id: policy.driver_id().to_string(), method: method.to_string(), reason: reason.to_string(), diff --git a/src/openhuman/memory/guard/provider_tests.rs b/src/openhuman/memory/guard/provider_tests.rs index f716454630..1918dc8642 100644 --- a/src/openhuman/memory/guard/provider_tests.rs +++ b/src/openhuman/memory/guard/provider_tests.rs @@ -13,7 +13,8 @@ use tinycortex_api::provider::{ use tinycortex_api::recall::OwnedRecallOpts; use tinycortex_api::types::{MemoryCategory, MemoryTaint}; -use crate::core::event_bus::{init_global, DomainEvent, DEFAULT_CAPACITY}; +use crate::core::bus::BUS; +use crate::core::events::DomainEvent; use crate::core::subsystem::DriverClass; use crate::openhuman::config::schema::MemoryHooksConfig; use crate::openhuman::memory::guard::policy::TRUSTED; @@ -193,9 +194,76 @@ async fn guard_does_not_budget_trim_an_export() { // ── Step 7 ────────────────────────────────────────────────────────────────── +/// Records every `MemoryGuardDenied` the global bus delivers. +/// +/// The guard publishes onto the process-wide [`BUS`], so an isolated bus +/// cannot see it — this has to subscribe to the real one. Delivery is async +/// now (tinybus routes through a broker), where the old `raw_receiver()` was a +/// synchronous broadcast channel, so callers poll rather than `try_recv` once. +struct DeniedRecorder { + seen: std::sync::Mutex>, +} + +#[async_trait::async_trait] +impl tinybus::EventHandler for DeniedRecorder { + fn name(&self) -> &str { + "memory::guard::test_recorder" + } + + async fn handle(&self, event: &DomainEvent) { + if let DomainEvent::MemoryGuardDenied { + driver_id, + method, + reason, + } = event + { + self.seen.lock().expect("recorder mutex").push(( + driver_id.clone(), + method.clone(), + reason.clone(), + )); + } + } +} + +/// Subscribe a recorder to the global bus, initialising it if needed. +/// +/// Returns the recorder and its subscription handle; the handle must stay +/// alive for the test's duration or the subscription is dropped. +async fn record_denials() -> (Arc, tinybus::SubscriptionHandle) { + crate::core::bus::init().await.expect("bus init"); + let recorder = Arc::new(DeniedRecorder { + seen: std::sync::Mutex::new(Vec::new()), + }); + let handle = BUS + .subscribe(recorder.clone()) + .expect("the bus was just initialised"); + (recorder, handle) +} + +/// Poll the recorder for up to ~2s for a denial matching `driver_id`. +/// +/// Bounded rather than unbounded so a regression fails the test instead of +/// hanging CI. +async fn await_denial(recorder: &DeniedRecorder, driver_id: &str) -> (String, String, String) { + for _ in 0..200 { + if let Some(found) = recorder + .seen + .lock() + .expect("recorder mutex") + .iter() + .find(|(id, _, _)| id == driver_id) + { + return found.clone(); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + panic!("no MemoryGuardDenied event for driver '{driver_id}' within 2s"); +} + #[tokio::test] async fn guard_publishes_memory_guard_denied_on_refusal() { - let mut rx = init_global(DEFAULT_CAPACITY).raw_receiver(); + let (recorder, _handle) = record_denials().await; let (driver, guard) = guarded(external_policy("untrusted")); let err = guard .store( @@ -211,27 +279,14 @@ async fn guard_publishes_memory_guard_denied_on_refusal() { assert!(err.to_string().contains("memory guard: ")); assert_eq!(driver.call_count(), 0, "the driver must never be reached"); - let mut seen = None; - while let Ok(event) = rx.try_recv() { - if let DomainEvent::MemoryGuardDenied { - driver_id, - method, - reason, - } = event - { - seen = Some((driver_id, method, reason)); - break; - } - } - let (driver_id, method, reason) = seen.expect("a MemoryGuardDenied event"); - assert_eq!(driver_id, "supermemory"); + let (_driver_id, method, reason) = await_denial(&recorder, "supermemory").await; assert_eq!(method, "core.store"); assert!(!reason.contains("hello"), "must never carry content"); } #[tokio::test] async fn guard_publishes_nothing_on_the_success_path() { - let mut rx = init_global(DEFAULT_CAPACITY).raw_receiver(); + let (recorder, _handle) = record_denials().await; let (_driver, guard) = guarded(embedded_policy()); guard .store( @@ -249,16 +304,18 @@ async fn guard_publishes_nothing_on_the_success_path() { .await .expect("recall"); - // Sibling tests share the process-global bus and run in parallel, so - // filter to *this* guard's driver id rather than asserting the channel is - // empty — `guard_publishes_memory_guard_denied_on_refusal` legitimately + // Delivery is async, so give a stray publish time to arrive — asserting + // "nothing" immediately after the calls would pass even if the guard did + // publish. Sibling tests share the process-global bus and run in parallel, + // so filter to *this* guard's driver id rather than asserting the recorder + // is empty: `guard_publishes_memory_guard_denied_on_refusal` legitimately // publishes one (for `supermemory`) at the same time. - while let Ok(event) = rx.try_recv() { - if let DomainEvent::MemoryGuardDenied { driver_id, .. } = &event { - assert_ne!( - driver_id, "recording", - "a guarded read/write must not publish on success" - ); - } - } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let seen = recorder.seen.lock().expect("recorder mutex"); + assert!( + !seen + .iter() + .any(|(driver_id, _, _)| driver_id == "recording"), + "a guarded read/write must not publish on success, saw: {seen:?}" + ); } diff --git a/src/openhuman/memory/ops/sync.rs b/src/openhuman/memory/ops/sync.rs index 0dcf900d29..3dc05a638e 100644 --- a/src/openhuman/memory/ops/sync.rs +++ b/src/openhuman/memory/ops/sync.rs @@ -61,11 +61,9 @@ pub async fn memory_sync_channel( ) -> Result, String> { // `channel_id` is a user/context identifier — keep it out of normal logs. tracing::info!("[memory.sync] memory_sync_channel: entry"); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::MemorySyncRequested { - channel_id: Some(params.channel_id.clone()), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::MemorySyncRequested { + channel_id: Some(params.channel_id.clone()), + }); emit_sync_stage( MemorySyncTrigger::Manual, MemorySyncStage::Requested, @@ -97,9 +95,8 @@ pub async fn memory_sync_channel( /// ingestion subscribers. pub async fn memory_sync_all() -> Result, String> { tracing::info!("[memory.sync] memory_sync_all: entry"); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::MemorySyncRequested { channel_id: None }, - ); + crate::core::bus::BUS + .publish(crate::core::events::DomainEvent::MemorySyncRequested { channel_id: None }); emit_sync_stage( MemorySyncTrigger::Manual, MemorySyncStage::Requested, @@ -243,8 +240,8 @@ mod tests { use tokio::time::{timeout, Duration}; use crate::core::bus::BUS; -use crate::core::events::DomainEvent; -use tinybus::EventHandler; + use crate::core::events::DomainEvent; + use tinybus::EventHandler; fn test_mutex() -> &'static std::sync::Mutex<()> { static LOCK: OnceLock> = OnceLock::new(); @@ -320,7 +317,8 @@ use tinybus::EventHandler; .unwrap_or_else(|poisoned| poisoned.into_inner()); let _ = crate::core::bus::init().await; let (tx, mut rx) = mpsc::unbounded_channel(); - let _subscription = BUS.subscribe(Arc::new(ChannelCapture { tx })) + let _subscription = BUS + .subscribe(Arc::new(ChannelCapture { tx })) .expect("global bus should be initialized"); let outcome = memory_sync_channel(SyncChannelParams { @@ -348,7 +346,8 @@ use tinybus::EventHandler; .unwrap_or_else(|poisoned| poisoned.into_inner()); let _ = crate::core::bus::init().await; let (tx, mut rx) = mpsc::unbounded_channel(); - let _subscription = BUS.subscribe(Arc::new(ChannelCapture { tx })) + let _subscription = BUS + .subscribe(Arc::new(ChannelCapture { tx })) .expect("global bus should be initialized"); let outcome = memory_sync_all().await.expect("memory_sync_all"); diff --git a/src/openhuman/memory/store/client.rs b/src/openhuman/memory/store/client.rs index 0b46502615..f0eb043603 100644 --- a/src/openhuman/memory/store/client.rs +++ b/src/openhuman/memory/store/client.rs @@ -191,14 +191,12 @@ impl MemoryClient { let queue_depth = state.snapshot().queue_depth; state.mark_running(&placeholder_id, &title, &namespace); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::MemoryIngestionStarted { - document_id: placeholder_id.clone(), - title, - namespace: namespace.clone(), - queue_depth, - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::MemoryIngestionStarted { + document_id: placeholder_id.clone(), + title, + namespace: namespace.clone(), + queue_depth, + }); let started = std::time::Instant::now(); let outcome = self.inner.ingest_document(request).await; @@ -214,15 +212,13 @@ impl MemoryClient { success, chrono::Utc::now().timestamp_millis(), ); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::MemoryIngestionCompleted { - document_id: placeholder_id, - namespace, - success, - elapsed_ms, - queue_depth: state.snapshot().queue_depth, - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::MemoryIngestionCompleted { + document_id: placeholder_id, + namespace, + success, + elapsed_ms, + queue_depth: state.snapshot().queue_depth, + }); outcome } diff --git a/src/openhuman/memory/sync/composio/bus.rs b/src/openhuman/memory/sync/composio/bus.rs index 7cf2dbd8a2..b845b4d278 100644 --- a/src/openhuman/memory/sync/composio/bus.rs +++ b/src/openhuman/memory/sync/composio/bus.rs @@ -53,12 +53,12 @@ use async_trait::async_trait; use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; -use tinybus::SubscriptionHandle; use crate::openhuman::agent::triage::{apply_decision, run_triage, TriageOutcome, TriggerEnvelope}; use crate::openhuman::config::rpc as config_rpc; use crate::openhuman::config::schema::COMPOSIO_MODE_DIRECT; use crate::openhuman::integrations::composio::trigger_history; +use tinybus::EventHandler; +use tinybus::SubscriptionHandle; use super::providers::{get_provider, ProviderContext}; use crate::openhuman::integrations::composio::client::ComposioClient; @@ -894,11 +894,9 @@ impl EventHandler for ComposioConfigChangedSubscriber { .collect(); toolkits.sort(); toolkits.dedup(); - crate::core::bus::BUS.publish( - DomainEvent::ComposioIntegrationsChanged { - toolkits: toolkits.clone(), - }, - ); + crate::core::bus::BUS.publish(DomainEvent::ComposioIntegrationsChanged { + toolkits: toolkits.clone(), + }); tracing::debug!( active_toolkits = ?toolkits, "[composio-cache] config changed eager warm complete; published integrations changed" diff --git a/src/openhuman/memory/sync_events.rs b/src/openhuman/memory/sync_events.rs index f80fc5749f..c0e05a6b68 100644 --- a/src/openhuman/memory/sync_events.rs +++ b/src/openhuman/memory/sync_events.rs @@ -307,8 +307,9 @@ mod tests { crate::core::bus::init().await.expect("bus init"); let collector = StageCollector::default(); - let _subscription = - BUS.subscribe(Arc::new(collector.clone())).expect("event bus initialized"); + let _subscription = BUS + .subscribe(Arc::new(collector.clone())) + .expect("event bus initialized"); let bridge = MemorySyncStageBridge; bridge @@ -346,8 +347,9 @@ mod tests { crate::core::bus::init().await.expect("bus init"); let collector = StageCollector::default(); - let _subscription = - BUS.subscribe(Arc::new(collector.clone())).expect("event bus initialized"); + let _subscription = BUS + .subscribe(Arc::new(collector.clone())) + .expect("event bus initialized"); let bridge = MemorySyncStageBridge; bridge @@ -437,8 +439,9 @@ mod tests { crate::core::bus::init().await.expect("bus init"); let collector = StageCollector::default(); - let _subscription = - BUS.subscribe(Arc::new(collector.clone())).expect("event bus initialized"); + let _subscription = BUS + .subscribe(Arc::new(collector.clone())) + .expect("event bus initialized"); let bridge = MemorySyncStageBridge; bridge @@ -485,8 +488,9 @@ mod tests { crate::core::bus::init().await.expect("bus init"); let collector = StageCollector::default(); - let _subscription = - BUS.subscribe(Arc::new(collector.clone())).expect("event bus initialized"); + let _subscription = BUS + .subscribe(Arc::new(collector.clone())) + .expect("event bus initialized"); let bridge = MemorySyncStageBridge; // Non-memory-source sync (e.g. Slack channel sync) should have source_id=None @@ -532,8 +536,9 @@ mod tests { crate::core::bus::init().await.expect("bus init"); let collector = StageCollector::default(); - let _subscription = - BUS.subscribe(Arc::new(collector.clone())).expect("event bus initialized"); + let _subscription = BUS + .subscribe(Arc::new(collector.clone())) + .expect("event bus initialized"); let bridge = MemorySyncStageBridge; bridge @@ -585,8 +590,9 @@ mod tests { crate::core::bus::init().await.expect("bus init"); let collector = StageCollector::default(); - let _subscription = - BUS.subscribe(Arc::new(collector.clone())).expect("event bus initialized"); + let _subscription = BUS + .subscribe(Arc::new(collector.clone())) + .expect("event bus initialized"); let bridge = MemorySyncStageBridge; // Non-memory-source ingestion (plain document_id, no mem_src prefix) diff --git a/src/openhuman/memory/sync_pipeline_e2e_tests.rs b/src/openhuman/memory/sync_pipeline_e2e_tests.rs index 12fd67d16d..667a13f4f8 100644 --- a/src/openhuman/memory/sync_pipeline_e2e_tests.rs +++ b/src/openhuman/memory/sync_pipeline_e2e_tests.rs @@ -22,8 +22,6 @@ use tempfile::TempDir; use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; -use tinybus::SubscriptionHandle; use crate::openhuman::config::Config; use crate::openhuman::memory::ingest_pipeline::ingest_chat; use crate::openhuman::memory::queue::{ @@ -36,6 +34,8 @@ use crate::openhuman::memory::store::trees::{store as tree_store, types::TreeKin use crate::openhuman::memory::sync_events::{emit_sync_stage, MemorySyncStage, MemorySyncTrigger}; use crate::openhuman::memory::tree::retrieval::{query_source, search_entities}; use crate::openhuman::memory::tree::score::store::lookup_entity; +use tinybus::EventHandler; +use tinybus::SubscriptionHandle; use tinycortex::memory::ingest::canonicalize::chat::{ChatBatch, ChatMessage}; // ── helpers ───────────────────────────────────────────────────────────── diff --git a/src/openhuman/memory/tinycortex/sync.rs b/src/openhuman/memory/tinycortex/sync.rs index 41b8003d6e..ea53db5ad9 100644 --- a/src/openhuman/memory/tinycortex/sync.rs +++ b/src/openhuman/memory/tinycortex/sync.rs @@ -674,16 +674,14 @@ impl SyncStateStore for HostSyncAdapter { #[async_trait] impl SyncEventSink for HostSyncAdapter { async fn emit(&self, event: SyncEvent) -> anyhow::Result<()> { - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::MemorySyncStageChanged { - trigger: "tinycortex".into(), - stage: stage_name(event.stage).into(), - provider: Some(event.toolkit), - connection_id: event.connection_id, - detail: event.message, - source_id: Some(event.source_id), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::MemorySyncStageChanged { + trigger: "tinycortex".into(), + stage: stage_name(event.stage).into(), + provider: Some(event.toolkit), + connection_id: event.connection_id, + detail: event.message, + source_id: Some(event.source_id), + }); Ok(()) } } diff --git a/src/openhuman/memory/tree/tree_runtime/bus.rs b/src/openhuman/memory/tree/tree_runtime/bus.rs index 09870fdcb8..e51c071bc5 100644 --- a/src/openhuman/memory/tree/tree_runtime/bus.rs +++ b/src/openhuman/memory/tree/tree_runtime/bus.rs @@ -4,8 +4,8 @@ //! Future subscribers can react to these events for cross-module workflows. use crate::core::events::DomainEvent; -use tinybus::EventHandler; use async_trait::async_trait; +use tinybus::EventHandler; /// Subscribes to tree summarizer events and logs activity. pub struct TreeSummarizerEventSubscriber; diff --git a/src/openhuman/security/approval/gate.rs b/src/openhuman/security/approval/gate.rs index b7eba0884e..f99321ab1d 100644 --- a/src/openhuman/security/approval/gate.rs +++ b/src/openhuman/security/approval/gate.rs @@ -3296,7 +3296,8 @@ mod tests { // (bridged to a broadcast Socket.IO event by `core::socketio`) and // the `flow-gate-approval` CoreNotification with its three actions. crate::core::bus::init().await.expect("bus init"); - let mut event_rx = crate::core::bus::BUS.get() + let mut event_rx = crate::core::bus::BUS + .get() .expect("event bus initialized above") .receiver(); let mut notif_rx = @@ -3368,7 +3369,7 @@ mod tests { }) if flow_id == expected_flow_id => return (request_id, run_id, tool_name), Some(_) => continue, None => panic!("the bus closed before the expected event arrived"), -} + } } } diff --git a/src/openhuman/security/credentials/bus.rs b/src/openhuman/security/credentials/bus.rs index 4d51eca32c..3852a60c2a 100644 --- a/src/openhuman/security/credentials/bus.rs +++ b/src/openhuman/security/credentials/bus.rs @@ -21,9 +21,9 @@ //! cron-driven LLM calls after session expiry). use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::cron::scheduler_gate; use async_trait::async_trait; +use tinybus::EventHandler; /// Subscribes to [`DomainEvent::SessionExpired`] and runs the canonical /// session-teardown. Singleton — register once at startup. diff --git a/src/openhuman/security/credentials/session_support.rs b/src/openhuman/security/credentials/session_support.rs index 5c26e4b4fb..fd3114f40f 100644 --- a/src/openhuman/security/credentials/session_support.rs +++ b/src/openhuman/security/credentials/session_support.rs @@ -219,14 +219,11 @@ pub fn require_live_session_token(config: &Config) -> Result { operation = "require_live_session_token", "[credentials] app-session token expired locally — publishing SessionExpired before any backend call" ); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::SessionExpired { - source: "credentials.local_expiry_precheck".to_string(), - reason: - "backend session token expired locally — re-authentication required" - .to_string(), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::SessionExpired { + source: "credentials.local_expiry_precheck".to_string(), + reason: "backend session token expired locally — re-authentication required" + .to_string(), + }); } Err( "SESSION_EXPIRED: backend session token expired locally — re-authentication required" diff --git a/src/openhuman/security/devices/bus.rs b/src/openhuman/security/devices/bus.rs index 9132a2338d..0bf8684fb5 100644 --- a/src/openhuman/security/devices/bus.rs +++ b/src/openhuman/security/devices/bus.rs @@ -11,8 +11,6 @@ use std::sync::{Arc, OnceLock}; use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; -use tinybus::SubscriptionHandle; use crate::openhuman::security::devices::crypto::{ base64url_decode, base64url_encode, derive_session_keys, TunnelCipher, TunnelRole, }; @@ -24,6 +22,8 @@ use crate::openhuman::security::devices::tunnel_client::emit_frame; use async_trait::async_trait; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; +use tinybus::EventHandler; +use tinybus::SubscriptionHandle; use x25519_dalek::{PublicKey, StaticSecret}; static DEVICE_TUNNEL_HANDLE: OnceLock = OnceLock::new(); diff --git a/src/openhuman/security/egress/emit_tests.rs b/src/openhuman/security/egress/emit_tests.rs index 79411d3379..9d1963f882 100644 --- a/src/openhuman/security/egress/emit_tests.rs +++ b/src/openhuman/security/egress/emit_tests.rs @@ -25,7 +25,7 @@ async fn find_pending( }) if descriptor.service == marker => return (descriptor, thread_id, client_id), Some(_) => continue, None => panic!("the bus closed before the expected event arrived"), -} + } } } @@ -73,7 +73,7 @@ async fn local_transfer_does_not_publish() { } Some(_) => continue, None => panic!("the bus closed before the expected event arrived"), -} + } } } @@ -139,7 +139,7 @@ async fn dedup_turn_scope_collapses_repeat_destination() { } Some(_) => continue, None => panic!("the bus closed before the expected event arrived"), -} + } } assert_eq!( dup_count, 1, @@ -178,7 +178,7 @@ async fn dedup_absent_outside_scope_publishes_each_time() { } Some(_) => continue, None => panic!("the bus closed before the expected event arrived"), -} + } } assert_eq!( count, 2, diff --git a/src/openhuman/security/keyring_consent/policy.rs b/src/openhuman/security/keyring_consent/policy.rs index 7575d0ce61..a2a2d9a28d 100644 --- a/src/openhuman/security/keyring_consent/policy.rs +++ b/src/openhuman/security/keyring_consent/policy.rs @@ -73,9 +73,8 @@ pub fn check_secret_access() -> PolicyDecision { debug!("{LOG_PREFIX} check_secret_access: keyring unavailable, no consent recorded"); if !CONSENT_EVENT_PUBLISHED.swap(true, Ordering::SeqCst) { info!("{LOG_PREFIX} publishing KeyringConsentRequired event"); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::KeyringConsentRequired, - ); + crate::core::bus::BUS + .publish(crate::core::events::DomainEvent::KeyringConsentRequired); } PolicyDecision::ConsentRequired } @@ -173,21 +172,17 @@ pub fn notify_master_key_unavailable(reason: &str) { warn!("{LOG_PREFIX} master key unavailable: {reason}"); if !CONSENT_EVENT_PUBLISHED.swap(true, Ordering::SeqCst) { info!("{LOG_PREFIX} publishing KeyringConsentRequired event (master key unavailable)"); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::KeyringConsentRequired, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::KeyringConsentRequired); } } /// Publish a decrypt-failure event for frontend notification. pub fn notify_decrypt_failure(field_name: &str, reason: &str) { warn!("{LOG_PREFIX} decrypt failure field={field_name} reason={reason}"); - crate::core::bus::BUS.publish( - crate::core::events::DomainEvent::KeyringDecryptFailed { - field_name: field_name.to_string(), - reason: reason.to_string(), - }, - ); + crate::core::bus::BUS.publish(crate::core::events::DomainEvent::KeyringDecryptFailed { + field_name: field_name.to_string(), + reason: reason.to_string(), + }); } fn classify_failure_reason(backend_name: &str) -> KeyringFailureReason { diff --git a/src/openhuman/skills/bus.rs b/src/openhuman/skills/bus.rs index 2ca4cb1714..dfe58b4699 100644 --- a/src/openhuman/skills/bus.rs +++ b/src/openhuman/skills/bus.rs @@ -13,11 +13,11 @@ use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; -use tinybus::SubscriptionHandle; use crate::openhuman::skills::Workflow; use async_trait::async_trait; use std::sync::{Arc, OnceLock}; +use tinybus::EventHandler; +use tinybus::SubscriptionHandle; // ── Trigger pattern ─────────────────────────────────────────────────────────── diff --git a/src/openhuman/skills/ops_create.rs b/src/openhuman/skills/ops_create.rs index 3b49a0e542..05ab5bc52b 100644 --- a/src/openhuman/skills/ops_create.rs +++ b/src/openhuman/skills/ops_create.rs @@ -691,7 +691,9 @@ mod render_skill_toml_tests { use tinybus::TryRecvError; crate::core::bus::init().await.expect("bus init"); - let mut rx = crate::core::bus::BUS.get().expect("event bus should be initialized") + let mut rx = crate::core::bus::BUS + .get() + .expect("event bus should be initialized") .receiver(); let home = tempfile::TempDir::new().expect("temp home"); diff --git a/src/openhuman/skills/webhooks/bus.rs b/src/openhuman/skills/webhooks/bus.rs index ea173659eb..75a1c33b08 100644 --- a/src/openhuman/skills/webhooks/bus.rs +++ b/src/openhuman/skills/webhooks/bus.rs @@ -7,13 +7,13 @@ use crate::core::bus::BUS; use crate::core::events::DomainEvent; -use tinybus::EventHandler; use crate::openhuman::platform::socket::global_socket_manager; use crate::openhuman::skills::webhooks::WebhookResponseData; use async_trait::async_trait; use serde_json::json; use std::collections::HashMap; use std::time::Instant; +use tinybus::EventHandler; /// Base64-encode a string (for webhook response bodies). fn base64_encode(input: &str) -> String { diff --git a/src/openhuman/voice/bus.rs b/src/openhuman/voice/bus.rs index 6eddbd95be..2262e88b00 100644 --- a/src/openhuman/voice/bus.rs +++ b/src/openhuman/voice/bus.rs @@ -27,10 +27,10 @@ pub fn publish_ptt_transcript_committed( mod tests { use super::*; use crate::core::bus::BUS; -use crate::core::events::DomainEvent; -use tinybus::EventHandler; + use crate::core::events::DomainEvent; use async_trait::async_trait; use std::sync::Arc; + use tinybus::EventHandler; use tokio::sync::Mutex as AsyncMutex; #[derive(Default)] diff --git a/src/openhuman/web_chat/event_bus.rs b/src/openhuman/web_chat/event_bus.rs index 9c5cf984bb..0b77205968 100644 --- a/src/openhuman/web_chat/event_bus.rs +++ b/src/openhuman/web_chat/event_bus.rs @@ -4,9 +4,9 @@ use std::sync::{Arc, OnceLock}; use tokio::sync::broadcast; use crate::core::events::DomainEvent; +use crate::core::socketio::WebChannelEvent; use tinybus::EventHandler; use tinybus::SubscriptionHandle; -use crate::core::socketio::WebChannelEvent; static EVENT_BUS: Lazy> = Lazy::new(|| { let (tx, _rx) = broadcast::channel(512); diff --git a/tests/agent_harness_e2e.rs b/tests/agent_harness_e2e.rs index 770b088266..5e04b20649 100644 --- a/tests/agent_harness_e2e.rs +++ b/tests/agent_harness_e2e.rs @@ -1313,9 +1313,7 @@ async fn ensure_approval_gate() { // `build_core_http_router` does NOT call `bootstrap_core_runtime`, so the // bus is not initialized by boot_stack. Standing it up is async now — it // connects to a broker — which is why this helper is too. Idempotent. - openhuman_core::core::bus::init() - .await - .expect("bus init"); + openhuman_core::core::bus::init().await.expect("bus init"); let mut cfg: openhuman_core::openhuman::config::Config = toml::from_str( r#"api_url = "http://127.0.0.1:1" diff --git a/tests/calendar_grounding_e2e.rs b/tests/calendar_grounding_e2e.rs index 0eb305ef57..22989798a1 100644 --- a/tests/calendar_grounding_e2e.rs +++ b/tests/calendar_grounding_e2e.rs @@ -56,7 +56,7 @@ impl ChatModel<()> for MockCalendarModel { raw: None, resolved_model: None, continue_turn: None, - served_from_cache: false, + served_from_cache: false, }) } else { // End the loop diff --git a/tests/composio_list_tools_stack_overflow_regression.rs b/tests/composio_list_tools_stack_overflow_regression.rs index b52fc1eb33..578ae72a28 100644 --- a/tests/composio_list_tools_stack_overflow_regression.rs +++ b/tests/composio_list_tools_stack_overflow_regression.rs @@ -226,7 +226,7 @@ impl ChatModel<()> for StubModel { raw: None, resolved_model: None, continue_turn: None, - served_from_cache: false, + served_from_cache: false, }) } else { Ok(ModelResponse::assistant("done")) diff --git a/tests/config_auth_app_state_connectivity_e2e.rs b/tests/config_auth_app_state_connectivity_e2e.rs index 73b2575ae3..1724ba556d 100644 --- a/tests/config_auth_app_state_connectivity_e2e.rs +++ b/tests/config_auth_app_state_connectivity_e2e.rs @@ -26,7 +26,6 @@ use openhuman_core::api::config::{ }; use openhuman_core::core::auth::{init_rpc_token, CORE_TOKEN_ENV_VAR}; use openhuman_core::core::events::DomainEvent; -use tinybus::EventHandler; use openhuman_core::core::jsonrpc::build_core_http_router; use openhuman_core::openhuman::config::schema::{ generate_provider_id, generate_voice_provider_id, is_slug_reserved, is_voice_slug_reserved, @@ -69,6 +68,7 @@ use openhuman_core::openhuman::security::credentials::{ list_provider_credentials_by_prefix, normalize_provider, rpc_store_composio_api_key, store_composio_api_key, AuthService, APP_SESSION_PROVIDER, COMPOSIO_DIRECT_PROVIDER, }; +use tinybus::EventHandler; const TEST_RPC_TOKEN: &str = "worker-a-domain-e2e-token"; diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index a24d3b2d07..b8d9065602 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -9548,24 +9548,27 @@ async fn whatsapp_data_agent_tools_e2e_1341() { } // Stand in for the shell store: register canned native handlers. - BUS.native().register::, _, _>( - methods::LIST_CHATS, - |_req| async move { Ok(vec![sample_chat("alice@c.us"), sample_chat("team@g.us")]) }, - ); - BUS.native().register::, _, _>( - methods::LIST_MESSAGES, - |_req| async move { Ok(vec![sample_msg("Send the umbrella report by Friday")]) }, - ); - BUS.native().register::, _, _>( - methods::SEARCH_MESSAGES, - |req| async move { - if req.query.to_lowercase().contains("umbrella") { - Ok(vec![sample_msg("Send the umbrella report by Friday")]) - } else { - Ok(vec![]) - } - }, - ); + BUS.native() + .register::, _, _>( + methods::LIST_CHATS, + |_req| async move { Ok(vec![sample_chat("alice@c.us"), sample_chat("team@g.us")]) }, + ); + BUS.native() + .register::, _, _>( + methods::LIST_MESSAGES, + |_req| async move { Ok(vec![sample_msg("Send the umbrella report by Friday")]) }, + ); + BUS.native() + .register::, _, _>( + methods::SEARCH_MESSAGES, + |req| async move { + if req.query.to_lowercase().contains("umbrella") { + Ok(vec![sample_msg("Send the umbrella report by Friday")]) + } else { + Ok(vec![]) + } + }, + ); fn parse_tool_output(result: openhuman_core::openhuman::skills::types::ToolResult) -> Value { assert!(!result.is_error, "tool returned error: {result:?}"); diff --git a/tests/monitor_agent_e2e.rs b/tests/monitor_agent_e2e.rs index d264817fa0..b5d4a5f15c 100644 --- a/tests/monitor_agent_e2e.rs +++ b/tests/monitor_agent_e2e.rs @@ -207,7 +207,7 @@ fn tool_response(id: &str, name: &str, arguments: serde_json::Value) -> ModelRes raw: None, resolved_model: None, continue_turn: None, - served_from_cache: false, + served_from_cache: false, } } diff --git a/tests/subconscious_conversation_e2e.rs b/tests/subconscious_conversation_e2e.rs index aafcaef4d1..f1b6b7dfda 100644 --- a/tests/subconscious_conversation_e2e.rs +++ b/tests/subconscious_conversation_e2e.rs @@ -237,9 +237,7 @@ struct Harness { impl Harness { async fn new(config: OrchestratorConfig) -> Self { - openhuman_core::core::bus::init() - .await - .expect("bus init"); + openhuman_core::core::bus::init().await.expect("bus init"); let transcript = Transcript::new(); let emit: Emitter = Arc::new(StdMutex::new(VecDeque::new())); let tmp = tempfile::tempdir().expect("tempdir"); @@ -258,20 +256,23 @@ impl Harness { // Capture proactive (subconscious → human) deliveries off the bus. let notifications = Arc::new(StdMutex::new(Vec::::new())); let sink = Arc::clone(¬ifications); - let sub = openhuman_core::core::bus::BUS.get().expect("bus").on("conv-e2e-notify", move |event| { - let sink = Arc::clone(&sink); - let event = event.clone(); - Box::pin(async move { - if let DomainEvent::ProactiveMessageRequested { - source, message, .. - } = &event - { - if source == "subconscious" { - sink.lock().unwrap().push(message.clone()); + let sub = openhuman_core::core::bus::BUS.get().expect("bus").on( + "conv-e2e-notify", + move |event| { + let sink = Arc::clone(&sink); + let event = event.clone(); + Box::pin(async move { + if let DomainEvent::ProactiveMessageRequested { + source, message, .. + } = &event + { + if source == "subconscious" { + sink.lock().unwrap().push(message.clone()); + } } - } - }) - }); + }) + }, + ); let loop_handle = Arc::clone(&orch); let task = tokio::spawn(async move { loop_handle.run_loop().await }); diff --git a/tests/subconscious_fullstack_e2e.rs b/tests/subconscious_fullstack_e2e.rs index 73ef0d431c..e54a71f017 100644 --- a/tests/subconscious_fullstack_e2e.rs +++ b/tests/subconscious_fullstack_e2e.rs @@ -123,7 +123,7 @@ impl MockLlm { raw: None, resolved_model: None, continue_turn: None, - served_from_cache: false, + served_from_cache: false, }; } else { "Mock orchestrator handled the promoted trigger.".to_string() diff --git a/tests/subconscious_triggers_e2e.rs b/tests/subconscious_triggers_e2e.rs index 733e35136e..45729eb7c9 100644 --- a/tests/subconscious_triggers_e2e.rs +++ b/tests/subconscious_triggers_e2e.rs @@ -486,7 +486,9 @@ async fn scenario_notify_user_delivers_and_persists() { let captured: Arc>> = Arc::new(StdMutex::new(Vec::new())); let sink = Arc::clone(&captured); - let _sub = openhuman_core::core::bus::BUS.get().expect("bus initialized") + let _sub = openhuman_core::core::bus::BUS + .get() + .expect("bus initialized") .on("e2e-notify-capture", move |event| { let sink = Arc::clone(&sink); let event = event.clone(); From 643f350fd8d561ce71d6029a4efc1cfc0a0a902d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 9 Aug 2026 11:26:12 +0300 Subject: [PATCH 2/2] refactor(whatsapp_data): flatten ingest handler registration Reformatted the ingest handler registration to use a single closure argument instead of a multi-line tuple, reducing visual nesting and making the code consistent with the other handler registrations in the same function. Auto-committed-on: dragonfly Co-authored-by: Medulla --- app/src-tauri/src/whatsapp_data/mod.rs | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/app/src-tauri/src/whatsapp_data/mod.rs b/app/src-tauri/src/whatsapp_data/mod.rs index af5d437451..cfb21b9dfc 100644 --- a/app/src-tauri/src/whatsapp_data/mod.rs +++ b/app/src-tauri/src/whatsapp_data/mod.rs @@ -87,15 +87,12 @@ pub fn register_native_handlers() { ops::search_messages(&store, req).map_err(|e| format!("{e:#}")) }, ); - native.register::( - methods::INGEST, - |req| async move { - let store = ensure_store().await?; - // `{e:#}` renders the full anyhow chain so the underlying SQLite - // cause (locked / malformed / FK) survives to the scanner's log. - ops::ingest(&store, req).map_err(|e| format!("[whatsapp_data] ingest failed: {e:#}")) - }, - ); + native.register::(methods::INGEST, |req| async move { + let store = ensure_store().await?; + // `{e:#}` renders the full anyhow chain so the underlying SQLite + // cause (locked / malformed / FK) survives to the scanner's log. + ops::ingest(&store, req).map_err(|e| format!("[whatsapp_data] ingest failed: {e:#}")) + }); log::info!( "[whatsapp_data] registered shell native handlers (list_chats / list_messages / search_messages / ingest)" );