How the OpenHuman Event Bus Implements Typed Pub/Sub Across Core Domains

OpenHuman's core architecture uses a process-wide typed event bus built on the tinybus crate, exposing a static OnceBus<DomainEvent> singleton that enables thread-safe publishing and subscription across memory, flows, security, and other domains with zero-cost native request/response capabilities.

The tinyhumansai/openhuman repository employs a sophisticated event-driven architecture that allows disparate core domains to communicate without tight coupling. At the heart of this system lies a typed pub/sub event bus that leverages Rust's type system to guarantee compile-time safety for cross-domain messages. This implementation uses the tinybus crate to provide both asynchronous event distribution and synchronous native method calls across the entire process.

The Process-Wide Bus Architecture

The event bus operates as a single process-wide singleton defined in src/core/bus.rs. It uses a OnceBus<DomainEvent> named BUS that guarantees idempotent initialization while remaining accessible from any thread context.

Static Singleton Initialization

The global bus is created once by core::bus::init() (or init_over_socket for out-of-process integrations). The OnceBus wrapper ensures that initialization happens exactly once, even when multiple domains attempt concurrent access during startup. Because the bus is wrapped in a static singleton, any code can call BUS.publish or BUS.subscribe safely before the bus is ready—these calls become no-ops or return None rather than panicking, with trace-level logging indicating the uninitialized state.

Thread-Safe Access Patterns

The BUS static provides thread-safe access through interior mutability patterns inherent to the tinybus implementation. This allows domains running on different async runtimes or threads to publish events without requiring external synchronization primitives.

Typed Event Catalog with DomainEvent

All events flowing through the system are defined in the centralized DomainEvent enum located in src/core/events.rs. This catalog provides the type safety guarantees that distinguish the OpenHuman implementation from string-based message buses.

Centralized Event Definitions

The DomainEvent enum contains variants for every significant state change across the system, such as SystemStartup, MemorySyncRequested, and FlowRunStarted. Each variant carries a concrete payload rather than a generic map, enabling pattern matching with exhaustive compile-time checks:

use crate::core::bus::BUS;
use crate::core::events::DomainEvent;

// A memory sync component announces that a sync is requested
BUS.publish(DomainEvent::MemorySyncRequested { channel_id: None });

Versioning and Interface Contracts

The event catalog is versioned via EVENTS_VERSION and advertised on the bus using a fixed object-path /ai/tinyhumans/openhuman/events and interface name ai.tinyhumans.openhuman.Events. These identifiers allow external processes to discover the event schema when using out-of-process integration modes.

Publishing Events Across Domains

Code that needs to announce state changes uses the fire-and-forget BUS.publish(event) method. This operation returns immediately without blocking the publisher, making it suitable for high-frequency operations within hot paths. The publish method includes safety checks for early-startup scenarios: if the bus hasn't finished initialization, the call logs at trace level and returns silently rather than failing.

Subscribing to Typed Events

Consumers receive events by implementing the tinybus::EventHandler<DomainEvent> trait and registering an Arc-wrapped instance with BUS.subscribe(handler). This typed subscription mechanism ensures that handlers receive the full DomainEvent value, enabling exhaustive pattern matching against variants:

use tinybus::EventHandler;
use crate::core::bus::BUS;
use crate::core::events::DomainEvent;
use std::sync::Arc;

#[derive(Debug)]
struct MyLogger;

#[async_trait::async_trait]
impl EventHandler<DomainEvent> for MyLogger {
    fn name(&self) -> &str { "my_logger" }
    
    async fn handle(&self, ev: &DomainEvent) {
        tracing::debug!(event = ?ev, "received event");
    }
}

// Register the handler
let _handle = BUS.subscribe(Arc::new(MyLogger));

The subscription returns a handle that controls the lifecycle of the listener; dropping the handle automatically unsubscribes the handler.

Native Request/Response for Complex Data

When domains need to exchange complex data structures—such as Arc references, trait objects, or channels—they bypass the serialization layer using BUS.native().request(method, request).await. This API stays entirely within the process, preserving Rust's zero-cost abstractions while maintaining the pub/sub architectural pattern:

let resp = BUS.native()
    .request("agent.run_turn", request_payload)
    .await?;

This approach avoids JSON or protobuf serialization overhead when communicating between trusted in-process domains, while still using the same bus infrastructure for discovery and routing.

Testing Isolation Strategies

Unit tests that verify event-driven behavior require isolation from the global singleton to prevent interference with other tests or the running system. The framework provides the isolated_bus() helper in src/core/bus_testing.rs to create fresh OnceBus instances:

#[tokio::test]
async fn isolated_bus_delivers() {
    let bus = crate::core::bus_testing::isolated_bus().await;
    // Attach a handler and publish as usual within this isolated scope
}

This pattern ensures that tests can verify subscription logic, event ordering, and handler behavior without side effects on the process-wide event bus.

Summary

• Static singleton: The BUS static in src/core/bus.rs provides a process-wide OnceBus<DomainEvent> with idempotent initialization via init() or init_over_socket().
• Type safety: All events are defined in the DomainEvent enum in src/core/events.rs, enabling compile-time guarantees and exhaustive pattern matching.
• Fire-and-forget publishing: Use BUS.publish(event) for asynchronous, thread-safe announcement of state changes across domains.
• Typed subscriptions: Implement EventHandler<DomainEvent> and register with BUS.subscribe() to receive type-safe event notifications.
• Zero-cost requests: Use BUS.native().request() for in-process method calls that bypass serialization when exchanging complex data.
• Test isolation: The isolated_bus() helper in src/core/bus_testing.rs creates private bus instances for deterministic unit testing.

Frequently Asked Questions

What crate does OpenHuman use for its event bus implementation?

OpenHuman uses the tinybus crate as the underlying implementation. This crate provides the OnceBus primitive, the EventHandler trait, and the native request/response API that powers the system's domain communication.

How does the bus handle type safety across different domains?

The system centralizes all possible events in the DomainEvent enum defined in src/core/events.rs. Because subscribers implement EventHandler<DomainEvent> rather than handling raw strings or bytes, the Rust compiler enforces that handlers account for all event variants, preventing runtime deserializations errors and ensuring type-safe access to event payloads.

Can domains communicate complex data without serialization overhead?

Yes. When domains need to exchange Arc references, trait objects, or channels, they use BUS.native().request(method, request).await. This API stays within the process and bypasses the serialization layer entirely, providing zero-cost abstraction while maintaining the architectural decoupling of the pub/sub pattern.

How do you prevent unit tests from interfering with the global event bus?

The testing framework includes an isolated_bus() helper function in src/core/bus_testing.rs that creates a fresh OnceBus<DomainEvent> instance. Tests use this isolated bus instead of the global BUS static, ensuring that event handlers, subscriptions, and publications remain scoped to the individual test case without affecting the global system state.

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