How the Event Bus in OpenHuman Enables Inter-Domain Communication via DomainEvent Variants

The event bus in OpenHuman facilitates decoupled inter-domain communication through a lazily-initialized static BUS instance and a unified DomainEvent enum that serves as the canonical type-safe contract for all cross-domain messages.

The OpenHuman framework partitions functionality into independent domains—including memory, threads, voice, skills, and web-chat—that must react to shared state changes without direct coupling. To solve this architectural challenge, the project implements a global publish/subscribe event bus centered on the DomainEvent enum, allowing any subsystem to emit or observe lifecycle events asynchronously.

Centralized Bus Architecture in src/core/bus.rs

The communication backbone resides in src/core/bus.rs, which exposes a lazily-initialized static BUS instance accessible throughout the codebase. This global bus provides two primary APIs: BUS.publish(event) for emitting messages and BUS.subscribe(handler) for registering listeners. By routing all messages through this single static instance, OpenHuman eliminates circular dependencies between domains while maintaining a consistent asynchronous messaging interface.

The DomainEvent Enum as a Canonical Contract

All cross-domain traffic flows through the DomainEvent enum defined in src/core/events.rs. This enum aggregates every possible inter-domain occurrence into distinct variants such as DomainEvent::ApprovalRequested, DomainEvent::Voice(VoiceEvent::…), DomainEvent::ThreadGoalUpdated, and DomainEvent::WebhookReceived. Because every domain imports this single type from crate::core::events::DomainEvent, the enum serves as the compile-time contract that guarantees type safety across subsystem boundaries.

Domain-specific events are typically namespaced using inner enums. For example, voice-related events wrap a VoiceEvent enum inside the DomainEvent::Voice variant, keeping the top-level namespace organized while allowing fine-grained pattern matching.

Publishing Events to the Global Bus

Any subsystem can broadcast state changes by constructing the appropriate DomainEvent variant and calling crate::core::bus::BUS.publish(event). The bus immediately forwards the event to all registered handlers without blocking the publisher.

For instance, when reclaiming task runs, src/openhuman/threads/todos/runs.rs publishes DomainEvent::TaskRunReclaimed. Similarly, upon receiving external HTTP callbacks, src/openhuman/skills/webhooks/bus.rs emits DomainEvent::WebhookIncomingRequest. This pattern allows upstream domains to remain ignorant of which downstream systems, if any, are listening.

use crate::core::bus::BUS;
use crate::core::events::DomainEvent;
use crate::openhuman::voice::VoiceEvent;

// Broadcasting a voice transcript completion
BUS.publish(DomainEvent::Voice(
    VoiceEvent::PttTranscriptCommitted {
        transcript_id: "abc123".into(),
        // additional fields...
    },
));

Subscribing with the EventHandler Trait

Domains consume events by implementing the generic EventHandler<DomainEvent> trait defined in src/core/events.rs. Implementations provide an async handle(&self, event: &DomainEvent) method and use pattern matching to act only on relevant variants.

Voice Domain Implementation

In src/openhuman/voice/bus.rs, the Capture struct implements EventHandler<DomainEvent> to react specifically to DomainEvent::Voice(v) events. This allows the voice subsystem to process microphone state changes without importing types from other domains.

Skills Domain Implementation

Similarly, src/openhuman/skills/bus.rs defines TriggeredSkillSubscriber, which listens for workflow-related variants such as DomainEvent::WorkflowTriggered.

The following example demonstrates a subscriber that reacts to workflow triggers:

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

pub struct TriggeredSkillSubscriber;

#[async_trait]
impl EventHandler<DomainEvent> for TriggeredSkillSubscriber {
    async fn handle(&self, event: &DomainEvent) {
        if let DomainEvent::WorkflowTriggered { workflow_id, .. } = event {
            // Execute the associated skill logic
            run_skill(workflow_id).await;
        }
    }
}

// Registration typically occurs during core startup
crate::core::bus::BUS.subscribe(Box::new(TriggeredSkillSubscriber));

Selective Event Handling

Subscribers can filter for specific state changes by matching on variant fields. A thread-monitoring component might implement:

use crate::core::events::{DomainEvent, EventHandler};

pub struct ThreadGoalSubscriber;

#[async_trait::async_trait]
impl EventHandler<DomainEvent> for ThreadGoalSubscriber {
    async fn handle(&self, event: &DomainEvent) {
        if let DomainEvent::ThreadGoalUpdated { thread_id, goal } = event {
            log::info!("Thread {} goal updated: {}", thread_id, goal);
        }
    }
}

Subscriber Registration and Startup

During application initialization, the system aggregates all domain-specific handlers through the register_subscribers logic in src/core/all.rs. This central registration point ensures the global BUS contains every active listener before the system begins processing user requests. Because registration occurs at startup, domains cannot dynamically subscribe at runtime, which simplifies lifecycle management and prevents late-binding errors.

Decoupling Benefits and Cross-Domain Features

The event bus architecture enables sophisticated cross-domain workflows while preserving strict separation of concerns. The approval gate converts DomainEvent::ApprovalRequested into server-sent events (SSE) for web clients, while the web-chat component in src/openhuman/web_chat/event_bus.rs transforms DomainEvent::RunQueueMessageQueued into UI updates.

Domains never invoke each other directly; they only need knowledge of the DomainEvent enum structure. Consequently, adding a new cross-domain signal requires only:

1. Adding a new variant to DomainEvent in src/core/events.rs
2. Optionally implementing a new EventHandler to react to that variant
3. Publishing the new variant from the source domain

Existing publishers and subscribers remain unchanged, demonstrating true loose coupling.

Summary

• The static BUS in src/core/bus.rs provides a global publish/subscribe API for all OpenHuman domains without direct coupling.
• The DomainEvent enum in src/core/events.rs acts as the single, type-safe contract for inter-domain messages through its distinct variants.
• Domains publish events via BUS.publish() and subscribe by implementing the EventHandler<DomainEvent> trait with async handle methods.
• Concrete implementations in src/openhuman/voice/bus.rs and src/openhuman/skills/bus.rs demonstrate pattern-matching on specific variants like DomainEvent::Voice and DomainEvent::WorkflowTriggered.
• Centralized registration in src/core/all.rs ensures all handlers are active at application startup, enabling reliable event propagation across memory, threads, voice, and skill subsystems.

Frequently Asked Questions

How do I publish an event from a custom domain in OpenHuman?

Construct the appropriate DomainEvent variant and call crate::core::bus::BUS.publish(event). For example, components in the voice domain publish DomainEvent::Voice(VoiceEvent::PttTranscriptCommitted { transcript_id: "xyz".into() }) to notify other systems of transcript completions, as seen in src/openhuman/voice/bus.rs.

What is the EventHandler trait and where is it defined?

The EventHandler<DomainEvent> trait is defined in src/core/events.rs. It requires implementing an async handle(&self, event: &DomainEvent) method that receives a reference to the enum. Implementations typically use if let or match statements to filter for specific variants relevant to their domain logic.

Can a single subscriber listen to multiple DomainEvent variants?

Yes. Since the handle method receives the complete DomainEvent enum, a single implementation can match against multiple variants. For instance, a logging subscriber could react to both DomainEvent::ThreadGoalUpdated and DomainEvent::TaskRunReclaimed within the same match expression, processing each variant with separate logic arms.

Where are event subscribers registered in the OpenHuman codebase?

Subscriber registration occurs in src/core/all.rs through the register_subscribers function. This startup routine instantiates all domain-specific handler structs—such as TriggeredSkillSubscriber and voice Capture—and adds them to the global BUS via BUS.subscribe(Box::new(handler)) before the application begins accepting events.

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