OpenHuman Event Bus Architecture: Typed Pub/Sub and Native Request/Response Patterns

OpenHuman implements a zero-allocation, type-safe event bus that enables loosely coupled communication between domains via compile-time checked pub/sub events and synchronous native request/response patterns.

The OpenHuman framework relies on a lightweight messaging layer to coordinate behavior across its core domains, including agent orchestration, memory management, and skill execution. At the heart of this system lies a Rust-based event bus architecture in OpenHuman that provides strongly typed publish-subscribe messaging alongside native request-response handling. This design eliminates runtime coupling while maintaining compile-time safety through Rust's type system.

Core Components of the Event Bus

The event bus implementation resides entirely within src/core/event_bus/ and centers on five primary abstractions that handle event distribution and type-safe messaging.

DomainEvent Enum

The DomainEvent enum (defined in src/core/event_bus/events.rs) enumerates every event that domains can emit, such as MemoryChunkAdded or ChannelMessageReceived. This central registry ensures that all publishable events are known at compile time, preventing invalid event types from propagating through the system.

EventBus and Global Publishing

The EventBus struct (in src/core/event_bus/bus.rs) maintains a registry of global subscribers and dispatches events to them. The publish_global() function serves as the primary entry point, offering a thin wrapper that forwards DomainEvent instances to the global singleton instance.

NativeRegistry for Request/Response

For synchronous operations, the NativeRegistry (located in src/core/event_bus/native_request.rs) maps string-based verbs like "memory.get_chunk" to strongly typed request-response pairs. Each registration binds a verb to specific request and response types, enabling the bus to enforce type safety at the API boundary.

SubscriptionHandle

The SubscriptionHandle type (from src/core/event_bus/subscriber.rs) manages subscriber lifecycles. When code calls subscribe_global(), it returns a handle that automatically unsubscribes the closure when dropped, preventing memory leaks and dangling callbacks.

Typed Pub/Sub Pattern

The publish-subscribe model allows domains to broadcast events without knowledge of subscribers. This decoupled approach uses Rust's enum variants to guarantee that subscribers handle every possible event type exhaustively.

Publishing a typed event requires calling publish_global() with a concrete DomainEvent variant:

use openhuman_core::event_bus::{publish_global, DomainEvent};

fn add_memory_chunk(chunk_id: u64) {
    // …logic to store the chunk…
    publish_global(DomainEvent::MemoryChunkAdded { chunk_id });
}

Subscribers receive events through closures that match on the enum. The subscribe_global() function returns a SubscriptionHandle that keeps the subscription alive:

use openhuman_core::event_bus::{subscribe_global, DomainEvent};

let _handle = subscribe_global(|ev| {
    if let DomainEvent::MemoryChunkAdded { chunk_id } = ev {
        println!("New memory chunk: {chunk_id}");
    }
});

Native Request/Response Pattern

While pub/sub handles notifications, the native request/response pattern facilitates synchronous, one-to-one communication between domains. This mechanism relies on the NativeRegistry to route requests based on string verbs while preserving type safety through generic parameters.

Making a native request involves specifying both request and response types alongside the verb:

use openhuman_core::event_bus::request_native_global;

let req = GetChunkReq { id: 42 };
let resp = request_native_global::<GetChunkReq, GetChunkResp>("memory.get_chunk", req)?;
println!("Chunk data: {:?}", resp.chunk);

Handlers are registered during startup, typically in src/core/all.rs, using register_native_global():

use openhuman_core::event_bus::{register_native_global, NativeHandler};

fn handle_get_chunk(req: GetChunkReq) -> GetChunkResp {
    // …fetch the chunk…
    GetChunkResp { chunk: /* data */ }
}

register_native_global(
    "memory.get_chunk",
    NativeHandler::new(handle_get_chunk),
);

The type system ensures that mismatched request or response types fail at compile time rather than runtime.

Domain Registration and Startup Flow

During system initialization, each domain registers its events and native verbs in src/core/all.rs. This centralized registration ensures that the memory domain's "memory.get_chunk" verb and DomainEvent::MemoryChunkAdded variant are available before any domain begins processing. The startup sequence establishes the global singleton instance that backs both publish_global() and request_native_global(), creating a cohesive communication layer that spans agent, memory, channels, cron, and skills domains.

Summary

• Type-safe by design: The DomainEvent enum and generic native request functions leverage Rust's type system to catch mismatches at compile time.
• Zero-allocation messaging: The event bus operates as an in-process communication layer without heap allocations for event distribution.
• Dual patterns: Supports both asynchronous pub/sub via publish_global() and synchronous request/response via request_native_global().
• Automatic cleanup: SubscriptionHandle ensures subscribers are removed when handles drop, preventing resource leaks.
• Centralized registration: Domain setup in src/core/all.rs coordinates event and verb availability across the entire system.

Frequently Asked Questions

What makes OpenHuman's event bus type-safe?

The architecture uses Rust's strongly typed enums for events and generic type parameters for native requests. When publishing, you pass a concrete DomainEvent variant, and when requesting, you specify request and response types at the call site. This ensures the compiler rejects any attempt to send the wrong data type or handle a non-existent event variant.

How does the native request/response pattern differ from standard pub/sub?

Pub/sub uses publish_global() to broadcast events to zero or more subscribers asynchronously, while native request/response uses request_native_global() for synchronous, one-to-one calls that return typed responses. The native pattern blocks until the handler returns, making it suitable for queries that require immediate results, whereas pub/sub suits notifications and event streaming.

Where are event handlers registered in the OpenHuman codebase?

Domain-specific handlers are registered during startup in src/core/all.rs. This file coordinates the initialization sequence, calling register_native_global() for request handlers and setting up the DomainEvent variants that each domain can emit. The EventBus singleton in src/core/event_bus/bus.rs maintains the actual subscriber lists at runtime.

Can subscribers accidentally miss events due to type mismatches?

No. Because subscribers receive the full DomainEvent enum and match on specific variants, Rust's pattern matching exhaustiveness checker ensures all variants are handled. Additionally, the DomainEvent definition in src/core/event_bus/events.rs acts as a single source of truth, preventing domains from emitting undeclared event types.

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