mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
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.
This commit is contained in:
Steven Enamakel
@@ -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 |
|
||||
|
||||
---
|
||||
|
||||
+6
-1
@@ -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"
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -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<String>,
|
||||
post_turn_hooks: Vec<Arc<dyn PostTurnHook>>,
|
||||
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<Vec<String>>,
|
||||
post_turn_hooks: Vec<Arc<dyn PostTurnHook>>,
|
||||
learning_enabled: bool,
|
||||
event_session_id: Option<String>,
|
||||
event_channel: Option<String>,
|
||||
}
|
||||
|
||||
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<String>,
|
||||
channel: impl Into<String>,
|
||||
) -> 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<Agent> {
|
||||
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::<AgentError>() {
|
||||
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::<Vec<_>>()
|
||||
.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<String>, channel: impl Into<String>) {
|
||||
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<String> {
|
||||
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.
|
||||
|
||||
@@ -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<AgentEvent>`.
|
||||
#[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<UsageInfo>,
|
||||
},
|
||||
|
||||
/// Tool calls were parsed from the LLM response.
|
||||
ToolCallsParsed {
|
||||
tool_names: Vec<String>,
|
||||
/// Full arguments per tool call (parallel with tool_names).
|
||||
tool_arguments: Vec<serde_json::Value>,
|
||||
/// Optional tool_call_id per call (parallel with tool_names).
|
||||
tool_call_ids: Vec<Option<String>>,
|
||||
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<String>,
|
||||
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<AgentEvent>,
|
||||
}
|
||||
|
||||
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<AgentEvent>) {
|
||||
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<AgentEvent> {
|
||||
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<dyn super::observer::ToolEventObserver>,
|
||||
}
|
||||
|
||||
impl ObserverBridge {
|
||||
pub fn new(observer: std::sync::Arc<dyn super::observer::ToolEventObserver>) -> 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<super::dispatcher::ParsedToolCall> = 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());
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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<String>) -> Option<String> {
|
||||
|
||||
fn build_session_agent(
|
||||
config: &Config,
|
||||
client_id: &str,
|
||||
thread_id: &str,
|
||||
model_override: Option<String>,
|
||||
temperature: Option<f64>,
|
||||
) -> Result<Agent, String> {
|
||||
@@ -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)]
|
||||
|
||||
@@ -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(),
|
||||
});
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<SecurityPolicy>, 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<String>) {
|
||||
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.
|
||||
|
||||
@@ -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<String>,
|
||||
},
|
||||
/// 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",
|
||||
),
|
||||
|
||||
@@ -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<SubscriptionHandle> = 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");
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
pub mod bus;
|
||||
mod core;
|
||||
pub mod ops;
|
||||
mod schemas;
|
||||
|
||||
@@ -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<SubscriptionHandle> = 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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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()),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user