How the OpenHuman Event Bus Enables Decoupled Communication Between Modules

The OpenHuman event bus provides a global publish-subscribe layer that allows Rust modules to communicate through broadcast events and typed native requests without direct dependencies, keeping the core business logic independent from UI and service implementations.

The tinyhumansai/openhuman repository follows a layered architecture where business-logic domains live in Rust (the openhuman crate) and the UI runs in TypeScript/React. To maintain strict separation between these domains, the core provides a global event bus found in crate::core::event_bus. This system eliminates tight coupling by offering two complementary communication surfaces: fire-and-forget broadcasts for notifications and zero-serialization native requests for synchronous operations requiring immediate results.

Architecture Overview: Broadcast Events vs. Native Requests

The OpenHuman event bus exposes two distinct APIs that serve different communication patterns.

Broadcast events — accessed via publish_global() and subscribe_global() — enable fire-and-forget notifications. Any module can emit events without knowing which subscribers exist, and any number of handlers can listen without knowing the event source. No return value is expected. For example, when a cron job finishes, it publishes DomainEvent::CronJobCompleted from src/openhuman/cron/scheduler.rs, and the notifications subscriber in src/openhuman/notifications/bus.rs reacts by showing a UI toast.

Native request/response — accessed via register_native_global() and request_native_global() — provides typed, zero-serialization calls between modules that need direct answers. When the memory subsystem needs to execute a skill, it sends a native request to the skill runtime and receives the output without JSON-RPC overhead, as implemented in src/openhuman/skills/bus.rs.

Defining Domain Events in events.rs

All possible events are enumerated in src/openhuman/core/event_bus/events.rs. Each variant follows a domain prefix convention (e.g., CronJobCompleted, MemoryIngestionStarted) to categorize events by subsystem. Adding a new event requires only extending this enum and optionally implementing a matching domain() method.

This centralized definition ensures type safety across the entire application. The enum serves as the single source of truth for what events can flow through the system, making it trivial to audit communication patterns and prevent naming collisions.

Publishing Events with publish_global

Any module can emit an event using the publish_global helper. Typical publishers include the memory ingestion logic, cron scheduler, and webhooks router.

From src/openhuman/memory/store/client.rs, the system publishes when ingestion begins:

use crate::core::event_bus::{publish_global, DomainEvent};

pub fn notify_ingestion_started(source: &str) {
    publish_global(DomainEvent::MemoryIngestionStarted {
        source: source.to_string(),
        description: "User requested recall".into(),
    });
}

Similarly, the cron scheduler in src/openhuman/cron/scheduler.rs emits completion signals when jobs finish execution. These calls are non-blocking; the publisher continues execution immediately without waiting for subscribers.

Subscribing with the EventHandler Trait

To receive events, a module implements the EventHandler trait defined in the bus system. This async trait requires three methods: name() for identification, domains() to filter which event categories to receive, and handle() to process the event.

The notifications bridge in src/openhuman/notifications/bus.rs demonstrates this pattern:

use crate::core::event_bus::{DomainEvent, EventHandler};
use async_trait::async_trait;

pub struct NotificationBridgeSubscriber;

#[async_trait]
impl EventHandler for NotificationBridgeSubscriber {
    fn name(&self) -> &str { "notifications::bridge" }
    
    fn domains(&self) -> Option<&[&str]> { 
        Some(&["cron", "memory"]) 
    }

    async fn handle(&self, event: &DomainEvent) {
        if let DomainEvent::CronJobCompleted { job_id, success } = event {
            // Forward to UI via WebSocket or log to monitoring
        }
    }
}

Subscribers register at startup via subscribe_global(Arc::new(...)). The service layer registers core-level handlers for restart, shutdown, and health checks in src/openhuman/service/bus.rs. This registration pattern allows the system to swap subscribers for testing or disable specific handlers without modifying publisher code.

Native Request/Response for Synchronous Operations

When modules require concrete results rather than notifications, the bus provides a typed request pathway that bypasses serialization. Because request and response types implement Send + 'static, the call occurs as direct intra-process communication.

Registering a handler in src/openhuman/skills/bus.rs looks like this:

use crate::core::event_bus::{register_native_global, request_native_global};

#[derive(Debug)]
pub struct SkillRequest {
    pub name: String,
    pub args: Vec<String>,
}

#[derive(Debug)]
pub struct SkillResult {
    pub output: String,
}

pub async fn register_skill_handler() {
    register_native_global::<SkillRequest, SkillResult, _, _>(
        "skill.execute",
        |req| async move {
            // Execute JavaScript in QuickJS engine or perform computation
            Ok(SkillResult {
                output: format!("Executed {} with {:?}", req.name, req.args),
            })
        },
    ).await;
}

The caller elsewhere in the codebase awaits a concrete response:

pub async fn execute_translation() -> anyhow::Result<SkillResult> {
    request_native_global(
        "skill.execute",
        SkillRequest {
            name: "translate".into(),
            args: vec!["en".into(), "fr".into(), "hello".into()],
        },
    ).await
}

This mechanism provides the performance of direct function calls while maintaining loose coupling, as the caller only knows the request type and channel name, not the specific implementation module.

Initializing the Global Bus

Before any communication occurs, the process initializes the singleton bus via init_global in src/openhuman/core/event_bus/mod.rs:

let bus = event_bus::init_global(event_bus::DEFAULT_CAPACITY);

This function creates a broadcast::Sender for broadcast events and a NativeRegistry for typed requests. The returned handle is typically held only for diagnostics; subsequent access occurs through the module-level functions like publish_global and request_native_global. Initialization happens exactly once per process, ensuring a single consistent communication fabric throughout the application lifecycle.

Summary

• The OpenHuman event bus acts as the central nervous system of the application, located in crate::core::event_bus, providing both broadcast and native request surfaces.
• Broadcast events via publish_global allow fire-and-forget communication from publishers like src/openhuman/cron/scheduler.rs to subscribers like src/openhuman/notifications/bus.rs.
• Native requests via request_native_global enable typed, zero-serialization synchronous calls between modules such as memory and skills.
• The EventHandler trait in src/openhuman/core/event_bus/subscriber.rs defines the subscription interface, supporting domain-specific filtering.
• All domain events are type-safe variants of the DomainEvent enum defined in src/openhuman/core/event_bus/events.rs.
• Initialization occurs once through event_bus::init_global in src/openhuman/core/event_bus/mod.rs, establishing global state accessible throughout the Rust core.

Frequently Asked Questions

How does the OpenHuman event bus differ from standard Rust message passing?

The OpenHuman event bus provides a global, process-wide communication layer that abstracts Tokio channels behind a type-safe API. While standard Rust message passing requires direct channel handle sharing, the bus allows modules to communicate by event type alone, with subscribers registering interest in specific domains rather than receiving all messages.

Can multiple subscribers handle the same broadcast event?

Yes, the broadcast nature of publish_global delivers events to all active subscribers that match the event domain. The underlying implementation uses a broadcast::Sender, cloning the message to each receiver that hasn't lagged behind, enabling truly decoupled multi-consumer patterns where notifications, logging, and metrics can all react to the same event independently.

What happens if a native request handler is not registered when request_native_global is called?

The request_native_global function returns a Result, and calling an unregistered channel name will return an error indicating the handler does not exist. This fail-fast behavior prevents silent failures in synchronous request paths and encourages explicit registration of all capability providers during system startup.

Is the OpenHuman event bus thread-safe for concurrent access?

Yes, the bus is designed for concurrent access across asynchronous tasks. The publish_global, subscribe_global, and request_native_global functions are thread-safe and can be called from any Tokio runtime thread. The underlying NativeRegistry and broadcast sender use appropriate synchronization primitives to ensure safe concurrent reads and writes without external locking.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}