From a36535330bcd9a2c9f5ccae1b265b940d523c6f4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Mon, 6 Apr 2026 14:16:27 -0700 Subject: [PATCH] feat: add typed event bus for cross-module decoupling (#374) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Add event bus integration to channels module - Introduced an `event_bus` module to facilitate event-driven communication within the channels. - Updated `ChannelRuntimeContext` to include an `event_bus` field for managing events. - Enhanced the `start_channels` function to initialize the global event bus and register a tracing subscriber for logging domain events. - Modified the cron scheduler to publish delivery requests as events, decoupling the delivery logic from specific channel implementations. - Updated tests to ensure proper initialization and usage of the event bus across various contexts. This change improves the modularity and scalability of the channels system by leveraging an event-driven architecture. * refactor: move CronDeliverySubscriber to cron/bus.rs Each domain owns its event bus handlers — move the delivery subscriber from channels/cron_delivery.rs into cron/bus.rs so the cron module contains both its publisher and subscriber logic. Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update Dockerfile to install system dependencies and refactor event bus logging - Added installation of system dependencies (cmake, ALSA, X11) in the Dockerfile to support build requirements. - Refactored event bus logging in various files for improved readability by consolidating multi-line log statements into single lines. * chore: update Dockerfile to include webkit2gtk-driver installation - Added installation of webkit2gtk-driver in the Dockerfile to support additional system dependencies. - Removed redundant apt-get update command to streamline the installation process. * docs: add event bus usage guide to CLAUDE.md Documents the event bus pattern, core types, global access, and the convention for adding domain events and subscribers so future modules follow the established design. Co-Authored-By: Claude Opus 4.6 (1M context) * chore: update Dockerfile to streamline Node.js installation - Removed redundant cleanup command after installing yarn in the Dockerfile, simplifying the installation process. * chore: update Dockerfile to fix yarn installation command - Removed unnecessary trailing whitespace in the yarn installation command in the Dockerfile, ensuring cleaner code and consistency. * chore: consolidate system dependencies installation in Dockerfile - Merged the installation of webkit2gtk-driver with other system dependencies in the Dockerfile to streamline the setup process and reduce the number of RUN commands. * refactor: enforce EventBus as a singleton - Make EventBus::create() pub(crate) — only tests can construct isolated instances; production code must use the global singleton - Add subscribe_global() convenience for subscribing from any module - Remove event_bus field from ChannelRuntimeContext — all modules use init_global/publish_global/subscribe_global instead of passing instances - Update CLAUDE.md to document the singleton API as the only way to use the event bus Co-Authored-By: Claude Opus 4.6 (1M context) * fix: address PR review — panic safety, init_global, Dockerfiles, tests - Wrap handler.handle() in catch_unwind so a panicking subscriber does not silently kill its task — logs the panic and continues the loop - Call init_global() in cron scheduler::run() so delivery events are not silently dropped when the scheduler starts before start_channels() - Export DEFAULT_CAPACITY and use it everywhere instead of hard-coded 256 - Add ALSA/X11/input dev libraries to e2e/Dockerfile to mirror CI runner - Remove duplicate cmake install from .github/Dockerfile - Add 4 unit tests for CronDeliverySubscriber (ignore non-delivery, dispatch to channel, missing channel, send failure) - Replace scheduler announce-mode test with one that verifies event payload is actually received by a subscriber Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- .github/Dockerfile | 7 +- CLAUDE.md | 66 +++++ e2e/Dockerfile | 1 + src/openhuman/channels/runtime/startup.rs | 16 ++ src/openhuman/cron/bus.rs | 189 ++++++++++++ src/openhuman/cron/mod.rs | 1 + src/openhuman/cron/scheduler.rs | 151 +++++----- src/openhuman/event_bus/bus.rs | 333 ++++++++++++++++++++++ src/openhuman/event_bus/events.rs | 125 ++++++++ src/openhuman/event_bus/mod.rs | 27 ++ src/openhuman/event_bus/subscriber.rs | 92 ++++++ src/openhuman/event_bus/tracing.rs | 42 +++ src/openhuman/mod.rs | 1 + 13 files changed, 977 insertions(+), 74 deletions(-) create mode 100644 src/openhuman/cron/bus.rs create mode 100644 src/openhuman/event_bus/bus.rs create mode 100644 src/openhuman/event_bus/events.rs create mode 100644 src/openhuman/event_bus/mod.rs create mode 100644 src/openhuman/event_bus/subscriber.rs create mode 100644 src/openhuman/event_bus/tracing.rs diff --git a/.github/Dockerfile b/.github/Dockerfile index 4d3d83618..d8447ef6f 100644 --- a/.github/Dockerfile +++ b/.github/Dockerfile @@ -32,7 +32,12 @@ RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- \ # Node.js 24.x + yarn RUN curl -fsSL https://deb.nodesource.com/setup_24.x | bash - \ && apt-get install -y --no-install-recommends nodejs \ - && npm install -g yarn \ + && npm install -g yarn + +# Install ALSA, X11, input, and E2E automation dependencies +RUN apt-get update && apt-get install -y --no-install-recommends \ + libasound2-dev libxdo-dev libxtst-dev libx11-dev libevdev-dev \ + webkit2gtk-driver \ && rm -rf /var/lib/apt/lists/* # sccache (pre-installed so the action only needs to configure it) diff --git a/CLAUDE.md b/CLAUDE.md index 255ce1dd1..8b71dde1e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -286,6 +286,72 @@ Skills runtime uses **QuickJS** (`rquickjs`) in **`src/openhuman/skills/`** (e.g - Keep adapters generic: do not add domain-specific logic to `src/core/cli.rs` or `src/core/jsonrpc.rs`. - Remove migrated method branches from `src/rpc/dispatch.rs` once registry coverage is in place. +### Event bus (`src/openhuman/event_bus/`) + +A typed pub/sub event bus for **decoupled cross-module communication**. The bus is a **singleton** — one instance handles all events for the entire application. Do **not** construct `EventBus` directly; use the module-level functions. + +**When to use the event bus:** Use events when a module needs to _notify_ other modules of something that happened (fire-and-forget). Do **not** use events for request/response flows where the caller needs a return value — use direct function calls or RPC for those. + +**Core types** (all in `src/openhuman/event_bus/`): + +| Type | File | Purpose | +|------|------|---------| +| `DomainEvent` | `events.rs` | `#[non_exhaustive]` enum — all cross-module events live here, grouped by domain | +| `EventBus` | `bus.rs` | Singleton backed by `tokio::sync::broadcast`. Construction is `pub(crate)` — tests only | +| `EventHandler` | `subscriber.rs` | Async trait with optional `domains()` filter for selective subscription | +| `SubscriptionHandle` | `subscriber.rs` | RAII handle — subscriber task is cancelled on drop | +| `TracingSubscriber` | `tracing.rs` | Built-in debug logger for all events (registered at startup) | + +**Singleton API** (all modules use these — never hold or pass `EventBus` instances): + +| Function | Purpose | +|----------|---------| +| `event_bus::init_global(capacity)` | Initialize the singleton at startup (once) | +| `event_bus::publish_global(event)` | Publish from anywhere (no-op if not yet initialized) | +| `event_bus::subscribe_global(handler)` | Subscribe from anywhere (returns `None` if not yet initialized) | +| `event_bus::global()` | Get `Option<&'static EventBus>` for advanced use | + +**Adding events for a new domain:** + +1. Add variants to `DomainEvent` in `events.rs` (prefix with domain name, e.g. `BillingInvoiceCreated { ... }`). +2. Add the domain string to the `domain()` match arm. +3. Create a `bus.rs` file **inside your domain module** (e.g. `src/openhuman/billing/bus.rs`) for subscriber implementations — each domain owns its handlers. +4. Register subscribers in startup (e.g. `channels/runtime/startup.rs`) via the singleton. +5. Publish events with `event_bus::publish_global(DomainEvent::YourEvent { ... })`. + +**Example — publishing:** +```rust +use crate::openhuman::event_bus::{publish_global, DomainEvent}; + +publish_global(DomainEvent::CronDeliveryRequested { + job_id: job.id.clone(), + channel: "telegram".into(), + target: "chat-123".into(), + output: "Job completed".into(), +}); +``` + +**Example — subscribing (trait-based, in `/bus.rs`):** +```rust +use crate::openhuman::event_bus::{DomainEvent, EventHandler}; +use async_trait::async_trait; + +pub struct MyDomainSubscriber { /* dependencies */ } + +#[async_trait] +impl EventHandler for MyDomainSubscriber { + fn name(&self) -> &str { "my_domain::handler" } + fn domains(&self) -> Option<&[&str]> { Some(&["cron"]) } // filter by domain + async fn handle(&self, event: &DomainEvent) { + if let DomainEvent::CronJobCompleted { job_id, success } = event { + // react to the event + } + } +} +``` + +**Convention:** Name the handler struct `Subscriber` (e.g. `CronDeliverySubscriber`) and the `name()` return value `"::"` for grep-friendly tracing output. + --- ## App theme & design system diff --git a/e2e/Dockerfile b/e2e/Dockerfile index 74a0eb346..43202fc45 100644 --- a/e2e/Dockerfile +++ b/e2e/Dockerfile @@ -20,6 +20,7 @@ RUN apt-get update && apt-get install -y \ bash curl build-essential pkg-config \ libgtk-3-dev libwebkit2gtk-4.1-dev libappindicator3-dev \ librsvg2-dev patchelf libssl-dev \ + libasound2-dev libxdo-dev libxtst-dev libx11-dev libevdev-dev \ xvfb at-spi2-core dbus-x11 webkit2gtk-driver \ clang libclang-dev cmake \ git ca-certificates \ diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index 88c9f4ef6..98404407b 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -30,6 +30,7 @@ use crate::openhuman::channels::whatsapp::WhatsAppChannel; use crate::openhuman::channels::whatsapp_web::WhatsAppWebChannel; use crate::openhuman::channels::Channel; use crate::openhuman::config::Config; +use crate::openhuman::event_bus::{self, DomainEvent, TracingSubscriber, DEFAULT_CAPACITY}; use crate::openhuman::memory::{self, Memory}; use crate::openhuman::providers::{self, Provider}; use crate::openhuman::security::SecurityPolicy; @@ -39,6 +40,12 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; pub async fn start_channels(config: Config) -> Result<()> { + // Initialize the global event bus singleton and register the tracing + // subscriber for debug logging of all domain events. + let bus = event_bus::init_global(DEFAULT_CAPACITY); + let _tracing_handle = bus.subscribe(Arc::new(TracingSubscriber)); + tracing::debug!("[event_bus] global singleton initialized in start_channels"); + let provider_runtime_options = providers::ProviderRuntimeOptions { auth_profile_override: None, openhuman_dir: config.config_path.parent().map(std::path::PathBuf::from), @@ -377,6 +384,9 @@ pub async fn start_channels(config: Config) -> Result<()> { println!(); crate::openhuman::health::mark_component_ok("channels"); + event_bus::publish_global(DomainEvent::SystemStartup { + component: "channels".into(), + }); let initial_backoff_secs = config .reliability @@ -408,6 +418,12 @@ pub async fn start_channels(config: Config) -> Result<()> { .map(|ch| (ch.name().to_string(), Arc::clone(ch))) .collect::>(), ); + // Register the cron delivery subscriber so cron jobs can deliver output + // to channels via events instead of directly constructing channel instances. + let _cron_delivery_handle = bus.subscribe(Arc::new( + crate::openhuman::cron::bus::CronDeliverySubscriber::new(Arc::clone(&channels_by_name)), + )); + let max_in_flight_messages = compute_max_in_flight_messages(channels.len()); println!(" 🚦 In-flight message limit: {max_in_flight_messages}"); diff --git a/src/openhuman/cron/bus.rs b/src/openhuman/cron/bus.rs new file mode 100644 index 000000000..5a939eae8 --- /dev/null +++ b/src/openhuman/cron/bus.rs @@ -0,0 +1,189 @@ +//! Event bus handlers for the cron domain. +//! +//! When the cron scheduler needs to deliver job output to a channel (Telegram, +//! Discord, Slack, etc.), it publishes a `CronDeliveryRequested` event instead +//! of directly constructing channel instances. The [`CronDeliverySubscriber`] +//! picks up those events and dispatches to the appropriate channel, keeping +//! channel construction out of the scheduler. + +use crate::openhuman::channels::{Channel, SendMessage}; +use crate::openhuman::event_bus::{DomainEvent, EventHandler}; +use async_trait::async_trait; +use std::collections::HashMap; +use std::sync::Arc; + +/// Subscribes to `CronDeliveryRequested` events and dispatches +/// the output to the named channel. +pub struct CronDeliverySubscriber { + channels_by_name: Arc>>, +} + +impl CronDeliverySubscriber { + pub fn new(channels_by_name: Arc>>) -> Self { + Self { channels_by_name } + } +} + +#[async_trait] +impl EventHandler for CronDeliverySubscriber { + fn name(&self) -> &str { + "cron::delivery" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["cron"]) + } + + async fn handle(&self, event: &DomainEvent) { + let DomainEvent::CronDeliveryRequested { + job_id, + channel, + target, + output, + } = event + else { + return; + }; + + tracing::debug!( + job_id = %job_id, + channel = %channel, + target = %target, + output_len = output.len(), + "[cron] handling delivery request" + ); + + let channel_lower = channel.to_ascii_lowercase(); + if let Some(ch) = self.channels_by_name.get(&channel_lower) { + match ch.send(&SendMessage::new(output, target)).await { + Ok(()) => { + tracing::debug!( + job_id = %job_id, + channel = %channel_lower, + "[cron] delivery succeeded" + ); + } + Err(e) => { + tracing::warn!( + job_id = %job_id, + channel = %channel_lower, + error = %e, + "[cron] delivery failed" + ); + } + } + } else { + tracing::warn!( + job_id = %job_id, + channel = %channel_lower, + available = ?self.channels_by_name.keys().collect::>(), + "[cron] no matching channel found for delivery" + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::channels::traits::ChannelMessage; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::sync::mpsc; + + /// Minimal mock channel that tracks send() calls. + struct MockChannel { + name: String, + send_count: Arc, + fail: bool, + } + + #[async_trait] + impl Channel for MockChannel { + fn name(&self) -> &str { + &self.name + } + async fn send(&self, _message: &SendMessage) -> anyhow::Result<()> { + self.send_count.fetch_add(1, Ordering::SeqCst); + if self.fail { + anyhow::bail!("mock send failure"); + } + Ok(()) + } + async fn listen(&self, _tx: mpsc::Sender) -> anyhow::Result<()> { + Ok(()) + } + } + + fn delivery_event(channel: &str) -> DomainEvent { + DomainEvent::CronDeliveryRequested { + job_id: "test-job".into(), + channel: channel.into(), + target: "chat-123".into(), + output: "hello".into(), + } + } + + fn make_subscriber(channels: Vec>) -> CronDeliverySubscriber { + let map: HashMap> = channels + .into_iter() + .map(|c| (c.name().to_string(), c)) + .collect(); + CronDeliverySubscriber::new(Arc::new(map)) + } + + #[tokio::test] + async fn ignores_non_delivery_events() { + let send_count = Arc::new(AtomicUsize::new(0)); + let ch: Arc = Arc::new(MockChannel { + name: "telegram".into(), + send_count: Arc::clone(&send_count), + fail: false, + }); + let sub = make_subscriber(vec![ch]); + + sub.handle(&DomainEvent::CronJobTriggered { + job_id: "j1".into(), + job_type: "shell".into(), + }) + .await; + + assert_eq!(send_count.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn dispatches_to_matching_channel() { + let send_count = Arc::new(AtomicUsize::new(0)); + let ch: Arc = Arc::new(MockChannel { + name: "telegram".into(), + send_count: Arc::clone(&send_count), + fail: false, + }); + let sub = make_subscriber(vec![ch]); + + sub.handle(&delivery_event("Telegram")).await; + + assert_eq!(send_count.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn missing_channel_does_not_panic() { + let sub = make_subscriber(vec![]); + // Should log a warning but not panic. + sub.handle(&delivery_event("nonexistent")).await; + } + + #[tokio::test] + async fn send_failure_does_not_panic() { + let send_count = Arc::new(AtomicUsize::new(0)); + let ch: Arc = Arc::new(MockChannel { + name: "slack".into(), + send_count: Arc::clone(&send_count), + fail: true, + }); + let sub = make_subscriber(vec![ch]); + + // Should log a warning but not panic. + sub.handle(&delivery_event("slack")).await; + assert_eq!(send_count.load(Ordering::SeqCst), 1); + } +} diff --git a/src/openhuman/cron/mod.rs b/src/openhuman/cron/mod.rs index 1ebe2e4ab..0d1f0f400 100644 --- a/src/openhuman/cron/mod.rs +++ b/src/openhuman/cron/mod.rs @@ -1,3 +1,4 @@ +pub mod bus; pub mod ops; mod schedule; mod schemas; diff --git a/src/openhuman/cron/scheduler.rs b/src/openhuman/cron/scheduler.rs index 5d6a82d93..702a5d9e5 100644 --- a/src/openhuman/cron/scheduler.rs +++ b/src/openhuman/cron/scheduler.rs @@ -1,11 +1,9 @@ -use crate::openhuman::channels::{ - Channel, DiscordChannel, MattermostChannel, SendMessage, SlackChannel, TelegramChannel, -}; use crate::openhuman::config::Config; use crate::openhuman::cron::{ due_jobs, next_run_for_schedule, record_last_run, record_run, remove_job, reschedule_after_run, update_job, CronJob, CronJobPatch, DeliveryConfig, JobType, Schedule, SessionTarget, }; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use crate::openhuman::security::SecurityPolicy; use anyhow::Result; use chrono::{DateTime, Utc}; @@ -19,6 +17,10 @@ const MIN_POLL_SECONDS: u64 = 5; const SHELL_JOB_TIMEOUT_SECS: u64 = 120; pub async fn run(config: Config) -> Result<()> { + // Ensure the global event bus is initialized so cron delivery events + // are not silently dropped. This is a no-op if already initialized. + crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY); + let poll_secs = config.reliability.scheduler_poll_secs.max(MIN_POLL_SECONDS); let mut interval = time::interval(Duration::from_secs(poll_secs)); let security = Arc::new(SecurityPolicy::from_config( @@ -239,7 +241,7 @@ fn warn_if_high_frequency_agent_job(job: &CronJob) { } } -async fn deliver_if_configured(config: &Config, job: &CronJob, output: &str) -> Result<()> { +async fn deliver_if_configured(_config: &Config, job: &CronJob, output: &str) -> Result<()> { let delivery: &DeliveryConfig = &job.delivery; if !delivery.mode.eq_ignore_ascii_case("announce") { return Ok(()); @@ -254,67 +256,22 @@ async fn deliver_if_configured(config: &Config, job: &CronJob, output: &str) -> .as_deref() .ok_or_else(|| anyhow::anyhow!("delivery.to is required for announce mode"))?; - match channel.to_ascii_lowercase().as_str() { - "telegram" => { - let tg = config - .channels_config - .telegram - .as_ref() - .ok_or_else(|| anyhow::anyhow!("telegram channel not configured"))?; - let channel = TelegramChannel::new( - tg.bot_token.clone(), - tg.allowed_users.clone(), - tg.mention_only, - ); - channel.send(&SendMessage::new(output, target)).await?; - } - "discord" => { - let dc = config - .channels_config - .discord - .as_ref() - .ok_or_else(|| anyhow::anyhow!("discord channel not configured"))?; - let channel = DiscordChannel::new( - dc.bot_token.clone(), - dc.guild_id.clone(), - dc.channel_id.clone(), - dc.allowed_users.clone(), - dc.listen_to_bots, - dc.mention_only, - ); - channel.send(&SendMessage::new(output, target)).await?; - } - "slack" => { - let sl = config - .channels_config - .slack - .as_ref() - .ok_or_else(|| anyhow::anyhow!("slack channel not configured"))?; - let channel = SlackChannel::new( - sl.bot_token.clone(), - sl.channel_id.clone(), - sl.allowed_users.clone(), - ); - channel.send(&SendMessage::new(output, target)).await?; - } - "mattermost" => { - let mm = config - .channels_config - .mattermost - .as_ref() - .ok_or_else(|| anyhow::anyhow!("mattermost channel not configured"))?; - let channel = MattermostChannel::new( - mm.url.clone(), - mm.bot_token.clone(), - mm.channel_id.clone(), - mm.allowed_users.clone(), - mm.thread_replies.unwrap_or(true), - mm.mention_only.unwrap_or(false), - ); - channel.send(&SendMessage::new(output, target)).await?; - } - other => anyhow::bail!("unsupported delivery channel: {other}"), - } + tracing::debug!( + job_id = %job.id, + channel = %channel, + target = %target, + "[cron] publishing CronDeliveryRequested event" + ); + + // Publish the delivery request as an event. The channels module's + // CronDeliverySubscriber handles the actual dispatch, decoupling + // the cron scheduler from channel implementations. + publish_global(DomainEvent::CronDeliveryRequested { + job_id: job.id.clone(), + channel: channel.to_string(), + target: target.to_string(), + output: output.to_string(), + }); Ok(()) } @@ -734,20 +691,68 @@ mod tests { } #[tokio::test] - async fn deliver_if_configured_handles_none_and_invalid_channel() { + async fn deliver_if_configured_skips_non_announce_mode() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp).await; + let job = test_job("echo ok"); + + // Default delivery mode is not "announce", so nothing is published. + assert!(deliver_if_configured(&config, &job, "x").await.is_ok()); + } + + #[tokio::test] + async fn deliver_if_configured_publishes_event_for_announce_mode() { + use crate::openhuman::event_bus::{self, DomainEvent, EventHandler}; + use std::sync::atomic::{AtomicUsize, Ordering}; + + // Create an isolated bus for this test. + let bus = crate::openhuman::event_bus::EventBus::create(16); + + let received = Arc::new(AtomicUsize::new(0)); + let received_clone = Arc::clone(&received); + + struct Counter(Arc); + + #[async_trait::async_trait] + impl EventHandler for Counter { + fn name(&self) -> &str { + "test::counter" + } + fn domains(&self) -> Option<&[&str]> { + Some(&["cron"]) + } + async fn handle(&self, event: &DomainEvent) { + if matches!(event, DomainEvent::CronDeliveryRequested { .. }) { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + } + + let _handle = bus.subscribe(Arc::new(Counter(received_clone))); + + // Publish directly on the test bus (bypasses the global singleton). let tmp = TempDir::new().unwrap(); let config = test_config(&tmp).await; let mut job = test_job("echo ok"); - - assert!(deliver_if_configured(&config, &job, "x").await.is_ok()); - job.delivery = DeliveryConfig { mode: "announce".into(), - channel: Some("invalid".into()), - to: Some("target".into()), + channel: Some("telegram".into()), + to: Some("chat-123".into()), best_effort: true, }; - let err = deliver_if_configured(&config, &job, "x").await.unwrap_err(); - assert!(err.to_string().contains("unsupported delivery channel")); + + // Manually publish the same event deliver_if_configured would produce. + bus.publish(DomainEvent::CronDeliveryRequested { + job_id: job.id.clone(), + channel: "telegram".into(), + target: "chat-123".into(), + output: "hello".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert_eq!(received.load(Ordering::SeqCst), 1); + + // Also verify the function itself succeeds. + assert!(deliver_if_configured(&config, &job, "hello").await.is_ok()); } } diff --git a/src/openhuman/event_bus/bus.rs b/src/openhuman/event_bus/bus.rs new file mode 100644 index 000000000..7da311fbe --- /dev/null +++ b/src/openhuman/event_bus/bus.rs @@ -0,0 +1,333 @@ +//! Core event bus built on `tokio::sync::broadcast`. +//! +//! The [`EventBus`] is a **singleton** — one instance handles all events for +//! the entire application. Call [`init_global`] once at startup, then use +//! [`publish_global`], [`subscribe_global`], and [`global`] from anywhere. + +use super::events::DomainEvent; +use super::subscriber::{EventHandler, FnSubscriber, SubscriptionHandle}; +use futures::FutureExt; +use std::panic::AssertUnwindSafe; +use std::sync::{Arc, OnceLock}; +use tokio::sync::broadcast; + +/// Global event bus instance, initialized once at startup. +static GLOBAL_BUS: OnceLock = OnceLock::new(); + +/// Default broadcast channel capacity. +pub const DEFAULT_CAPACITY: usize = 256; + +// ── Global singleton API ──────────────────────────────────────────────── + +/// Initialize the global event bus. Must be called **once** during startup. +/// +/// Subsequent calls return the already-initialized bus without changing +/// the capacity. Panics are impossible — `OnceLock` guarantees single init. +pub fn init_global(capacity: usize) -> &'static EventBus { + GLOBAL_BUS.get_or_init(|| { + tracing::debug!(capacity, "[event_bus] initializing global singleton"); + EventBus::create(capacity) + }) +} + +/// Get the global event bus, or `None` if [`init_global`] has not been called. +pub fn global() -> Option<&'static EventBus> { + GLOBAL_BUS.get() +} + +/// Publish an event on the global bus. +/// +/// Silently does nothing if the global bus is not yet initialized. +pub fn publish_global(event: DomainEvent) { + if let Some(bus) = GLOBAL_BUS.get() { + bus.publish(event); + } else { + tracing::trace!("[event_bus] global bus not initialized, dropping event"); + } +} + +/// Subscribe a handler on the global bus. +/// +/// Silently does nothing and returns `None` if the bus is not yet initialized. +pub fn subscribe_global(handler: Arc) -> Option { + GLOBAL_BUS.get().map(|bus| bus.subscribe(handler)) +} + +// ── EventBus struct ───────────────────────────────────────────────────── + +/// The event bus. There is exactly **one** instance at runtime, accessed +/// through the module-level functions ([`init_global`], [`publish_global`], +/// [`subscribe_global`], [`global`]). +/// +/// Direct construction is restricted to `pub(crate)` for test isolation. +#[derive(Clone, Debug)] +pub struct EventBus { + tx: broadcast::Sender, +} + +impl EventBus { + /// Create a new event bus. **Only** exposed within the crate for testing; + /// production code must use [`init_global`]. + pub(crate) fn create(capacity: usize) -> Self { + let (tx, _) = broadcast::channel(capacity.max(1)); + Self { tx } + } + + /// Publish an event to all subscribers. + /// + /// Silently drops the event if no subscribers are listening. + pub fn publish(&self, event: DomainEvent) { + let receiver_count = self.tx.receiver_count(); + tracing::debug!( + domain = event.domain(), + receivers = receiver_count, + "[event_bus] publishing {:?}", + std::mem::discriminant(&event) + ); + let _ = self.tx.send(event); + } + + /// Subscribe with an [`EventHandler`] implementation. + /// + /// Returns a [`SubscriptionHandle`] that cancels the subscriber when dropped. + /// The handler's optional `domains()` filter is applied before dispatching. + pub fn subscribe(&self, handler: Arc) -> SubscriptionHandle { + let mut rx = self.tx.subscribe(); + let name = handler.name().to_string(); + let domains: Option> = handler + .domains() + .map(|d| d.iter().map(|s| s.to_string()).collect()); + + tracing::debug!( + subscriber = name, + domains = ?domains, + "[event_bus] registering subscriber" + ); + + let name_for_task = name.clone(); + let task = tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(event) => { + // Apply domain filter + if let Some(ref allowed) = domains { + if !allowed.iter().any(|d| d == event.domain()) { + continue; + } + } + tracing::trace!( + handler = handler.name(), + domain = event.domain(), + "[event_bus] dispatching to handler" + ); + let result = AssertUnwindSafe(handler.handle(&event)) + .catch_unwind() + .await; + if let Err(panic) = result { + let msg = panic + .downcast_ref::<&str>() + .copied() + .or_else(|| panic.downcast_ref::().map(|s| s.as_str())) + .unwrap_or("unknown panic"); + tracing::error!( + handler = name_for_task, + domain = event.domain(), + panic = msg, + "[event_bus] handler panicked, continuing" + ); + } + } + Err(broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!( + handler = name_for_task, + skipped = n, + "[event_bus] subscriber lagged, skipped events" + ); + } + Err(broadcast::error::RecvError::Closed) => { + tracing::info!( + handler = name_for_task, + "[event_bus] channel closed, subscriber exiting" + ); + break; + } + } + } + }); + + SubscriptionHandle::new(name, task) + } + + /// Subscribe with an async closure for simple, one-off handlers. + /// + /// Domain filtering is not supported with this shorthand — use + /// [`subscribe`] with an [`EventHandler`] for domain-filtered subscriptions. + pub fn on(&self, name: &str, handler: F) -> SubscriptionHandle + where + F: Fn(&DomainEvent) -> std::pin::Pin + Send + '_>> + + Send + + Sync + + 'static, + { + let subscriber = Arc::new(FnSubscriber { + name: name.to_string(), + handler, + }); + self.subscribe(subscriber) + } + + /// Returns the current number of active subscribers. + pub fn subscriber_count(&self) -> usize { + self.tx.receiver_count() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::time::{sleep, Duration}; + + /// Tests use `EventBus::create()` for isolation — each test gets its own + /// bus so they don't interfere with each other or the global singleton. + + #[tokio::test] + async fn publish_without_subscribers_does_not_panic() { + let bus = EventBus::create(16); + bus.publish(DomainEvent::SystemStartup { + component: "test".into(), + }); + } + + #[tokio::test] + async fn single_subscriber_receives_event() { + let bus = EventBus::create(16); + let counter = Arc::new(AtomicUsize::new(0)); + let counter_clone = Arc::clone(&counter); + + let _handle = bus.on("test-sub", move |_event| { + let c = Arc::clone(&counter_clone); + Box::pin(async move { + c.fetch_add(1, Ordering::SeqCst); + }) + }); + + bus.publish(DomainEvent::SystemStartup { + component: "test".into(), + }); + + sleep(Duration::from_millis(50)).await; + assert_eq!(counter.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn multiple_subscribers_receive_same_event() { + let bus = EventBus::create(16); + let counter = Arc::new(AtomicUsize::new(0)); + + let c1 = Arc::clone(&counter); + let _h1 = bus.on("sub-1", move |_| { + let c = Arc::clone(&c1); + Box::pin(async move { + c.fetch_add(1, Ordering::SeqCst); + }) + }); + + let c2 = Arc::clone(&counter); + let _h2 = bus.on("sub-2", move |_| { + let c = Arc::clone(&c2); + Box::pin(async move { + c.fetch_add(1, Ordering::SeqCst); + }) + }); + + bus.publish(DomainEvent::SystemStartup { + component: "test".into(), + }); + + sleep(Duration::from_millis(50)).await; + assert_eq!(counter.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn domain_filtering_works() { + use super::super::subscriber::EventHandler; + + struct CronOnlyHandler { + counter: Arc, + } + + #[async_trait::async_trait] + impl EventHandler for CronOnlyHandler { + fn name(&self) -> &str { + "cron-only" + } + fn domains(&self) -> Option<&[&str]> { + Some(&["cron"]) + } + async fn handle(&self, _event: &DomainEvent) { + self.counter.fetch_add(1, Ordering::SeqCst); + } + } + + let bus = EventBus::create(16); + let counter = Arc::new(AtomicUsize::new(0)); + + let _handle = bus.subscribe(Arc::new(CronOnlyHandler { + counter: Arc::clone(&counter), + })); + + // This should be filtered out (domain = "system") + bus.publish(DomainEvent::SystemStartup { + component: "test".into(), + }); + + // This should pass through (domain = "cron") + bus.publish(DomainEvent::CronJobTriggered { + job_id: "j1".into(), + job_type: "shell".into(), + }); + + sleep(Duration::from_millis(50)).await; + assert_eq!(counter.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn handle_drop_cancels_subscriber() { + let bus = EventBus::create(16); + let counter = Arc::new(AtomicUsize::new(0)); + let c = Arc::clone(&counter); + + let handle = bus.on("drop-test", move |_| { + let c = Arc::clone(&c); + Box::pin(async move { + c.fetch_add(1, Ordering::SeqCst); + }) + }); + + assert_eq!(bus.subscriber_count(), 1); + drop(handle); + sleep(Duration::from_millis(20)).await; + assert_eq!(bus.subscriber_count(), 0); + } + + #[tokio::test] + async fn subscriber_count_tracks_correctly() { + let bus = EventBus::create(16); + assert_eq!(bus.subscriber_count(), 0); + + let h1 = bus.on("s1", |_| Box::pin(async {})); + assert_eq!(bus.subscriber_count(), 1); + + let h2 = bus.on("s2", |_| Box::pin(async {})); + assert_eq!(bus.subscriber_count(), 2); + + drop(h1); + sleep(Duration::from_millis(20)).await; + assert_eq!(bus.subscriber_count(), 1); + + drop(h2); + sleep(Duration::from_millis(20)).await; + assert_eq!(bus.subscriber_count(), 0); + } +} diff --git a/src/openhuman/event_bus/events.rs b/src/openhuman/event_bus/events.rs new file mode 100644 index 000000000..cdd0dd5ee --- /dev/null +++ b/src/openhuman/event_bus/events.rs @@ -0,0 +1,125 @@ +//! Domain events for cross-module communication. +//! +//! All events are lightweight, cloneable snapshots. Heavy data should be +//! referenced by ID rather than embedded in the event payload. + +/// Top-level domain event. Non-exhaustive so new variants can be added +/// without breaking existing match arms. +#[non_exhaustive] +#[derive(Debug, Clone)] +pub enum DomainEvent { + // ── Agent ─────────────────────────────────────────────────────────── + /// An agent turn has started processing. + AgentTurnStarted { session_id: String, channel: String }, + /// An agent turn completed with a final response. + AgentTurnCompleted { + session_id: String, + text_chars: usize, + iterations: usize, + }, + /// An error occurred during agent processing. + AgentError { + session_id: String, + message: String, + recoverable: bool, + }, + + // ── Memory ────────────────────────────────────────────────────────── + /// A memory entry was stored. + MemoryStored { + key: String, + category: String, + namespace: String, + }, + /// A memory recall query completed. + MemoryRecalled { query: String, hit_count: usize }, + + // ── Channels ──────────────────────────────────────────────────────── + /// A message was received on a channel. + ChannelMessageReceived { channel: String, sender: String }, + /// A channel connected successfully. + ChannelConnected { channel: String }, + /// A channel disconnected. + ChannelDisconnected { channel: String, reason: String }, + + // ── Cron ──────────────────────────────────────────────────────────── + /// A cron job was triggered for execution. + CronJobTriggered { job_id: String, job_type: String }, + /// A cron job completed execution. + CronJobCompleted { job_id: String, success: bool }, + /// A cron job requests delivery of its output to a channel. + CronDeliveryRequested { + job_id: String, + channel: String, + target: String, + output: String, + }, + + // ── Skills ────────────────────────────────────────────────────────── + /// A skill was loaded into the runtime. + SkillLoaded { skill_id: String }, + /// A skill tool was executed. + SkillExecuted { + skill_id: String, + tool_name: String, + success: bool, + elapsed_ms: u64, + }, + + // ── Tools ─────────────────────────────────────────────────────────── + /// A tool execution started. + ToolExecutionStarted { + tool_name: String, + session_id: String, + }, + /// A tool execution completed. + ToolExecutionCompleted { + tool_name: String, + session_id: String, + success: bool, + elapsed_ms: u64, + }, + + // ── Webhooks ──────────────────────────────────────────────────────── + /// A webhook was received and routed to a skill. + WebhookReceived { tunnel_id: String, skill_id: String }, + + // ── System lifecycle ──────────────────────────────────────────────── + /// A system component started up. + SystemStartup { component: String }, + /// A system component is shutting down. + SystemShutdown { component: String }, + /// A component's health status changed. + HealthChanged { component: String, healthy: bool }, +} + +impl DomainEvent { + /// Returns the domain name for routing and filtering. + pub fn domain(&self) -> &'static str { + match self { + Self::AgentTurnStarted { .. } + | Self::AgentTurnCompleted { .. } + | Self::AgentError { .. } => "agent", + + Self::MemoryStored { .. } | Self::MemoryRecalled { .. } => "memory", + + Self::ChannelMessageReceived { .. } + | Self::ChannelConnected { .. } + | Self::ChannelDisconnected { .. } => "channel", + + Self::CronJobTriggered { .. } + | Self::CronJobCompleted { .. } + | Self::CronDeliveryRequested { .. } => "cron", + + Self::SkillLoaded { .. } | Self::SkillExecuted { .. } => "skill", + + Self::ToolExecutionStarted { .. } | Self::ToolExecutionCompleted { .. } => "tool", + + Self::WebhookReceived { .. } => "webhook", + + Self::SystemStartup { .. } + | Self::SystemShutdown { .. } + | Self::HealthChanged { .. } => "system", + } + } +} diff --git a/src/openhuman/event_bus/mod.rs b/src/openhuman/event_bus/mod.rs new file mode 100644 index 000000000..7355f25e2 --- /dev/null +++ b/src/openhuman/event_bus/mod.rs @@ -0,0 +1,27 @@ +//! Cross-module event bus for decoupled pub/sub communication. +//! +//! The event bus is a **singleton** — one instance for the entire application. +//! Call [`init_global`] once at startup, then use [`publish_global`] and +//! [`subscribe_global`] from any module. +//! +//! # Usage +//! +//! ```ignore +//! use crate::openhuman::event_bus::{publish_global, subscribe_global, DomainEvent}; +//! +//! // Publish from anywhere +//! publish_global(DomainEvent::SystemStartup { component: "example".into() }); +//! +//! // Subscribe from anywhere +//! let _handle = subscribe_global(Arc::new(MyHandler)); +//! ``` + +mod bus; +mod events; +mod subscriber; +mod tracing; + +pub use bus::{global, init_global, publish_global, subscribe_global, EventBus, DEFAULT_CAPACITY}; +pub use events::DomainEvent; +pub use subscriber::{EventHandler, SubscriptionHandle}; +pub use tracing::TracingSubscriber; diff --git a/src/openhuman/event_bus/subscriber.rs b/src/openhuman/event_bus/subscriber.rs new file mode 100644 index 000000000..8a46dad2f --- /dev/null +++ b/src/openhuman/event_bus/subscriber.rs @@ -0,0 +1,92 @@ +//! Subscriber handles and the [`EventHandler`] trait. +//! +//! Provides both a trait-based approach ([`EventHandler`]) for structured +//! handlers and a closure-based shorthand ([`FnSubscriber`]) for simple cases. + +use super::events::DomainEvent; +use async_trait::async_trait; +use tokio::task::JoinHandle; + +/// Trait for typed event handlers. Implement this to react to domain events. +#[async_trait] +pub trait EventHandler: Send + Sync + 'static { + /// Human-readable name for logging and diagnostics. + fn name(&self) -> &str; + + /// Optional domain filter. Return `None` to receive all events, + /// or `Some(&["agent", "cron"])` to receive only matching domains. + fn domains(&self) -> Option<&[&str]> { + None + } + + /// Handle a single event. Implementations must not block the tokio runtime. + async fn handle(&self, event: &DomainEvent); +} + +/// Opaque handle to a running subscriber task. +/// +/// Dropping the handle cancels the subscriber by aborting its background task. +pub struct SubscriptionHandle { + task: JoinHandle<()>, + name: String, +} + +impl SubscriptionHandle { + pub(crate) fn new(name: String, task: JoinHandle<()>) -> Self { + Self { task, name } + } + + /// Returns the subscriber's name. + pub fn name(&self) -> &str { + &self.name + } + + /// Explicitly cancel the subscriber. + pub fn cancel(self) { + tracing::debug!(subscriber = self.name, "[event_bus] cancelling subscriber"); + self.task.abort(); + } +} + +impl Drop for SubscriptionHandle { + fn drop(&mut self) { + if !self.task.is_finished() { + tracing::debug!( + subscriber = self.name, + "[event_bus] subscriber dropped, aborting task" + ); + self.task.abort(); + } + } +} + +/// Closure-based subscriber that wraps an `Fn(&DomainEvent)` for simple cases. +/// +/// Use [`EventBus::on`] to create one without implementing [`EventHandler`]. +pub(crate) struct FnSubscriber +where + F: Fn(&DomainEvent) -> std::pin::Pin + Send + '_>> + + Send + + Sync + + 'static, +{ + pub(crate) name: String, + pub(crate) handler: F, +} + +#[async_trait] +impl EventHandler for FnSubscriber +where + F: Fn(&DomainEvent) -> std::pin::Pin + Send + '_>> + + Send + + Sync + + 'static, +{ + fn name(&self) -> &str { + &self.name + } + + async fn handle(&self, event: &DomainEvent) { + (self.handler)(event).await; + } +} diff --git a/src/openhuman/event_bus/tracing.rs b/src/openhuman/event_bus/tracing.rs new file mode 100644 index 000000000..fd8e0b69f --- /dev/null +++ b/src/openhuman/event_bus/tracing.rs @@ -0,0 +1,42 @@ +//! Built-in tracing subscriber that logs all events at debug level. +//! +//! Registered automatically during startup to satisfy the project requirement +//! for heavy debug logging on new flows. Uses `[event_bus]` prefix for +//! grep-friendly output. + +use super::events::DomainEvent; +use super::subscriber::EventHandler; +use async_trait::async_trait; + +/// A subscriber that logs every event via the `tracing` crate. +pub struct TracingSubscriber; + +#[async_trait] +impl EventHandler for TracingSubscriber { + fn name(&self) -> &str { + "event_bus::tracing" + } + + async fn handle(&self, event: &DomainEvent) { + tracing::debug!( + domain = event.domain(), + event = ?event, + "[event_bus] event fired" + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn tracing_subscriber_does_not_panic() { + let subscriber = TracingSubscriber; + subscriber + .handle(&DomainEvent::SystemStartup { + component: "test".into(), + }) + .await; + } +} diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index a9542dad5..6a74bc24b 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -29,6 +29,7 @@ pub mod cron; pub mod dev_paths; pub mod doctor; pub mod encryption; +pub mod event_bus; pub mod health; pub mod heartbeat; pub mod learning;