From 3d17bfb23d7e45ff03a4ee8071e607ba3d1e4c43 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Thu, 9 Apr 2026 16:44:53 -0700 Subject: [PATCH] Refactor core cross-domain flows onto event bus (#468) * Refactor core cross-domain flows onto event bus * feat(agent): enhance error handling and message processing - Introduced a new constant for maximum error message length in the Agent. - Added methods for comparing conversation messages and determining new entries for turns. - Implemented a function to sanitize error messages, improving clarity and consistency in error reporting. - Updated the `run_single` method to utilize the new error handling and message processing logic, ensuring better tracking of conversation history and error states. * feat(supervision): initialize global event bus and register health subscriber - Added initialization of the global event bus with default capacity to ensure channel health events have a live bus and subscriber target. - Registered a health subscriber to enhance monitoring capabilities within the supervised listener context. --- docs/memory-sync-functions.md | 4 - src/core/jsonrpc.rs | 7 +- src/openhuman/agent/agent.rs | 144 +++++++++++- src/openhuman/agent/events.rs | 216 ------------------ src/openhuman/agent/mod.rs | 2 - src/openhuman/agent/observer.rs | 22 -- src/openhuman/channels/providers/web.rs | 25 +- src/openhuman/channels/runtime/startup.rs | 3 +- src/openhuman/channels/runtime/supervision.rs | 15 +- src/openhuman/cron/scheduler.rs | 39 +++- src/openhuman/event_bus/events.rs | 18 +- src/openhuman/health/bus.rs | 115 ++++++++++ src/openhuman/health/mod.rs | 1 + src/openhuman/skills/bus.rs | 83 ++++++- src/openhuman/skills/qjs_engine.rs | 2 - src/openhuman/update/scheduler.rs | 20 +- 16 files changed, 435 insertions(+), 281 deletions(-) delete mode 100644 src/openhuman/agent/events.rs delete mode 100644 src/openhuman/agent/observer.rs create mode 100644 src/openhuman/health/bus.rs diff --git a/docs/memory-sync-functions.md b/docs/memory-sync-functions.md index 9b754e185..7693b0293 100644 --- a/docs/memory-sync-functions.md +++ b/docs/memory-sync-functions.md @@ -198,11 +198,7 @@ let relations = client.graph_query(None, None, None).await?; |----------|-------------| | `delete_document(namespace, id)` | Remove a specific document | | `kv_delete(namespace, key)` | Remove a specific KV entry | -<<<<<<< HEAD | `clear_skill_memory(skill_id, integration_id)` | Disconnect / revoke: clears skill-scoped memory in the shared `skill-{skill_id}` namespace. Storage is not isolated per integration—multiple integrations share that namespace; `integration_id` identifies the integration in the API contract (see implementation in `MemoryClient::clear_skill_memory`) | -======= -| `clear_skill_memory(skill_id, integration_id)` | Skill OAuth/auth revoked — wipes `skill-{skill_id}` namespace | ->>>>>>> 5b6b1e2 (feat(subconscious): stabilize heartbeat + subconscious loop (#392)) | `clear_namespace(namespace)` | Wipe an arbitrary namespace | --- diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index b9eb1855f..ed425d5cc 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -777,11 +777,16 @@ fn register_domain_subscribers() { log::warn!("[event_bus] failed to register channel subscriber — bus not initialized"); } + crate::openhuman::health::bus::register_health_subscriber(); + crate::openhuman::skills::bus::register_skill_cleanup_subscriber(); + // Restart requests go through a subscriber so every trigger path shares // the same respawn logic. crate::openhuman::service::bus::register_restart_subscriber(); - log::info!("[event_bus] webhook, channel, and restart subscribers registered"); + log::info!( + "[event_bus] webhook, channel, health, skill, and restart subscribers registered" + ); }); } diff --git a/src/openhuman/agent/agent.rs b/src/openhuman/agent/agent.rs index 4f96e8dc4..608e3ddb1 100644 --- a/src/openhuman/agent/agent.rs +++ b/src/openhuman/agent/agent.rs @@ -8,11 +8,13 @@ use super::dispatcher::{ NativeToolDispatcher, ParsedToolCall, ToolDispatcher, ToolExecutionResult, XmlToolDispatcher, }; +use super::error::AgentError; use super::hooks::{self, sanitize_tool_output, PostTurnHook, ToolCallRecord, TurnContext}; use super::memory_loader::{DefaultMemoryLoader, MemoryLoader}; use super::prompt::{PromptContext, SystemPromptBuilder}; use crate::openhuman::agent::host_runtime; use crate::openhuman::config::Config; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use crate::openhuman::memory::{self, Memory, MemoryCategory}; use crate::openhuman::providers::{ self, ChatMessage, ChatRequest, ConversationMessage, Provider, ToolCall, @@ -49,6 +51,8 @@ pub struct Agent { available_hints: Vec, post_turn_hooks: Vec>, learning_enabled: bool, + event_session_id: String, + event_channel: String, } /// A builder for creating `Agent` instances with custom configuration. @@ -70,6 +74,8 @@ pub struct AgentBuilder { available_hints: Option>, post_turn_hooks: Vec>, learning_enabled: bool, + event_session_id: Option, + event_channel: Option, } impl Default for AgentBuilder { @@ -99,6 +105,8 @@ impl AgentBuilder { available_hints: None, post_turn_hooks: Vec::new(), learning_enabled: false, + event_session_id: None, + event_channel: None, } } @@ -210,6 +218,16 @@ impl AgentBuilder { self } + pub fn event_context( + mut self, + session_id: impl Into, + channel: impl Into, + ) -> Self { + self.event_session_id = Some(session_id.into()); + self.event_channel = Some(channel.into()); + self + } + /// Validates the configuration and builds the `Agent` instance. pub fn build(self) -> Result { let tools = self @@ -251,11 +269,95 @@ impl AgentBuilder { available_hints: self.available_hints.unwrap_or_default(), post_turn_hooks: self.post_turn_hooks, learning_enabled: self.learning_enabled, + event_session_id: self + .event_session_id + .unwrap_or_else(|| "standalone".to_string()), + event_channel: self.event_channel.unwrap_or_else(|| "internal".to_string()), }) } } impl Agent { + const EVENT_ERROR_MAX_CHARS: usize = 256; + + fn event_session_id(&self) -> &str { + &self.event_session_id + } + + fn event_channel(&self) -> &str { + &self.event_channel + } + + fn count_iterations(messages: &[ConversationMessage]) -> usize { + messages + .iter() + .filter(|message| matches!(message, ConversationMessage::AssistantToolCalls { .. })) + .count() + + 1 + } + + fn conversation_message_eq(left: &ConversationMessage, right: &ConversationMessage) -> bool { + serde_json::to_string(left).ok() == serde_json::to_string(right).ok() + } + + fn message_slice_eq(left: &[ConversationMessage], right: &[ConversationMessage]) -> bool { + left.len() == right.len() + && left + .iter() + .zip(right.iter()) + .all(|(left, right)| Self::conversation_message_eq(left, right)) + } + + fn new_entries_for_turn<'a>( + history_snapshot: &[ConversationMessage], + current_history: &'a [ConversationMessage], + ) -> &'a [ConversationMessage] { + let common_prefix_len = history_snapshot + .iter() + .zip(current_history.iter()) + .take_while(|(left, right)| Self::conversation_message_eq(left, right)) + .count(); + + if common_prefix_len == history_snapshot.len() { + return ¤t_history[common_prefix_len..]; + } + + let max_overlap = history_snapshot.len().min(current_history.len()); + for overlap in (0..=max_overlap).rev() { + let snapshot_suffix = &history_snapshot[history_snapshot.len() - overlap..]; + let current_prefix = ¤t_history[..overlap]; + if Self::message_slice_eq(snapshot_suffix, current_prefix) { + return ¤t_history[overlap..]; + } + } + + current_history + } + + fn sanitize_event_error_message(err: &anyhow::Error) -> String { + let kind = match err.downcast_ref::() { + Some(AgentError::ProviderError { .. }) => Some("provider_error"), + Some(AgentError::ContextLimitExceeded { .. }) => Some("context_limit_exceeded"), + Some(AgentError::ToolExecutionError { .. }) => Some("tool_execution_error"), + Some(AgentError::CostBudgetExceeded { .. }) => Some("cost_budget_exceeded"), + Some(AgentError::MaxIterationsExceeded { .. }) => Some("max_iterations_exceeded"), + Some(AgentError::CompactionFailed { .. }) => Some("compaction_failed"), + Some(AgentError::PermissionDenied { .. }) => Some("permission_denied"), + Some(AgentError::Other(_)) | None => None, + }; + + if let Some(kind) = kind { + return kind.to_string(); + } + + let scrubbed = providers::sanitize_api_error(&err.to_string()) + .replace(['\n', '\r', '\t'], " ") + .split_whitespace() + .collect::>() + .join(" "); + truncate_with_ellipsis(&scrubbed, Self::EVENT_ERROR_MAX_CHARS) + } + /// Injects unique IDs into tool calls that are missing them. /// /// This is necessary for some tool dispatchers to correctly track and @@ -309,6 +411,11 @@ impl Agent { &self.history } + pub fn set_event_context(&mut self, session_id: impl Into, channel: impl Into) { + self.event_session_id = session_id.into(); + self.event_channel = channel.into(); + } + /// Clears the agent's conversation history. pub fn clear_history(&mut self) { self.history.clear(); @@ -633,6 +740,10 @@ impl Agent { call: &ParsedToolCall, ) -> (ToolExecutionResult, ToolCallRecord) { let started = std::time::Instant::now(); + publish_global(DomainEvent::ToolExecutionStarted { + tool_name: call.name.clone(), + session_id: self.event_session_id().to_string(), + }); log::info!("[agent_loop] tool start name={}", call.name); let (result, success) = if let Some(tool) = self.tools.iter().find(|t| t.name() == call.name) { @@ -650,6 +761,12 @@ impl Agent { (format!("Unknown tool: {}", call.name), false) }; let elapsed_ms = started.elapsed().as_millis() as u64; + publish_global(DomainEvent::ToolExecutionCompleted { + tool_name: call.name.clone(), + session_id: self.event_session_id().to_string(), + success, + elapsed_ms, + }); log::info!( "[agent_loop] tool finish name={} elapsed_ms={} output_chars={} success={}", call.name, @@ -930,7 +1047,32 @@ impl Agent { /// Runs a single turn with the given message and returns the response. pub async fn run_single(&mut self, message: &str) -> Result { - self.turn(message).await + let history_snapshot = self.history.clone(); + publish_global(DomainEvent::AgentTurnStarted { + session_id: self.event_session_id().to_string(), + channel: self.event_channel().to_string(), + }); + + match self.turn(message).await { + Ok(response) => { + let new_entries = Self::new_entries_for_turn(&history_snapshot, &self.history); + publish_global(DomainEvent::AgentTurnCompleted { + session_id: self.event_session_id().to_string(), + text_chars: response.chars().count(), + iterations: Self::count_iterations(new_entries), + }); + Ok(response) + } + Err(err) => { + let sanitized_message = Self::sanitize_event_error_message(&err); + publish_global(DomainEvent::AgentError { + session_id: self.event_session_id().to_string(), + message: sanitized_message, + recoverable: false, + }); + Err(err) + } + } } /// Runs an interactive CLI loop, reading from standard input and printing to standard output. diff --git a/src/openhuman/agent/events.rs b/src/openhuman/agent/events.rs deleted file mode 100644 index 86a8c8231..000000000 --- a/src/openhuman/agent/events.rs +++ /dev/null @@ -1,216 +0,0 @@ -//! Typed event system for agent loop observability. -//! -//! Replaces the basic `ToolEventObserver` with a comprehensive `AgentEvent` -//! enum broadcast via `tokio::sync::broadcast`. Multiple consumers (Socket.IO -//! relay, logging, cost tracking) can subscribe to the same event stream. - -use crate::openhuman::providers::UsageInfo; - -/// Events emitted during agent loop execution. -/// -/// Subscribers receive these via `tokio::sync::broadcast::Receiver`. -#[derive(Debug, Clone)] -pub enum AgentEvent { - /// An LLM inference call is about to be made. - InferenceStart { - iteration: usize, - message_count: usize, - }, - - /// An LLM inference call completed. - InferenceComplete { - iteration: usize, - has_tool_calls: bool, - usage: Option, - }, - - /// Tool calls were parsed from the LLM response. - ToolCallsParsed { - tool_names: Vec, - /// Full arguments per tool call (parallel with tool_names). - tool_arguments: Vec, - /// Optional tool_call_id per call (parallel with tool_names). - tool_call_ids: Vec>, - iteration: usize, - }, - - /// A single tool execution is starting. - ToolExecutionStart { name: String, iteration: usize }, - - /// A single tool execution completed. - ToolExecutionComplete { - name: String, - /// The actual tool output string. - output: String, - output_chars: usize, - elapsed_ms: u64, - success: bool, - tool_call_id: Option, - iteration: usize, - }, - - /// Context compaction was triggered. - CompactionTriggered { - messages_before: usize, - messages_after: usize, - }, - - /// Context compaction failed. - CompactionFailed { - error: String, - consecutive_failures: u8, - }, - - /// The agent turn completed with a final text response. - TurnComplete { - text_chars: usize, - total_iterations: usize, - }, - - /// An error occurred during the agent loop. - Error { message: String, recoverable: bool }, - - /// Cost update after an inference call. - CostUpdate { - total_input_tokens: u64, - total_output_tokens: u64, - total_cost_microdollars: u64, - }, -} - -/// Convenience sender wrapper that silently drops events if no receivers are listening. -#[derive(Debug, Clone)] -pub struct EventSender { - tx: tokio::sync::broadcast::Sender, -} - -impl EventSender { - /// Create a new event sender with the given channel capacity. - /// Capacity is clamped to at least 1 to avoid a broadcast channel panic. - pub fn new(capacity: usize) -> (Self, tokio::sync::broadcast::Receiver) { - let cap = capacity.max(1); - let (tx, rx) = tokio::sync::broadcast::channel(cap); - (Self { tx }, rx) - } - - /// Emit an event. Silently drops if no receivers are listening. - pub fn emit(&self, event: AgentEvent) { - tracing::trace!( - event = ?std::mem::discriminant(&event), - receivers = self.tx.receiver_count(), - "[agent_events] emitting event" - ); - let _ = self.tx.send(event); - } - - /// Subscribe to the event stream. - pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver { - self.tx.subscribe() - } -} - -/// Default broadcast channel capacity for agent events. -pub const DEFAULT_EVENT_CHANNEL_CAPACITY: usize = 128; - -/// Bridge adapter that converts `AgentEvent`s into `ToolEventObserver` callbacks, -/// allowing gradual migration from the old observer pattern. -pub struct ObserverBridge { - observer: std::sync::Arc, -} - -impl ObserverBridge { - pub fn new(observer: std::sync::Arc) -> Self { - Self { observer } - } - - /// Process an event and forward to the legacy observer if applicable. - pub fn handle_event(&self, event: &AgentEvent) { - tracing::trace!( - event = ?std::mem::discriminant(event), - "[agent_events] ObserverBridge handling event" - ); - match event { - AgentEvent::ToolCallsParsed { - tool_names, - tool_arguments, - tool_call_ids, - iteration, - } => { - let calls: Vec = tool_names - .iter() - .enumerate() - .map(|(i, name)| super::dispatcher::ParsedToolCall { - name: name.clone(), - arguments: tool_arguments - .get(i) - .cloned() - .unwrap_or(serde_json::Value::Null), - tool_call_id: tool_call_ids.get(i).cloned().flatten(), - }) - .collect(); - self.observer.on_tool_calls(&calls, *iteration as u32); - } - AgentEvent::ToolExecutionComplete { - name, - output, - success, - tool_call_id, - iteration, - .. - } => { - let results = vec![super::dispatcher::ToolExecutionResult { - name: name.clone(), - output: output.clone(), - success: *success, - tool_call_id: tool_call_id.clone(), - }]; - self.observer.on_tool_results(&results, *iteration as u32); - } - _ => {} // Other events have no legacy equivalent - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn event_sender_works_without_receivers() { - let (sender, _rx) = EventSender::new(16); - // Should not panic even with no active receivers - drop(_rx); - sender.emit(AgentEvent::TurnComplete { - text_chars: 100, - total_iterations: 1, - }); - } - - #[test] - fn event_sender_delivers_to_subscriber() { - let (sender, mut rx) = EventSender::new(16); - sender.emit(AgentEvent::InferenceStart { - iteration: 1, - message_count: 5, - }); - let event = rx.try_recv().unwrap(); - assert!(matches!( - event, - AgentEvent::InferenceStart { iteration: 1, .. } - )); - } - - #[test] - fn multiple_subscribers_receive_events() { - let (sender, mut rx1) = EventSender::new(16); - let mut rx2 = sender.subscribe(); - - sender.emit(AgentEvent::TurnComplete { - text_chars: 42, - total_iterations: 2, - }); - - assert!(rx1.try_recv().is_ok()); - assert!(rx2.try_recv().is_ok()); - } -} diff --git a/src/openhuman/agent/mod.rs b/src/openhuman/agent/mod.rs index 2ca3b4cbe..0bbfc5825 100644 --- a/src/openhuman/agent/mod.rs +++ b/src/openhuman/agent/mod.rs @@ -4,7 +4,6 @@ pub mod classifier; pub mod cost; pub mod dispatcher; pub mod error; -pub mod events; pub mod harness; pub mod hooks; pub mod host_runtime; @@ -12,7 +11,6 @@ pub mod identity; pub mod loop_; pub mod memory_loader; pub mod multimodal; -pub mod observer; pub mod prompt; mod schemas; pub mod traits; diff --git a/src/openhuman/agent/observer.rs b/src/openhuman/agent/observer.rs deleted file mode 100644 index 047b08152..000000000 --- a/src/openhuman/agent/observer.rs +++ /dev/null @@ -1,22 +0,0 @@ -//! Legacy observer trait for tool events. -//! -//! Deprecated: prefer the typed `AgentEvent` system in `events.rs`. -//! This module is kept for backward compatibility; use `events::ObserverBridge` -//! to connect legacy observers to the new event stream. - -use super::dispatcher::{ParsedToolCall, ToolExecutionResult}; - -/// Observer for tool events emitted during the agent loop. -/// -/// Implementors receive callbacks as tool calls are parsed and executed, -/// enabling real-time event publishing (e.g. to Socket.IO) rather than -/// batch-publishing after the entire loop completes. -/// -/// **Deprecated**: Use `AgentEvent` broadcast channel from `events.rs` instead. -pub trait ToolEventObserver: Send + Sync { - /// Called after tool calls are parsed from the LLM response, before execution. - fn on_tool_calls(&self, calls: &[ParsedToolCall], round: u32); - - /// Called after all tool calls in a round have been executed. - fn on_tool_results(&self, results: &[ToolExecutionResult], round: u32); -} diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs index 765dad4ff..482687d93 100644 --- a/src/openhuman/channels/providers/web.rs +++ b/src/openhuman/channels/providers/web.rs @@ -51,6 +51,14 @@ fn key_for(client_id: &str, thread_id: &str) -> String { format!("{client_id}::{thread_id}") } +fn event_session_id_for(client_id: &str, thread_id: &str) -> String { + json!({ + "client_id": client_id, + "thread_id": thread_id, + }) + .to_string() +} + pub async fn start_chat( client_id: &str, thread_id: &str, @@ -245,7 +253,13 @@ async fn run_chat_task( { entry.agent } - Some(_) | None => build_session_agent(&config, model_override.clone(), temperature)?, + Some(_) | None => build_session_agent( + &config, + client_id, + thread_id, + model_override.clone(), + temperature, + )?, }; let history_before = agent.history().len(); @@ -379,6 +393,8 @@ fn normalize_model_override(model_override: Option) -> Option { fn build_session_agent( config: &Config, + client_id: &str, + thread_id: &str, model_override: Option, temperature: Option, ) -> Result { @@ -390,7 +406,12 @@ fn build_session_agent( effective.default_temperature = temp; } - Agent::from_config(&effective).map_err(|e| e.to_string()) + Agent::from_config(&effective) + .map(|mut agent| { + agent.set_event_context(event_session_id_for(client_id, thread_id), "web_channel"); + agent + }) + .map_err(|e| e.to_string()) } #[derive(Debug, Deserialize)] diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index 825eaefe1..600cb3256 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -44,6 +44,8 @@ pub async fn start_channels(config: Config) -> Result<()> { // subscriber for debug logging of all domain events. let bus = event_bus::init_global(DEFAULT_CAPACITY); let _tracing_handle = bus.subscribe(Arc::new(TracingSubscriber)); + crate::openhuman::health::bus::register_health_subscriber(); + crate::openhuman::skills::bus::register_skill_cleanup_subscriber(); tracing::debug!("[event_bus] global singleton initialized in start_channels"); // Note: WebhookRequestSubscriber and ChannelInboundSubscriber are registered // in bootstrap_skill_runtime() (src/core/jsonrpc.rs) to avoid double-registration @@ -386,7 +388,6 @@ pub async fn start_channels(config: Config) -> Result<()> { println!(" Listening for messages... (Ctrl+C to stop)"); println!(); - crate::openhuman::health::mark_component_ok("channels"); event_bus::publish_global(DomainEvent::SystemStartup { component: "channels".into(), }); diff --git a/src/openhuman/channels/runtime/supervision.rs b/src/openhuman/channels/runtime/supervision.rs index c974d430c..36101c57b 100644 --- a/src/openhuman/channels/runtime/supervision.rs +++ b/src/openhuman/channels/runtime/supervision.rs @@ -15,13 +15,17 @@ pub(crate) fn spawn_supervised_listener( initial_backoff_secs: u64, max_backoff_secs: u64, ) -> tokio::task::JoinHandle<()> { + // This helper is used directly in tests and isolated runtime paths, so make + // sure channel health events always have a live bus + subscriber target. + crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY); + crate::openhuman::health::bus::register_health_subscriber(); + tokio::spawn(async move { let component = format!("channel:{}", ch.name()); let mut backoff = initial_backoff_secs.max(1); let max_backoff = max_backoff_secs.max(backoff); loop { - crate::openhuman::health::mark_component_ok(&component); publish_global(DomainEvent::ChannelConnected { channel: ch.name().to_string(), }); @@ -34,10 +38,6 @@ pub(crate) fn spawn_supervised_listener( match result { Ok(()) => { tracing::warn!("Channel {} exited unexpectedly; restarting", ch.name()); - crate::openhuman::health::mark_component_error( - &component, - "listener exited unexpectedly", - ); publish_global(DomainEvent::ChannelDisconnected { channel: ch.name().to_string(), reason: "exited unexpectedly".to_string(), @@ -47,7 +47,6 @@ pub(crate) fn spawn_supervised_listener( } Err(e) => { tracing::error!("Channel {} error: {e}; restarting", ch.name()); - crate::openhuman::health::mark_component_error(&component, e.to_string()); publish_global(DomainEvent::ChannelDisconnected { channel: ch.name().to_string(), reason: e.to_string(), @@ -55,7 +54,9 @@ pub(crate) fn spawn_supervised_listener( } } - crate::openhuman::health::bump_component_restart(&component); + publish_global(DomainEvent::HealthRestarted { + component: component.clone(), + }); tokio::time::sleep(Duration::from_secs(backoff)).await; // Double backoff AFTER sleeping so first error uses initial_backoff backoff = backoff.saturating_mul(2).min(max_backoff); diff --git a/src/openhuman/cron/scheduler.rs b/src/openhuman/cron/scheduler.rs index 702a5d9e5..3812705b6 100644 --- a/src/openhuman/cron/scheduler.rs +++ b/src/openhuman/cron/scheduler.rs @@ -20,6 +20,7 @@ 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); + crate::openhuman::health::bus::register_health_subscriber(); let poll_secs = config.reliability.scheduler_poll_secs.max(MIN_POLL_SECONDS); let mut interval = time::interval(Duration::from_secs(poll_secs)); @@ -28,7 +29,9 @@ pub async fn run(config: Config) -> Result<()> { &config.workspace_dir, )); - crate::openhuman::health::mark_component_ok("scheduler"); + publish_global(DomainEvent::SystemStartup { + component: "scheduler".to_string(), + }); loop { interval.tick().await; @@ -36,7 +39,11 @@ pub async fn run(config: Config) -> Result<()> { let jobs = match due_jobs(&config, Utc::now()) { Ok(jobs) => jobs, Err(e) => { - crate::openhuman::health::mark_component_error("scheduler", e.to_string()); + publish_global(DomainEvent::HealthChanged { + component: "scheduler".to_string(), + healthy: false, + message: Some(e.to_string()), + }); tracing::warn!("Scheduler query failed: {e}"); continue; } @@ -95,12 +102,19 @@ async fn process_due_jobs(config: &Config, security: &Arc, jobs: })) .buffer_unordered(max_concurrent); - while let Some((job_id, success)) = in_flight.next().await { - if !success { - crate::openhuman::health::mark_component_error( - "scheduler", - format!("job {job_id} failed"), - ); + while let Some((job_id, success, failure_message)) = in_flight.next().await { + if success { + publish_global(DomainEvent::HealthChanged { + component: "scheduler".to_string(), + healthy: true, + message: None, + }); + } else { + publish_global(DomainEvent::HealthChanged { + component: "scheduler".to_string(), + healthy: false, + message: Some(failure_message.unwrap_or_else(|| format!("job {job_id} failed"))), + }); } } } @@ -109,16 +123,17 @@ async fn execute_and_persist_job( config: &Config, security: &SecurityPolicy, job: &CronJob, -) -> (String, bool) { - crate::openhuman::health::mark_component_ok("scheduler"); +) -> (String, bool, Option) { warn_if_high_frequency_agent_job(job); let started_at = Utc::now(); let (success, output) = execute_job_with_retry(config, security, job).await; let finished_at = Utc::now(); let success = persist_job_result(config, job, success, &output, started_at, finished_at).await; + let failure_message = + (!success).then(|| crate::openhuman::util::truncate_with_ellipsis(&output, 256)); - (job.id.clone(), success) + (job.id.clone(), success, failure_message) } async fn run_agent_job(config: &Config, job: &CronJob) -> (bool, String) { @@ -702,7 +717,7 @@ mod tests { #[tokio::test] async fn deliver_if_configured_publishes_event_for_announce_mode() { - use crate::openhuman::event_bus::{self, DomainEvent, EventHandler}; + use crate::openhuman::event_bus::{DomainEvent, EventHandler}; use std::sync::atomic::{AtomicUsize, Ordering}; // Create an isolated bus for this test. diff --git a/src/openhuman/event_bus/events.rs b/src/openhuman/event_bus/events.rs index 9da997ed6..04eed9538 100644 --- a/src/openhuman/event_bus/events.rs +++ b/src/openhuman/event_bus/events.rs @@ -181,7 +181,13 @@ pub enum DomainEvent { /// A restart of the current core process was requested. SystemRestartRequested { source: String, reason: String }, /// A component's health status changed. - HealthChanged { component: String, healthy: bool }, + HealthChanged { + component: String, + healthy: bool, + message: Option, + }, + /// A component restart was observed. + HealthRestarted { component: String }, } impl DomainEvent { @@ -226,7 +232,8 @@ impl DomainEvent { Self::SystemStartup { .. } | Self::SystemShutdown { .. } | Self::SystemRestartRequested { .. } - | Self::HealthChanged { .. } => "system", + | Self::HealthChanged { .. } + | Self::HealthRestarted { .. } => "system", } } } @@ -517,6 +524,13 @@ mod tests { DomainEvent::HealthChanged { component: "c".into(), healthy: true, + message: None, + }, + "system", + ), + ( + DomainEvent::HealthRestarted { + component: "c".into(), }, "system", ), diff --git a/src/openhuman/health/bus.rs b/src/openhuman/health/bus.rs new file mode 100644 index 000000000..830c45bfa --- /dev/null +++ b/src/openhuman/health/bus.rs @@ -0,0 +1,115 @@ +use std::sync::{Arc, OnceLock}; + +use async_trait::async_trait; + +use crate::openhuman::event_bus::{DomainEvent, EventHandler, SubscriptionHandle}; + +static HEALTH_HANDLE: OnceLock = OnceLock::new(); + +/// Register the health subscriber on the global event bus. +pub fn register_health_subscriber() { + if HEALTH_HANDLE.get().is_some() { + return; + } + + match crate::openhuman::event_bus::subscribe_global(Arc::new(HealthSubscriber)) { + Some(handle) => { + let _ = HEALTH_HANDLE.set(handle); + } + None => { + log::warn!("[event_bus] failed to register health subscriber — bus not initialized"); + } + } +} + +pub struct HealthSubscriber; + +#[async_trait] +impl EventHandler for HealthSubscriber { + fn name(&self) -> &str { + "health::registry" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["system", "channel"]) + } + + async fn handle(&self, event: &DomainEvent) { + match event { + DomainEvent::SystemStartup { component } => { + crate::openhuman::health::mark_component_ok(component); + } + DomainEvent::HealthChanged { + component, + healthy, + message, + } => { + if *healthy { + crate::openhuman::health::mark_component_ok(component); + } else { + crate::openhuman::health::mark_component_error( + component, + message.as_deref().unwrap_or("unknown health error"), + ); + } + } + DomainEvent::HealthRestarted { component } => { + crate::openhuman::health::bump_component_restart(component); + } + DomainEvent::ChannelConnected { channel } => { + crate::openhuman::health::mark_component_ok(&format!("channel:{channel}")); + } + DomainEvent::ChannelDisconnected { channel, reason } => { + crate::openhuman::health::mark_component_error( + &format!("channel:{channel}"), + reason, + ); + } + _ => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn unique_component(prefix: &str) -> String { + format!("{prefix}-{}", uuid::Uuid::new_v4()) + } + + #[tokio::test] + async fn health_changed_false_records_error() { + let component = unique_component("health-bus-error"); + let sub = HealthSubscriber; + sub.handle(&DomainEvent::HealthChanged { + component: component.clone(), + healthy: false, + message: Some("boom".into()), + }) + .await; + + let snapshot = crate::openhuman::health::snapshot(); + let entry = snapshot.components.get(&component).unwrap(); + assert_eq!(entry.status, "error"); + assert_eq!(entry.last_error.as_deref(), Some("boom")); + } + + #[tokio::test] + async fn channel_disconnected_marks_channel_component_error() { + let channel = format!("health-bus-channel-{}", uuid::Uuid::new_v4()); + let sub = HealthSubscriber; + sub.handle(&DomainEvent::ChannelDisconnected { + channel: channel.clone(), + reason: "offline".into(), + }) + .await; + + let snapshot = crate::openhuman::health::snapshot(); + let entry = snapshot + .components + .get(&format!("channel:{channel}")) + .unwrap(); + assert_eq!(entry.status, "error"); + } +} diff --git a/src/openhuman/health/mod.rs b/src/openhuman/health/mod.rs index 926f76196..3990cce26 100644 --- a/src/openhuman/health/mod.rs +++ b/src/openhuman/health/mod.rs @@ -1,3 +1,4 @@ +pub mod bus; mod core; pub mod ops; mod schemas; diff --git a/src/openhuman/skills/bus.rs b/src/openhuman/skills/bus.rs index cc1119597..032ad5211 100644 --- a/src/openhuman/skills/bus.rs +++ b/src/openhuman/skills/bus.rs @@ -1,8 +1,79 @@ //! Event bus handlers for the skills domain. //! -//! Placeholder for future cross-domain subscribers that react to skill lifecycle -//! events (e.g. cascading restarts, dependency tracking, metrics collection). -//! -//! Skill events are currently consumed by the [`TracingSubscriber`] for -//! observability. Add domain-specific handlers here as needed following the -//! pattern in `crate::openhuman::cron::bus`. +//! Keeps skill lifecycle side effects behind the message bus so callers can +//! stop skills without importing cron/webhook cleanup concerns directly. + +use std::sync::{Arc, OnceLock}; + +use async_trait::async_trait; + +use crate::openhuman::event_bus::{DomainEvent, EventHandler, SubscriptionHandle}; + +static SKILL_CLEANUP_HANDLE: OnceLock = OnceLock::new(); + +/// Register the long-lived skill cleanup subscriber on the global event bus. +pub fn register_skill_cleanup_subscriber() { + if SKILL_CLEANUP_HANDLE.get().is_some() { + return; + } + + match crate::openhuman::event_bus::subscribe_global(Arc::new(SkillCleanupSubscriber)) { + Some(handle) => { + let _ = SKILL_CLEANUP_HANDLE.set(handle); + } + None => { + log::warn!( + "[event_bus] failed to register skill cleanup subscriber — bus not initialized" + ); + } + } +} + +pub struct SkillCleanupSubscriber; + +#[async_trait] +impl EventHandler for SkillCleanupSubscriber { + fn name(&self) -> &str { + "skill::cleanup" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["skill"]) + } + + async fn handle(&self, event: &DomainEvent) { + let DomainEvent::SkillStopped { skill_id } = event else { + return; + }; + + let Some(engine) = crate::openhuman::skills::global_engine() else { + log::warn!( + "[skills:bus] received SkillStopped for '{}' but runtime engine is unavailable", + skill_id + ); + return; + }; + + log::debug!( + "[skills:bus] handling SkillStopped cleanup for '{}'", + skill_id + ); + engine.cron_scheduler().unregister_all_for_skill(skill_id); + engine.webhook_router().unregister_skill(skill_id); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn ignores_non_skill_stop_events() { + let sub = SkillCleanupSubscriber; + sub.handle(&DomainEvent::SkillLoaded { + skill_id: "gmail".into(), + runtime: "qjs".into(), + }) + .await; + } +} diff --git a/src/openhuman/skills/qjs_engine.rs b/src/openhuman/skills/qjs_engine.rs index ee3c1b4f1..d1e7ede92 100644 --- a/src/openhuman/skills/qjs_engine.rs +++ b/src/openhuman/skills/qjs_engine.rs @@ -519,8 +519,6 @@ impl RuntimeEngine { /// and the webhook router, and updates its status. pub async fn stop_skill(&self, skill_id: &str) -> Result<(), String> { self.registry.stop_skill(skill_id).await?; - self.cron_scheduler.unregister_all_for_skill(skill_id); - self.webhook_router.unregister_skill(skill_id); self.emit_status_change(skill_id); publish_global(DomainEvent::SkillStopped { skill_id: skill_id.to_string(), diff --git a/src/openhuman/update/scheduler.rs b/src/openhuman/update/scheduler.rs index 5e80c8a75..5f071e69c 100644 --- a/src/openhuman/update/scheduler.rs +++ b/src/openhuman/update/scheduler.rs @@ -7,6 +7,7 @@ use std::time::Duration; use crate::openhuman::config::UpdateConfig; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use crate::openhuman::update::core as update_core; /// Minimum allowed interval to avoid hammering the GitHub API. @@ -20,6 +21,12 @@ pub async fn run(config: UpdateConfig) { return; } + crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY); + crate::openhuman::health::bus::register_health_subscriber(); + publish_global(DomainEvent::SystemStartup { + component: "update_checker".to_string(), + }); + let interval_mins = config.interval_minutes.max(MIN_INTERVAL_MINUTES); let interval = Duration::from_secs(u64::from(interval_mins) * 60); @@ -49,19 +56,26 @@ async fn tick() { info.latest_version, info.download_url.as_deref().unwrap_or("(no asset)") ); - crate::openhuman::health::mark_component_ok("update_checker"); } else { log::info!( "[update:scheduler] up to date (current: {}, latest: {})", info.current_version, info.latest_version ); - crate::openhuman::health::mark_component_ok("update_checker"); } + publish_global(DomainEvent::HealthChanged { + component: "update_checker".to_string(), + healthy: true, + message: None, + }); } Err(e) => { log::warn!("[update:scheduler] update check failed: {e}"); - crate::openhuman::health::mark_component_error("update_checker", &e); + publish_global(DomainEvent::HealthChanged { + component: "update_checker".to_string(), + healthy: false, + message: Some(e.to_string()), + }); } } }