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;