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

> Discover how OpenHuman's event bus implements typed pub/sub across core domains using a thread-safe singleton for seamless communication. Learn about zero-cost request/response.

- Repository: [Tiny Humans/openhuman](https://github.com/tinyhumansai/openhuman)
- Tags: internals
- Published: 2026-08-24

---

**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`](https://github.com/tinyhumansai/openhuman/blob/main/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`](https://github.com/tinyhumansai/openhuman/blob/main/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:

```rust
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:

```rust
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:

```rust
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`](https://github.com/tinyhumansai/openhuman/blob/main/src/core/bus_testing.rs) to create fresh `OnceBus` instances:

```rust
#[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`](https://github.com/tinyhumansai/openhuman/blob/main/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`](https://github.com/tinyhumansai/openhuman/blob/main/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`](https://github.com/tinyhumansai/openhuman/blob/main/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`](https://github.com/tinyhumansai/openhuman/blob/main/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`](https://github.com/tinyhumansai/openhuman/blob/main/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.