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..cfb21b9dfc 100644
--- a/app/src-tauri/src/whatsapp_data/mod.rs
+++ b/app/src-tauri/src/whatsapp_data/mod.rs
@@ -60,38 +60,39 @@ pub async fn ensure_store() -> Result<Arc<WhatsAppDataStore>, 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::<ListChatsRequest, Vec<WhatsAppChat>, _, _>(
+    native.register::<ListChatsRequest, Vec<WhatsAppChat>, _, _>(
         methods::LIST_CHATS,
         |req| async move {
             let store = ensure_store().await?;
             ops::list_chats(&store, req).map_err(|e| format!("{e:#}"))
         },
     );
-    register_native_global::<ListMessagesRequest, Vec<WhatsAppMessage>, _, _>(
+    native.register::<ListMessagesRequest, Vec<WhatsAppMessage>, _, _>(
         methods::LIST_MESSAGES,
         |req| async move {
             let store = ensure_store().await?;
             ops::list_messages(&store, req).map_err(|e| format!("{e:#}"))
         },
     );
-    register_native_global::<SearchMessagesRequest, Vec<WhatsAppMessage>, _, _>(
+    native.register::<SearchMessagesRequest, Vec<WhatsAppMessage>, _, _>(
         methods::SEARCH_MESSAGES,
         |req| async move {
             let store = ensure_store().await?;
             ops::search_messages(&store, req).map_err(|e| format!("{e:#}"))
         },
     );
-    register_native_global::<IngestRequest, IngestResult, _, _>(
-        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::<IngestRequest, IngestResult, _, _>(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)"
     );
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<Result<Event, std::convert::Infallible>> {
-            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<Result<Event, std::convert::Infallible>> {
+        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::<AgentTurnRequest, AgentTurnResponse, _, _>(
-        AGENT_RUN_TURN_METHOD,
-        |req| {
+    BUS.native()
+        .register::<AgentTurnRequest, AgentTurnResponse, _, _>(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<F, Fut>(
-    handler: F,
-) -> crate::core::bus_testing::MockBusGuard
+pub async fn mock_agent_run_turn<F, Fut>(handler: F) -> crate::core::bus_testing::MockBusGuard
 where
     F: Fn(AgentTurnRequest) -> Fut + Send + Sync + 'static,
     Fut: std::future::Future<Output = Result<AgentTurnResponse, String>> + 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::<AgentTurnRequest, AgentTurnResponse, F, Fut, _>(
+        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::<DomainEvent>::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<dyn ChatModel<()>> = 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<Option<SubscriptionHandle>> = 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::<AgentTurnRequest, AgentTurnResponse>(
-        AGENT_RUN_TURN_METHOD,
-        request,
-    )
-    .await
+    let response = match BUS
+        .native()
+        .request::<AgentTurnRequest, AgentTurnResponse>(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::<DomainEvent>::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<Arc<crate::openhuman::security::approval::ApprovalGate>> {
@@ -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::<AgentTurnRequest, AgentTurnResponse>(
-            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::<AgentTurnRequest, AgentTurnResponse>(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::<AgentTurnRequest, AgentTurnResponse, _, _>(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::<AgentTurnRequest, AgentTurnResponse, _, _>(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::<Vec<_>>()
-                    .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::<Vec<_>>()
+                        .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<DomainEvent> 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<RpcOutcome<serd
         .map_err(|e| format!("[composio-direct] save config failed: {e}"))?;
     direct_auth::reset_all_direct_auth_failures();
 
-    crate::core::bus::BUS.publish(
-        crate::core::events::DomainEvent::ComposioConfigChanged {
-            mode: "backend".to_string(),
-            api_key_set: false,
-        },
-    );
+    crate::core::bus::BUS.publish(crate::core::events::DomainEvent::ComposioConfigChanged {
+        mode: "backend".to_string(),
+        api_key_set: false,
+    });
     tracing::debug!("[composio-cache] published ComposioConfigChanged after clear_api_key");
 
     Ok(RpcOutcome::new(
diff --git a/src/openhuman/integrations/task_sources/bus.rs b/src/openhuman/integrations/task_sources/bus.rs
index 80f117aa68..e6c4cc1321 100644
--- a/src/openhuman/integrations/task_sources/bus.rs
+++ b/src/openhuman/integrations/task_sources/bus.rs
@@ -11,9 +11,9 @@ use async_trait::async_trait;
 
 use crate::core::bus::BUS;
 use crate::core::events::DomainEvent;
+use crate::openhuman::config::rpc as config_rpc;
 use tinybus::EventHandler;
 use tinybus::SubscriptionHandle;
-use crate::openhuman::config::rpc as config_rpc;
 
 use super::types::{FetchReason, ProviderSlug};
 use super::{pipeline, store};
diff --git a/src/openhuman/meet/backend_bot/calendar.rs b/src/openhuman/meet/backend_bot/calendar.rs
index 683ddeaa99..a0bf7c5bbf 100644
--- a/src/openhuman/meet/backend_bot/calendar.rs
+++ b/src/openhuman/meet/backend_bot/calendar.rs
@@ -24,14 +24,14 @@ use async_trait::async_trait;
 
 use crate::core::bus::BUS;
 use crate::core::events::DomainEvent;
-use tinybus::EventHandler;
-use tinybus::SubscriptionHandle;
 use crate::openhuman::config::rpc as config_rpc;
 use crate::openhuman::desktop::app_state::peek_cached_current_user_identity;
 use crate::openhuman::desktop::notifications::bus::publish_core_notification;
 use crate::openhuman::desktop::notifications::types::{
     CoreNotificationAction, CoreNotificationCategory, CoreNotificationEvent,
 };
+use tinybus::EventHandler;
+use tinybus::SubscriptionHandle;
 
 use super::store;
 use super::types::{AutoJoinSource, MeetingSession, MeetingSessionStatus};
diff --git a/src/openhuman/memory/binding.rs b/src/openhuman/memory/binding.rs
index 474631e004..74d64429c5 100644
--- a/src/openhuman/memory/binding.rs
+++ b/src/openhuman/memory/binding.rs
@@ -395,8 +395,8 @@ fn build(workspace_dir: &Path, cfg: &MemorySubsystemConfig) -> 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<Vec<String>>) -> 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<Option<MemoryClientRef>, 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<MemoryClientRef, String> {
+fn cache_client(workspace_dir: &Path, client: &MemoryClientRef) -> Result<MemoryClientRef, String> {
     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<Vec<(String, String, String)>>,
+}
+
+#[async_trait::async_trait]
+impl tinybus::EventHandler<DomainEvent> 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<DeniedRecorder>, 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<RpcOutcome<SyncChannelResult>, 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<RpcOutcome<SyncAllResult>, 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<std::sync::Mutex<()>> = 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<DomainEvent> 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<String, String> {
                     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<SubscriptionHandle> = 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<broadcast::Sender<WebChannelEvent>> = 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::<ListChatsRequest, Vec<WhatsAppChat>, _, _>(
-        methods::LIST_CHATS,
-        |_req| async move { Ok(vec![sample_chat("alice@c.us"), sample_chat("team@g.us")]) },
-    );
-    BUS.native().register::<ListMessagesRequest, Vec<WhatsAppMessage>, _, _>(
-        methods::LIST_MESSAGES,
-        |_req| async move { Ok(vec![sample_msg("Send the umbrella report by Friday")]) },
-    );
-    BUS.native().register::<SearchMessagesRequest, Vec<WhatsAppMessage>, _, _>(
-        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::<ListChatsRequest, Vec<WhatsAppChat>, _, _>(
+            methods::LIST_CHATS,
+            |_req| async move { Ok(vec![sample_chat("alice@c.us"), sample_chat("team@g.us")]) },
+        );
+    BUS.native()
+        .register::<ListMessagesRequest, Vec<WhatsAppMessage>, _, _>(
+            methods::LIST_MESSAGES,
+            |_req| async move { Ok(vec![sample_msg("Send the umbrella report by Friday")]) },
+        );
+    BUS.native()
+        .register::<SearchMessagesRequest, Vec<WhatsAppMessage>, _, _>(
+            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::<String>::new()));
         let sink = Arc::clone(&notifications);
-        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<StdMutex<Vec<DomainEvent>>> = 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();