From 07b1df4f24a01c1be175cb66e92c600cf5ce09ef Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Mon, 6 Apr 2026 15:58:50 -0700 Subject: [PATCH] feat(event_bus): wire webhooks, channels & skills through the event bus (#379) * feat(event_bus): enhance domain event handling across modules - Added new `DomainEvent` variants for channel and skill events, including `ChannelMessageReceived`, `ChannelMessageProcessed`, `ChannelConnected`, `ChannelDisconnected`, `SkillLoaded`, `SkillStopped`, and `SkillStartFailed`. - Implemented event publishing in the channels and skills modules to track message processing and skill lifecycle events. - Created dedicated event bus handler files for the skills and webhooks domains, preparing for future subscriber implementations. - Updated documentation in `CLAUDE.md` to reflect the new domain events and their usage. These changes improve the observability and modularity of the system by leveraging an event-driven architecture for cross-module communication. * feat(event_bus): implement channel and webhook event handling - Introduced `ChannelInboundSubscriber` to handle inbound channel messages, triggering the agent inference loop and sending replies via the backend REST API. - Added `WebhookRequestSubscriber` to manage incoming webhook requests, routing them to the appropriate skill and handling responses. - Updated the global event bus initialization in `bootstrap_skill_runtime` to register both channel and webhook subscribers. - Enhanced `DomainEvent` with new variants for channel inbound messages and webhook requests, improving event-driven communication across modules. These changes enhance the modularity and responsiveness of the system by leveraging an event-driven architecture for channel and webhook interactions. * refactor(event_bus): update domain event documentation and subscriber initialization - Revised the documentation in `CLAUDE.md` to provide a concise overview of domain events and their associated subscriber files, enhancing clarity for future development. - Updated the `start_channels` function to initialize `WebhookRequestSubscriber` and `ChannelInboundSubscriber`, ensuring proper event handling for webhooks and channel messages. - Streamlined the event bus subscriber registration process, reinforcing the modular architecture of the system. These changes improve the maintainability and usability of the event bus framework, facilitating better cross-module communication. * style: apply cargo fmt formatting Co-Authored-By: Claude Opus 4.6 (1M context) * fix(event_bus): remove duplicate subscriber registration in start_channels WebhookRequestSubscriber and ChannelInboundSubscriber were registered in both bootstrap_skill_runtime() and start_channels(), causing events to be handled twice when both paths run in the same process. Co-Authored-By: Claude Opus 4.6 (1M context) * fix(event_bus): prevent subscriber handles from being dropped on function exit SubscriptionHandle::drop aborts the background task. Since bootstrap_skill_runtime() returns immediately after setup, the local handles were dropped, cancelling both subscribers. Use std::mem::forget to leak the handles so the tasks live for the entire process. Co-Authored-By: Claude Opus 4.6 (1M context) * fix(event_bus): ensure subscriber handles persist beyond function exit Modified the handling of subscriber registration to prevent premature dropping of handles in `bootstrap_skill_runtime()`. This change ensures that the background tasks for subscribers remain active for the entire process lifecycle, enhancing event handling reliability. Co-Authored-By: Claude Opus 4.6 (1M context) * fix(webhooks): use proper JSON serialization for error response bodies Hand-escaped JSON strings only handled double quotes, not backslashes, newlines, or other control chars. Replaced with serde_json serialization via an error_body() helper. Co-Authored-By: Claude Opus 4.6 (1M context) * fix(webhooks): use proper JSON serialization for error response bodies Hand-escaped JSON strings only handled double quotes, not backslashes, newlines, or other control chars. Replaced with serde_json serialization via an error_body() helper. Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- CLAUDE.md | 8 + src/core/jsonrpc.rs | 16 + src/openhuman/channels/bus.rs | 179 +++++++ src/openhuman/channels/mod.rs | 1 + src/openhuman/channels/runtime/dispatch.rs | 48 +- src/openhuman/channels/runtime/startup.rs | 3 + src/openhuman/channels/runtime/supervision.rs | 12 + src/openhuman/event_bus/events.rs | 335 +++++++++++- src/openhuman/skills/bus.rs | 8 + src/openhuman/skills/mod.rs | 1 + src/openhuman/skills/qjs_engine.rs | 56 +- src/openhuman/socket/event_handlers.rs | 490 ++++-------------- src/openhuman/webhooks/bus.rs | 237 +++++++++ src/openhuman/webhooks/mod.rs | 1 + src/openhuman/webhooks/router.rs | 37 +- 15 files changed, 1009 insertions(+), 423 deletions(-) create mode 100644 src/openhuman/channels/bus.rs create mode 100644 src/openhuman/skills/bus.rs create mode 100644 src/openhuman/webhooks/bus.rs diff --git a/CLAUDE.md b/CLAUDE.md index 8b71dde1e..901a33dfa 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -311,6 +311,14 @@ A typed pub/sub event bus for **decoupled cross-module communication**. The bus | `event_bus::subscribe_global(handler)` | Subscribe from anywhere (returns `None` if not yet initialized) | | `event_bus::global()` | Get `Option<&'static EventBus>` for advanced use | +**Domains:** `agent`, `memory`, `channel`, `cron`, `skill`, `tool`, `webhook`, `system`. See `events.rs` for the full variant list — events carry rich payloads so subscribers have everything they need. + +**Domain subscriber files** — each domain owns its `bus.rs` with `EventHandler` impls: +- `cron/bus.rs` — `CronDeliverySubscriber` (delivers job output to channels) +- `webhooks/bus.rs` — `WebhookRequestSubscriber` (routes incoming requests to skills, emits responses via socket) +- `channels/bus.rs` — `ChannelInboundSubscriber` (runs agent loop for inbound socket messages) +- `skills/bus.rs` — stub for future subscribers + **Adding events for a new domain:** 1. Add variants to `DomainEvent` in `events.rs` (prefix with domain name, e.g. `BillingInvoiceCreated { ... }`). diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 5a0536b33..1ce6f7ff0 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -689,6 +689,22 @@ pub async fn bootstrap_skill_runtime() { let _ = std::fs::create_dir_all(&workspace_dir); engine.set_workspace_dir(workspace_dir); + // --- Event bus bootstrap --- + // Ensure the global event bus is initialized (no-op if already done by start_channels). + let bus = + crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY); + // Register domain subscribers for cross-module event handling. + // Leak the handles so the background tasks live for the entire process — + // SubscriptionHandle::drop aborts the task, and bootstrap_skill_runtime() + // returns immediately after setup. + std::mem::forget(bus.subscribe(Arc::new( + crate::openhuman::webhooks::bus::WebhookRequestSubscriber::new(), + ))); + std::mem::forget(bus.subscribe(Arc::new( + crate::openhuman::channels::bus::ChannelInboundSubscriber::new(), + ))); + log::info!("[event_bus] webhook and channel subscribers registered"); + // --- Socket manager bootstrap --- let socket_mgr = Arc::new(SocketManager::new()); set_global_socket_manager(socket_mgr.clone()); diff --git a/src/openhuman/channels/bus.rs b/src/openhuman/channels/bus.rs new file mode 100644 index 000000000..086747f5f --- /dev/null +++ b/src/openhuman/channels/bus.rs @@ -0,0 +1,179 @@ +//! Event bus handlers for the channels domain. +//! +//! The [`ChannelInboundSubscriber`] handles inbound channel messages published +//! by the socket transport layer. It runs the agent inference loop via the web +//! channel provider and sends the reply back through the REST API. + +use crate::openhuman::event_bus::{DomainEvent, EventHandler}; +use async_trait::async_trait; +use serde_json::json; + +/// Subscribes to `ChannelInboundMessage` events and runs the agent loop, +/// sending replies back to the originating channel via the backend REST API. +pub struct ChannelInboundSubscriber; + +impl ChannelInboundSubscriber { + pub fn new() -> Self { + Self + } +} + +#[async_trait] +impl EventHandler for ChannelInboundSubscriber { + fn name(&self) -> &str { + "channel::inbound_handler" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["channel"]) + } + + async fn handle(&self, event: &DomainEvent) { + let DomainEvent::ChannelInboundMessage { + event_name: _, + channel, + message, + raw_data: _, + } = event + else { + return; + }; + + tracing::info!( + "[channel-inbound] received message from channel='{}' len={}", + channel, + message.len() + ); + + let thread_id = format!("channel:{}", channel); + let client_id = "inbound".to_string(); + + let mut event_rx = + crate::openhuman::channels::providers::web::subscribe_web_channel_events(); + + let request_id = match crate::openhuman::channels::providers::web::start_chat( + &client_id, &thread_id, message, None, None, + ) + .await + { + Ok(rid) => { + tracing::debug!( + "[channel-inbound] agent started request_id={} thread={}", + rid, + thread_id + ); + rid + } + Err(err) => { + tracing::error!("[channel-inbound] start_chat failed: {}", err); + send_channel_reply( + channel, + &format!("Sorry, I couldn't process your message: {err}"), + ) + .await; + return; + } + }; + + let timeout = tokio::time::Duration::from_secs(180); + let deadline = tokio::time::Instant::now() + timeout; + + loop { + tokio::select! { + event = event_rx.recv() => { + match event { + Ok(ev) if ev.request_id == request_id => { + if ev.event == "chat_done" || ev.event == "chat:done" { + let reply = ev.full_response.unwrap_or_default(); + if reply.trim().is_empty() { + tracing::warn!("[channel-inbound] agent returned empty response"); + return; + } + tracing::info!( + "[channel-inbound] agent done, replying to channel='{}' len={}", + channel, + reply.len() + ); + send_channel_reply(channel, &reply).await; + return; + } + if ev.event == "chat_error" || ev.event == "chat:error" { + let err_msg = ev.message.unwrap_or_else(|| "unknown error".to_string()); + tracing::error!("[channel-inbound] agent error: {}", err_msg); + send_channel_reply( + channel, + &format!("Sorry, I encountered an error: {err_msg}"), + ) + .await; + return; + } + } + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!("[channel-inbound] event bus lagged, skipped {} events", n); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + tracing::error!("[channel-inbound] event bus closed unexpectedly"); + return; + } + } + } + _ = tokio::time::sleep_until(deadline) => { + tracing::error!("[channel-inbound] agent timed out after {}s", timeout.as_secs()); + send_channel_reply(channel, "Sorry, the request timed out.").await; + return; + } + } + } + } +} + +/// Send a text reply back to a channel via the backend REST API. +async fn send_channel_reply(channel: &str, text: &str) { + let config = match crate::openhuman::config::rpc::load_config_with_timeout().await { + Ok(c) => c, + Err(e) => { + tracing::error!("[channel-inbound] failed to load config: {}", e); + return; + } + }; + + let api_url = crate::api::config::effective_api_url(&config.api_url); + let jwt = match crate::api::jwt::get_session_token(&config) { + Ok(Some(t)) => t, + Ok(None) => { + tracing::error!("[channel-inbound] no session JWT — cannot reply"); + return; + } + Err(e) => { + tracing::error!("[channel-inbound] failed to get session token: {}", e); + return; + } + }; + + let client = match crate::api::rest::BackendOAuthClient::new(&api_url) { + Ok(c) => c, + Err(e) => { + tracing::error!("[channel-inbound] failed to create API client: {}", e); + return; + } + }; + + let body = json!({ "text": text }); + match client.send_channel_message(channel, &jwt, body).await { + Ok(resp) => { + tracing::info!( + "[channel-inbound] reply sent to channel='{}' response={:?}", + channel, + resp + ); + } + Err(e) => { + tracing::error!( + "[channel-inbound] failed to send reply to channel='{}': {}", + channel, + e + ); + } + } +} diff --git a/src/openhuman/channels/mod.rs b/src/openhuman/channels/mod.rs index 53c7cbf9b..bb777a8a4 100644 --- a/src/openhuman/channels/mod.rs +++ b/src/openhuman/channels/mod.rs @@ -1,5 +1,6 @@ //! Channel implementations and runtime orchestration. +pub mod bus; pub mod cli; pub mod controllers; pub mod providers; diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index 970115a9d..326ea6cd9 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/src/openhuman/channels/runtime/dispatch.rs @@ -11,6 +11,7 @@ use crate::openhuman::channels::routes::{ }; use crate::openhuman::channels::traits; use crate::openhuman::channels::{Channel, SendMessage}; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use crate::openhuman::providers::{self, ChatMessage}; use crate::openhuman::util::truncate_with_ellipsis; use std::sync::Arc; @@ -73,6 +74,13 @@ pub(crate) async fn process_channel_message( truncate_with_ellipsis(&msg.content, 80) ); + publish_global(DomainEvent::ChannelMessageReceived { + channel: msg.channel.clone(), + sender: msg.sender.clone(), + content: msg.content.clone(), + thread_ts: msg.thread_ts.clone(), + }); + let target_channel = ctx.channels_by_name.get(&msg.channel).cloned(); if handle_runtime_command_if_needed(ctx.as_ref(), &msg, target_channel.as_ref()).await { return; @@ -243,7 +251,7 @@ pub(crate) async fn process_channel_message( log_worker_join_result(handle.await); } - match llm_result { + let (success, response_text) = match llm_result { Ok(Ok(response)) => { // Save user + assistant turn to per-sender history { @@ -280,7 +288,7 @@ pub(crate) async fn process_channel_message( } } else if let Err(e) = channel .send( - &SendMessage::new(response, &msg.reply_target) + &SendMessage::new(&response, &msg.reply_target) .in_thread(msg.thread_ts.clone()), ) .await @@ -288,6 +296,7 @@ pub(crate) async fn process_channel_message( eprintln!(" ❌ Failed to reply on {}: {e}", channel.name()); } } + (true, response) } Ok(Err(e)) => { if is_context_window_overflow_error(&e) { @@ -316,9 +325,19 @@ pub(crate) async fn process_channel_message( .await; } } + + publish_global(DomainEvent::ChannelMessageProcessed { + channel: msg.channel.clone(), + sender: msg.sender.clone(), + content: msg.content.clone(), + response: error_text.to_string(), + elapsed_ms: started_at.elapsed().as_millis() as u64, + success: false, + }); return; } + let error_response = format!("⚠️ Error: {e}"); eprintln!( " ❌ LLM error after {}ms: {e}", started_at.elapsed().as_millis() @@ -326,17 +345,18 @@ pub(crate) async fn process_channel_message( if let Some(channel) = target_channel.as_ref() { if let Some(ref draft_id) = draft_message_id { let _ = channel - .finalize_draft(&msg.reply_target, draft_id, &format!("⚠️ Error: {e}")) + .finalize_draft(&msg.reply_target, draft_id, &error_response) .await; } else { let _ = channel .send( - &SendMessage::new(format!("⚠️ Error: {e}"), &msg.reply_target) + &SendMessage::new(&error_response, &msg.reply_target) .in_thread(msg.thread_ts.clone()), ) .await; } } + (false, error_response) } Err(_) => { let timeout_msg = format!("LLM response timed out after {}s", ctx.message_timeout_secs); @@ -345,24 +365,34 @@ pub(crate) async fn process_channel_message( timeout_msg, started_at.elapsed().as_millis() ); + let error_text = + "⚠️ Request timed out while waiting for the model. Please try again.".to_string(); if let Some(channel) = target_channel.as_ref() { - let error_text = - "⚠️ Request timed out while waiting for the model. Please try again."; if let Some(ref draft_id) = draft_message_id { let _ = channel - .finalize_draft(&msg.reply_target, draft_id, error_text) + .finalize_draft(&msg.reply_target, draft_id, &error_text) .await; } else { let _ = channel .send( - &SendMessage::new(error_text, &msg.reply_target) + &SendMessage::new(&error_text, &msg.reply_target) .in_thread(msg.thread_ts.clone()), ) .await; } } + (false, error_text) } - } + }; + + publish_global(DomainEvent::ChannelMessageProcessed { + channel: msg.channel.clone(), + sender: msg.sender.clone(), + content: msg.content.clone(), + response: response_text, + elapsed_ms: started_at.elapsed().as_millis() as u64, + success, + }); } pub(crate) async fn run_message_dispatch_loop( diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index 98404407b..3894393c6 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -45,6 +45,9 @@ pub async fn start_channels(config: Config) -> Result<()> { 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"); + // Note: WebhookRequestSubscriber and ChannelInboundSubscriber are registered + // in bootstrap_skill_runtime() (src/core/jsonrpc.rs) to avoid double-registration + // when both startup paths run in the same process. let provider_runtime_options = providers::ProviderRuntimeOptions { auth_profile_override: None, diff --git a/src/openhuman/channels/runtime/supervision.rs b/src/openhuman/channels/runtime/supervision.rs index 85b778f17..c974d430c 100644 --- a/src/openhuman/channels/runtime/supervision.rs +++ b/src/openhuman/channels/runtime/supervision.rs @@ -5,6 +5,7 @@ use super::super::context::{ }; use super::super::traits; use super::super::Channel; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use std::sync::Arc; use std::time::Duration; @@ -21,6 +22,9 @@ pub(crate) fn spawn_supervised_listener( loop { crate::openhuman::health::mark_component_ok(&component); + publish_global(DomainEvent::ChannelConnected { + channel: ch.name().to_string(), + }); let result = ch.listen(tx.clone()).await; if tx.is_closed() { @@ -34,12 +38,20 @@ pub(crate) fn spawn_supervised_listener( &component, "listener exited unexpectedly", ); + publish_global(DomainEvent::ChannelDisconnected { + channel: ch.name().to_string(), + reason: "exited unexpectedly".to_string(), + }); // Clean exit — reset backoff since the listener ran successfully backoff = initial_backoff_secs.max(1); } 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(), + }); } } diff --git a/src/openhuman/event_bus/events.rs b/src/openhuman/event_bus/events.rs index cdd0dd5ee..f6de061ec 100644 --- a/src/openhuman/event_bus/events.rs +++ b/src/openhuman/event_bus/events.rs @@ -1,7 +1,8 @@ //! 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. +//! Events carry full payloads so subscribers have everything they need without +//! secondary lookups. The broadcast channel clones each event per subscriber, +//! which is fine — richness beats round-trips. /// Top-level domain event. Non-exhaustive so new variants can be added /// without breaking existing match arms. @@ -35,8 +36,29 @@ pub enum DomainEvent { MemoryRecalled { query: String, hit_count: usize }, // ── Channels ──────────────────────────────────────────────────────── + /// An inbound channel message from the transport layer, ready for processing. + ChannelInboundMessage { + event_name: String, + channel: String, + message: String, + raw_data: serde_json::Value, + }, /// A message was received on a channel. - ChannelMessageReceived { channel: String, sender: String }, + ChannelMessageReceived { + channel: String, + sender: String, + content: String, + thread_ts: Option, + }, + /// A channel message was fully processed (LLM response sent or error). + ChannelMessageProcessed { + channel: String, + sender: String, + content: String, + response: String, + elapsed_ms: u64, + success: bool, + }, /// A channel connected successfully. ChannelConnected { channel: String }, /// A channel disconnected. @@ -57,11 +79,17 @@ pub enum DomainEvent { // ── Skills ────────────────────────────────────────────────────────── /// A skill was loaded into the runtime. - SkillLoaded { skill_id: String }, + SkillLoaded { skill_id: String, runtime: String }, + /// A skill was stopped. + SkillStopped { skill_id: String }, + /// A skill failed to start. + SkillStartFailed { skill_id: String, error: String }, /// A skill tool was executed. SkillExecuted { skill_id: String, tool_name: String, + arguments: serde_json::Value, + result: Option, success: bool, elapsed_ms: u64, }, @@ -81,8 +109,38 @@ pub enum DomainEvent { }, // ── Webhooks ──────────────────────────────────────────────────────── + /// An incoming webhook request from the transport layer, ready for routing. + WebhookIncomingRequest { + request: crate::openhuman::webhooks::WebhookRequest, + raw_data: serde_json::Value, + }, /// A webhook was received and routed to a skill. - WebhookReceived { tunnel_id: String, skill_id: String }, + WebhookReceived { + tunnel_id: String, + skill_id: String, + method: String, + path: String, + correlation_id: String, + }, + /// A webhook tunnel was registered to a skill. + WebhookRegistered { + tunnel_id: String, + skill_id: String, + tunnel_name: Option, + }, + /// A webhook tunnel was unregistered from a skill. + WebhookUnregistered { tunnel_id: String, skill_id: String }, + /// A webhook request was fully processed (includes timing and status). + WebhookProcessed { + tunnel_id: String, + skill_id: String, + method: String, + path: String, + correlation_id: String, + status_code: u16, + elapsed_ms: u64, + error: Option, + }, // ── System lifecycle ──────────────────────────────────────────────── /// A system component started up. @@ -103,7 +161,9 @@ impl DomainEvent { Self::MemoryStored { .. } | Self::MemoryRecalled { .. } => "memory", - Self::ChannelMessageReceived { .. } + Self::ChannelInboundMessage { .. } + | Self::ChannelMessageReceived { .. } + | Self::ChannelMessageProcessed { .. } | Self::ChannelConnected { .. } | Self::ChannelDisconnected { .. } => "channel", @@ -111,11 +171,18 @@ impl DomainEvent { | Self::CronJobCompleted { .. } | Self::CronDeliveryRequested { .. } => "cron", - Self::SkillLoaded { .. } | Self::SkillExecuted { .. } => "skill", + Self::SkillLoaded { .. } + | Self::SkillStopped { .. } + | Self::SkillStartFailed { .. } + | Self::SkillExecuted { .. } => "skill", Self::ToolExecutionStarted { .. } | Self::ToolExecutionCompleted { .. } => "tool", - Self::WebhookReceived { .. } => "webhook", + Self::WebhookIncomingRequest { .. } + | Self::WebhookReceived { .. } + | Self::WebhookRegistered { .. } + | Self::WebhookUnregistered { .. } + | Self::WebhookProcessed { .. } => "webhook", Self::SystemStartup { .. } | Self::SystemShutdown { .. } @@ -123,3 +190,255 @@ impl DomainEvent { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn all_variants_have_correct_domain() { + let cases: Vec<(DomainEvent, &str)> = vec![ + // Agent + ( + DomainEvent::AgentTurnStarted { + session_id: "s".into(), + channel: "c".into(), + }, + "agent", + ), + ( + DomainEvent::AgentTurnCompleted { + session_id: "s".into(), + text_chars: 0, + iterations: 0, + }, + "agent", + ), + ( + DomainEvent::AgentError { + session_id: "s".into(), + message: "e".into(), + recoverable: false, + }, + "agent", + ), + // Memory + ( + DomainEvent::MemoryStored { + key: "k".into(), + category: "c".into(), + namespace: "n".into(), + }, + "memory", + ), + ( + DomainEvent::MemoryRecalled { + query: "q".into(), + hit_count: 0, + }, + "memory", + ), + // Channel + ( + DomainEvent::ChannelInboundMessage { + event_name: "telegram:message".into(), + channel: "telegram".into(), + message: "hi".into(), + raw_data: serde_json::Value::Null, + }, + "channel", + ), + ( + DomainEvent::ChannelMessageReceived { + channel: "c".into(), + sender: "s".into(), + content: "hi".into(), + thread_ts: None, + }, + "channel", + ), + ( + DomainEvent::ChannelMessageProcessed { + channel: "c".into(), + sender: "s".into(), + content: "hi".into(), + response: "hello".into(), + elapsed_ms: 0, + success: true, + }, + "channel", + ), + ( + DomainEvent::ChannelConnected { + channel: "c".into(), + }, + "channel", + ), + ( + DomainEvent::ChannelDisconnected { + channel: "c".into(), + reason: "r".into(), + }, + "channel", + ), + // Cron + ( + DomainEvent::CronJobTriggered { + job_id: "j".into(), + job_type: "t".into(), + }, + "cron", + ), + ( + DomainEvent::CronJobCompleted { + job_id: "j".into(), + success: true, + }, + "cron", + ), + ( + DomainEvent::CronDeliveryRequested { + job_id: "j".into(), + channel: "c".into(), + target: "t".into(), + output: "o".into(), + }, + "cron", + ), + // Skill + ( + DomainEvent::SkillLoaded { + skill_id: "s".into(), + runtime: "quickjs".into(), + }, + "skill", + ), + ( + DomainEvent::SkillStopped { + skill_id: "s".into(), + }, + "skill", + ), + ( + DomainEvent::SkillStartFailed { + skill_id: "s".into(), + error: "e".into(), + }, + "skill", + ), + ( + DomainEvent::SkillExecuted { + skill_id: "s".into(), + tool_name: "t".into(), + arguments: serde_json::Value::Null, + result: None, + success: true, + elapsed_ms: 0, + }, + "skill", + ), + // Tool + ( + DomainEvent::ToolExecutionStarted { + tool_name: "t".into(), + session_id: "s".into(), + }, + "tool", + ), + ( + DomainEvent::ToolExecutionCompleted { + tool_name: "t".into(), + session_id: "s".into(), + success: true, + elapsed_ms: 0, + }, + "tool", + ), + // Webhook + ( + DomainEvent::WebhookIncomingRequest { + request: crate::openhuman::webhooks::WebhookRequest { + correlation_id: "c".into(), + tunnel_id: "t".into(), + tunnel_uuid: "u".into(), + tunnel_name: "n".into(), + method: "GET".into(), + path: "/".into(), + headers: Default::default(), + query: Default::default(), + body: String::new(), + }, + raw_data: serde_json::Value::Null, + }, + "webhook", + ), + ( + DomainEvent::WebhookReceived { + tunnel_id: "t".into(), + skill_id: "s".into(), + method: "GET".into(), + path: "/".into(), + correlation_id: "c".into(), + }, + "webhook", + ), + ( + DomainEvent::WebhookRegistered { + tunnel_id: "t".into(), + skill_id: "s".into(), + tunnel_name: None, + }, + "webhook", + ), + ( + DomainEvent::WebhookUnregistered { + tunnel_id: "t".into(), + skill_id: "s".into(), + }, + "webhook", + ), + ( + DomainEvent::WebhookProcessed { + tunnel_id: "t".into(), + skill_id: "s".into(), + method: "GET".into(), + path: "/".into(), + correlation_id: "c".into(), + status_code: 200, + elapsed_ms: 0, + error: None, + }, + "webhook", + ), + // System + ( + DomainEvent::SystemStartup { + component: "c".into(), + }, + "system", + ), + ( + DomainEvent::SystemShutdown { + component: "c".into(), + }, + "system", + ), + ( + DomainEvent::HealthChanged { + component: "c".into(), + healthy: true, + }, + "system", + ), + ]; + + for (event, expected_domain) in cases { + assert_eq!( + event.domain(), + expected_domain, + "Wrong domain for {:?}", + std::mem::discriminant(&event) + ); + } + } +} diff --git a/src/openhuman/skills/bus.rs b/src/openhuman/skills/bus.rs new file mode 100644 index 000000000..cc1119597 --- /dev/null +++ b/src/openhuman/skills/bus.rs @@ -0,0 +1,8 @@ +//! 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`. diff --git a/src/openhuman/skills/mod.rs b/src/openhuman/skills/mod.rs index 44c40ff56..d0fcf2ce4 100644 --- a/src/openhuman/skills/mod.rs +++ b/src/openhuman/skills/mod.rs @@ -7,6 +7,7 @@ //! - Various operations for skill lifecycle management (`ops.rs`, `registry_ops.rs`) pub mod bridge; +pub mod bus; pub mod cron_scheduler; pub mod manifest; pub mod ops; diff --git a/src/openhuman/skills/qjs_engine.rs b/src/openhuman/skills/qjs_engine.rs index 5d3210527..4e754e0f1 100644 --- a/src/openhuman/skills/qjs_engine.rs +++ b/src/openhuman/skills/qjs_engine.rs @@ -30,6 +30,7 @@ pub fn require_engine() -> Result, String> { global_engine().ok_or_else(|| "skill runtime not initialized".to_string()) } +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use crate::openhuman::memory::{MemoryClient, MemoryClientRef}; use crate::openhuman::skills::cron_scheduler::CronScheduler; use crate::openhuman::skills::manifest::SkillManifest; @@ -399,6 +400,7 @@ impl RuntimeEngine { )); } + let runtime_name = manifest.runtime.clone(); let config = manifest.to_config(); let data_dir = self.skills_data_dir.join(skill_id); @@ -452,7 +454,10 @@ impl RuntimeEngine { match current_status { SkillStatus::Running => { self.emit_status_change(&skill_id_owned); - + publish_global(DomainEvent::SkillLoaded { + skill_id: skill_id_owned.clone(), + runtime: runtime_name.clone(), + }); return Ok(instance.snapshot()); } SkillStatus::Error => { @@ -464,6 +469,10 @@ impl RuntimeEngine { // Don't unregister — keep the skill with Error status so the // UI can see what happened and allow restart. self.emit_status_change(&skill_id_owned); + publish_global(DomainEvent::SkillStartFailed { + skill_id: skill_id_owned.clone(), + error: error_msg.clone(), + }); return Err(format!( "Skill '{}' failed to start: {}", skill_id_owned, error_msg @@ -507,6 +516,9 @@ impl RuntimeEngine { 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(), + }); Ok(()) } @@ -549,9 +561,24 @@ impl RuntimeEngine { tool_name: &str, arguments: serde_json::Value, ) -> Result { - self.registry - .call_tool(skill_id, tool_name, arguments) - .await + let started_at = std::time::Instant::now(); + let result = self + .registry + .call_tool(skill_id, tool_name, arguments.clone()) + .await; + let result_text = match &result { + Ok(r) => Some(r.output()), + Err(e) => Some(e.clone()), + }; + publish_global(DomainEvent::SkillExecuted { + skill_id: skill_id.to_string(), + tool_name: tool_name.to_string(), + arguments, + result: result_text, + success: result.is_ok(), + elapsed_ms: started_at.elapsed().as_millis() as u64, + }); + result } /// Call a tool from inside a running skill. Enforces self-only invocation. @@ -565,16 +592,31 @@ impl RuntimeEngine { tool_name: &str, arguments: serde_json::Value, ) -> Result { - self.registry + let started_at = std::time::Instant::now(); + let result = self + .registry .call_tool_scoped( ToolCallOrigin::SkillSelf { skill_id: caller_skill_id.to_string(), }, target_skill_id, tool_name, - arguments, + arguments.clone(), ) - .await + .await; + let result_text = match &result { + Ok(r) => Some(r.output()), + Err(e) => Some(e.clone()), + }; + publish_global(DomainEvent::SkillExecuted { + skill_id: target_skill_id.to_string(), + tool_name: tool_name.to_string(), + arguments, + result: result_text, + success: result.is_ok(), + elapsed_ms: started_at.elapsed().as_millis() as u64, + }); + result } /// Broadcast a server event to all running skills. diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs index 99c9cd366..c678c4a54 100644 --- a/src/openhuman/socket/event_handlers.rs +++ b/src/openhuman/socket/event_handlers.rs @@ -1,7 +1,8 @@ //! Socket.IO event routing and protocol handlers. //! -//! Dispatches incoming Socket.IO events to the appropriate handler: -//! webhook tunnel requests, channel inbound messages, or generic event logging. +//! Thin transport layer: parses incoming Socket.IO events and publishes them +//! to the event bus for domain-specific handling. Webhook routing lives in +//! `webhooks::bus`, channel inbound processing lives in `channels::bus`. use std::sync::Arc; @@ -9,6 +10,7 @@ use serde_json::json; use tokio::sync::mpsc; use crate::api::models::socket::ConnectionStatus; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use crate::openhuman::webhooks::WebhookRequest; use super::manager::{emit_server_event, emit_state_change, SharedState}; @@ -21,7 +23,7 @@ use super::manager::{emit_server_event, emit_state_change, SharedState}; pub(super) fn handle_sio_event( event_name: &str, data: serde_json::Value, - emit_tx: &mpsc::UnboundedSender, + _emit_tx: &mpsc::UnboundedSender, shared: &Arc, ) { // Log every incoming event for observability. @@ -47,26 +49,102 @@ pub(super) fn handle_sio_event( *shared.status.write() = ConnectionStatus::Error; emit_state_change(shared); } - // Webhook tunnel — route to owning skill and relay response + // Webhook tunnel — publish to event bus for routing by WebhookRequestSubscriber "webhook:request" => { - log::info!("[socket] Routing webhook:request to handler"); - let shared = Arc::clone(shared); - let tx = emit_tx.clone(); - tokio::spawn(async move { - handle_webhook_request(&shared, data, &tx).await; - }); + log::info!("[socket] Publishing webhook:request to event bus"); + match serde_json::from_value::(data.clone()) { + Ok(request) => { + publish_global(DomainEvent::WebhookIncomingRequest { + request, + raw_data: data, + }); + } + Err(e) => { + log::error!("[socket] Failed to parse webhook:request payload: {e}"); + // Publish with a minimal request so the subscriber can still + // emit an error response. Build a request from what we can parse. + let cid = data + .get("correlationId") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + let _tunnel_uuid = data + .get("tunnelUuid") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + // Record parse error in router debug log if available + if let Some(router) = shared.webhook_router.read().clone() { + router.record_parse_error( + cid.clone(), + data.get("tunnelUuid") + .and_then(|v| v.as_str()) + .map(|v| v.to_string()), + data.get("method") + .and_then(|v| v.as_str()) + .map(|v| v.to_string()), + data.get("path") + .and_then(|v| v.as_str()) + .map(|v| v.to_string()), + data.clone(), + format!("bad request: {e}"), + ); + } + + // Emit error response directly via socket manager + if let Some(mgr) = crate::openhuman::socket::global_socket_manager() { + let err_json = json!({ "error": format!("Bad request: {e}") }); + let body = base64_encode(&err_json.to_string()); + let response_data = json!({ + "correlationId": cid, + "statusCode": 400, + "headers": {}, + "body": body, + }); + let mgr = mgr.clone(); + tokio::spawn(async move { + if let Err(e) = mgr.emit("webhook:response", response_data).await { + log::error!("[socket] Failed to emit webhook error response: {e}"); + } + }); + } + } + } } - // Any event ending with ":message" is treated as an inbound channel - // message that triggers the agent loop. This covers channel:message, - // telegram:message, discord:message, slack:message, and any future - // channel integration without requiring code changes. + // Channel inbound message — publish to event bus for ChannelInboundSubscriber _ if event_name.ends_with(":message") => { log::info!( - "[socket] Inbound channel message via '{}' — triggering agent loop", + "[socket] Publishing inbound channel message '{}' to event bus", event_name ); - tokio::spawn(async move { - handle_channel_inbound_message(data).await; + + let channel = data + .get("channel") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let message = data + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim() + .to_string(); + + if channel.is_empty() { + log::warn!("[socket] channel:message missing 'channel' field"); + return; + } + if message.is_empty() { + log::debug!("[socket] channel:message empty or missing 'message'"); + return; + } + + publish_global(DomainEvent::ChannelInboundMessage { + event_name: event_name.to_string(), + channel, + message, + raw_data: data, }); } _ => { @@ -76,387 +154,11 @@ pub(super) fn handle_sio_event( } } -// --------------------------------------------------------------------------- -// Webhook tunnel handler -// --------------------------------------------------------------------------- - -/// Handle an incoming `webhook:request` event from the backend. -/// -/// Routes the request to the owning skill via the WebhookRouter, waits for the -/// skill's response, and emits `webhook:response` back through the socket. -async fn handle_webhook_request( - shared: &SharedState, - data: serde_json::Value, - emit_tx: &mpsc::UnboundedSender, -) { - let request: WebhookRequest = match serde_json::from_value(data.clone()) { - Ok(r) => r, - Err(e) => { - log::error!("[socket] Failed to parse webhook:request payload: {e}"); - let cid = data - .get("correlationId") - .and_then(|v| v.as_str()) - .unwrap_or("unknown") - .to_string(); - if let Some(router) = shared.webhook_router.read().clone() { - router.record_parse_error( - cid.clone(), - data.get("tunnelUuid") - .and_then(|value| value.as_str()) - .map(|value| value.to_string()), - data.get("method") - .and_then(|value| value.as_str()) - .map(|value| value.to_string()), - data.get("path") - .and_then(|value| value.as_str()) - .map(|value| value.to_string()), - data.clone(), - format!("bad request: {e}"), - ); - } - emit_via_channel( - emit_tx, - "webhook:response", - json!({ - "correlationId": cid, - "statusCode": 400, - "headers": {}, - "body": base64_encode(&format!( - "{{\"error\":\"Bad request: {}\"}}", - e.to_string().replace('"', "\\\"") - )), - }), - ); - return; - } - }; - - let correlation_id = request.correlation_id.clone(); - let tunnel_uuid = request.tunnel_uuid.clone(); - let tunnel_name = request.tunnel_name.clone(); - let method = request.method.clone(); - let path = request.path.clone(); - - log::info!( - "[socket] webhook:request {} {} (tunnel={}, correlationId={})", - method, - path, - tunnel_uuid, - correlation_id, - ); - - let router = shared.webhook_router.read().clone(); - let registration = router.as_ref().and_then(|r| r.registration(&tunnel_uuid)); - let skill_id = registration.as_ref().and_then(|registration| { - (registration.target_kind == "skill").then(|| registration.skill_id.clone()) - }); - if let Some(router) = router.as_ref() { - router.record_request(&request, skill_id.clone()); - } - - let (response, resolved_skill_id, response_error) = match registration.as_ref() { - Some(registration) if registration.target_kind == "echo" => ( - crate::openhuman::webhooks::ops::build_echo_response(&request), - Some("echo".to_string()), - None, - ), - Some(registration) if registration.target_kind == "channel" => ( - crate::openhuman::webhooks::WebhookResponseData { - correlation_id: correlation_id.clone(), - status_code: 501, - headers: std::collections::HashMap::new(), - body: base64_encode(&format!( - "{{\"error\":\"channel webhook target '{}' is not implemented in this runtime yet\"}}", - registration.skill_id.replace('"', "\\\"") - )), - }, - Some(registration.skill_id.clone()), - Some("channel webhook target not implemented".to_string()), - ), - Some(registration) if registration.target_kind == "skill" => { - let sid = registration.skill_id.clone(); - log::debug!("[socket] webhook:request routed to skill '{}'", sid); - - let registry = crate::openhuman::skills::global_engine() - .map(|e| e.registry()); - match registry { - Some(registry) => { - let result = registry - .send_webhook_request( - &sid, - correlation_id.clone(), - request.method.clone(), - request.path.clone(), - request.headers.clone(), - request.query.clone(), - request.body.clone(), - request.tunnel_id.clone(), - request.tunnel_name.clone(), - ) - .await; - - match result { - Ok(resp) => (resp, Some(sid), None), - Err(e) => { - log::warn!("[socket] Skill webhook handler error: {}", e); - ( - crate::openhuman::webhooks::WebhookResponseData { - correlation_id: correlation_id.clone(), - status_code: 500, - headers: std::collections::HashMap::new(), - body: base64_encode(&format!( - "{{\"error\":\"Skill handler error: {}\"}}", - e.replace('"', "\\\"") - )), - }, - Some(sid), - Some(e), - ) - } - } - } - None => { - log::warn!("[socket] No skill registry available for webhook"); - ( - crate::openhuman::webhooks::WebhookResponseData { - correlation_id: correlation_id.clone(), - status_code: 503, - headers: std::collections::HashMap::new(), - body: base64_encode("{\"error\":\"Runtime not ready\"}"), - }, - None, - Some("runtime not ready".to_string()), - ) - } - } - } - Some(registration) => ( - crate::openhuman::webhooks::WebhookResponseData { - correlation_id: correlation_id.clone(), - status_code: 400, - headers: std::collections::HashMap::new(), - body: base64_encode(&format!( - "{{\"error\":\"unknown webhook target kind '{}'\"}}", - registration.target_kind.replace('"', "\\\"") - )), - }, - Some(registration.skill_id.clone()), - Some("unknown webhook target kind".to_string()), - ), - None => { - log::debug!( - "[socket] No skill registered for tunnel {}", - tunnel_uuid, - ); - ( - crate::openhuman::webhooks::WebhookResponseData { - correlation_id: correlation_id.clone(), - status_code: 404, - headers: std::collections::HashMap::new(), - body: base64_encode("{\"error\":\"No handler registered for this tunnel\"}"), - }, - None, - Some("no handler registered for this tunnel".to_string()), - ) - } - }; - - if let Some(router) = router.as_ref() { - router.record_response( - &request, - &response, - resolved_skill_id.clone(), - response_error.clone(), - ); - } - - emit_via_channel( - emit_tx, - "webhook:response", - json!({ - "correlationId": response.correlation_id, - "statusCode": response.status_code, - "headers": response.headers, - "body": response.body, - }), - ); - - log::info!( - "[socket] webhook activity: {} {} → status={}, skill={:?}, tunnel={}", - method, - path, - response.status_code, - resolved_skill_id, - tunnel_name, - ); -} - -// --------------------------------------------------------------------------- -// Channel inbound message → agent loop → reply -// --------------------------------------------------------------------------- - -/// Handle an inbound message from a channel (Telegram, Discord, etc.). -/// -/// Runs the agent inference loop via `web::start_chat` and sends the response -/// back to the originating channel via the REST API. -async fn handle_channel_inbound_message(data: serde_json::Value) { - let channel = match data.get("channel").and_then(|v| v.as_str()) { - Some(c) => c.to_string(), - None => { - log::warn!("[channel-inbound] channel:message missing 'channel' field"); - return; - } - }; - let message = match data.get("message").and_then(|v| v.as_str()) { - Some(m) if !m.trim().is_empty() => m.trim().to_string(), - _ => { - log::debug!("[channel-inbound] channel:message empty or missing 'message'"); - return; - } - }; - - log::info!( - "[channel-inbound] received message from channel='{}' len={}", - channel, - message.len() - ); - - let thread_id = format!("channel:{}", channel); - let client_id = "inbound".to_string(); - - let mut event_rx = crate::openhuman::channels::providers::web::subscribe_web_channel_events(); - - let request_id = match crate::openhuman::channels::providers::web::start_chat( - &client_id, &thread_id, &message, None, None, - ) - .await - { - Ok(rid) => { - log::debug!( - "[channel-inbound] agent started request_id={} thread={}", - rid, - thread_id - ); - rid - } - Err(err) => { - log::error!("[channel-inbound] start_chat failed: {}", err); - send_channel_reply( - &channel, - &format!("Sorry, I couldn't process your message: {err}"), - ) - .await; - return; - } - }; - - let timeout = tokio::time::Duration::from_secs(180); - let deadline = tokio::time::Instant::now() + timeout; - - loop { - tokio::select! { - event = event_rx.recv() => { - match event { - Ok(ev) if ev.request_id == request_id => { - if ev.event == "chat_done" || ev.event == "chat:done" { - let reply = ev.full_response.unwrap_or_default(); - if reply.trim().is_empty() { - log::warn!("[channel-inbound] agent returned empty response"); - return; - } - log::info!( - "[channel-inbound] agent done, replying to channel='{}' len={}", - channel, - reply.len() - ); - send_channel_reply(&channel, &reply).await; - return; - } - if ev.event == "chat_error" || ev.event == "chat:error" { - let err_msg = ev.message.unwrap_or_else(|| "unknown error".to_string()); - log::error!("[channel-inbound] agent error: {}", err_msg); - send_channel_reply( - &channel, - &format!("Sorry, I encountered an error: {err_msg}"), - ) - .await; - return; - } - } - Ok(_) => {} - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - log::warn!("[channel-inbound] event bus lagged, skipped {} events", n); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - log::error!("[channel-inbound] event bus closed unexpectedly"); - return; - } - } - } - _ = tokio::time::sleep_until(deadline) => { - log::error!("[channel-inbound] agent timed out after {}s", timeout.as_secs()); - send_channel_reply(&channel, "Sorry, the request timed out.").await; - return; - } - } - } -} - -/// Send a text reply back to a channel via the backend REST API. -async fn send_channel_reply(channel: &str, text: &str) { - let config = match crate::openhuman::config::rpc::load_config_with_timeout().await { - Ok(c) => c, - Err(e) => { - log::error!("[channel-inbound] failed to load config: {}", e); - return; - } - }; - - let api_url = crate::api::config::effective_api_url(&config.api_url); - let jwt = match crate::api::jwt::get_session_token(&config) { - Ok(Some(t)) => t, - Ok(None) => { - log::error!("[channel-inbound] no session JWT — cannot reply"); - return; - } - Err(e) => { - log::error!("[channel-inbound] failed to get session token: {}", e); - return; - } - }; - - let client = match crate::api::rest::BackendOAuthClient::new(&api_url) { - Ok(c) => c, - Err(e) => { - log::error!("[channel-inbound] failed to create API client: {}", e); - return; - } - }; - - let body = json!({ "text": text }); - match client.send_channel_message(channel, &jwt, body).await { - Ok(resp) => { - log::info!( - "[channel-inbound] reply sent to channel='{}' response={:?}", - channel, - resp - ); - } - Err(e) => { - log::error!( - "[channel-inbound] failed to send reply to channel='{}': {}", - channel, - e - ); - } - } -} - // --------------------------------------------------------------------------- // Utility functions // --------------------------------------------------------------------------- -/// Base64-encode a string (for webhook response bodies). +/// Base64-encode a string (for webhook error response bodies). fn base64_encode(input: &str) -> String { use base64::Engine; base64::engine::general_purpose::STANDARD.encode(input.as_bytes()) diff --git a/src/openhuman/webhooks/bus.rs b/src/openhuman/webhooks/bus.rs new file mode 100644 index 000000000..bdf5c3d62 --- /dev/null +++ b/src/openhuman/webhooks/bus.rs @@ -0,0 +1,237 @@ +//! Event bus handlers for the webhook domain. +//! +//! The [`WebhookRequestSubscriber`] handles incoming webhook requests published +//! by the socket transport layer. It routes each request to the owning skill (or +//! echo target), waits for the response, and emits it back through the socket. +//! This decouples the socket module from webhook routing logic. + +use crate::openhuman::event_bus::{publish_global, DomainEvent, EventHandler}; +use crate::openhuman::socket::global_socket_manager; +use async_trait::async_trait; +use serde_json::json; +use std::time::Instant; + +/// Base64-encode a string (for webhook response bodies). +fn base64_encode(input: &str) -> String { + use base64::Engine; + base64::engine::general_purpose::STANDARD.encode(input.as_bytes()) +} + +/// Build a base64-encoded JSON error body using proper serialization. +fn error_body(message: &str) -> String { + let obj = serde_json::json!({ "error": message }); + base64_encode(&obj.to_string()) +} + +/// Subscribes to `WebhookIncomingRequest` events and handles the full routing +/// flow: lookup tunnel → dispatch to skill/echo → emit response via socket. +pub struct WebhookRequestSubscriber; + +impl WebhookRequestSubscriber { + pub fn new() -> Self { + Self + } +} + +#[async_trait] +impl EventHandler for WebhookRequestSubscriber { + fn name(&self) -> &str { + "webhook::request_handler" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["webhook"]) + } + + async fn handle(&self, event: &DomainEvent) { + let DomainEvent::WebhookIncomingRequest { + request, + raw_data: _, + } = event + else { + return; + }; + + let started_at = Instant::now(); + + let correlation_id = request.correlation_id.clone(); + let tunnel_uuid = request.tunnel_uuid.clone(); + let tunnel_name = request.tunnel_name.clone(); + let method = request.method.clone(); + let path = request.path.clone(); + + tracing::info!( + "[webhook] incoming request {} {} (tunnel={}, correlationId={})", + method, + path, + tunnel_uuid, + correlation_id, + ); + + // Get the webhook router from the skill engine + let router = crate::openhuman::skills::global_engine().map(|e| e.webhook_router()); + + let registration = router.as_ref().and_then(|r| r.registration(&tunnel_uuid)); + let skill_id = registration + .as_ref() + .and_then(|reg| (reg.target_kind == "skill").then(|| reg.skill_id.clone())); + if let Some(ref router) = router { + router.record_request(request, skill_id.clone()); + } + + let (response, resolved_skill_id, response_error) = match registration.as_ref() { + Some(reg) if reg.target_kind == "echo" => ( + crate::openhuman::webhooks::ops::build_echo_response(request), + Some("echo".to_string()), + None, + ), + Some(reg) if reg.target_kind == "channel" => ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 501, + headers: std::collections::HashMap::new(), + body: error_body(&format!( + "channel webhook target '{}' is not implemented in this runtime yet", + reg.skill_id + )), + }, + Some(reg.skill_id.clone()), + Some("channel webhook target not implemented".to_string()), + ), + Some(reg) if reg.target_kind == "skill" => { + let sid = reg.skill_id.clone(); + tracing::debug!("[webhook] request routed to skill '{}'", sid); + + let registry = crate::openhuman::skills::global_engine().map(|e| e.registry()); + match registry { + Some(registry) => { + let result = registry + .send_webhook_request( + &sid, + correlation_id.clone(), + request.method.clone(), + request.path.clone(), + request.headers.clone(), + request.query.clone(), + request.body.clone(), + request.tunnel_id.clone(), + request.tunnel_name.clone(), + ) + .await; + + match result { + Ok(resp) => (resp, Some(sid), None), + Err(e) => { + tracing::warn!("[webhook] skill handler error: {}", e); + ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 500, + headers: std::collections::HashMap::new(), + body: error_body(&format!("Skill handler error: {}", e)), + }, + Some(sid), + Some(e), + ) + } + } + } + None => { + tracing::warn!("[webhook] no skill registry available"); + ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 503, + headers: std::collections::HashMap::new(), + body: error_body("Runtime not ready"), + }, + None, + Some("runtime not ready".to_string()), + ) + } + } + } + Some(reg) => ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 400, + headers: std::collections::HashMap::new(), + body: error_body(&format!( + "unknown webhook target kind '{}'", + reg.target_kind + )), + }, + Some(reg.skill_id.clone()), + Some("unknown webhook target kind".to_string()), + ), + None => { + tracing::debug!("[webhook] no handler registered for tunnel {}", tunnel_uuid,); + ( + crate::openhuman::webhooks::WebhookResponseData { + correlation_id: correlation_id.clone(), + status_code: 404, + headers: std::collections::HashMap::new(), + body: error_body("No handler registered for this tunnel"), + }, + None, + Some("no handler registered for this tunnel".to_string()), + ) + } + }; + + // Record in debug log + if let Some(ref router) = router { + router.record_response( + request, + &response, + resolved_skill_id.clone(), + response_error.clone(), + ); + } + + // Publish notification events + if let Some(ref sid) = resolved_skill_id { + publish_global(DomainEvent::WebhookReceived { + tunnel_id: tunnel_uuid.clone(), + skill_id: sid.clone(), + method: method.clone(), + path: path.clone(), + correlation_id: correlation_id.clone(), + }); + } + publish_global(DomainEvent::WebhookProcessed { + tunnel_id: tunnel_uuid.clone(), + skill_id: resolved_skill_id.clone().unwrap_or_default(), + method: method.clone(), + path: path.clone(), + correlation_id: correlation_id.clone(), + status_code: response.status_code, + elapsed_ms: started_at.elapsed().as_millis() as u64, + error: response_error.clone(), + }); + + // Emit response back through the socket + if let Some(mgr) = global_socket_manager() { + let response_data = json!({ + "correlationId": response.correlation_id, + "statusCode": response.status_code, + "headers": response.headers, + "body": response.body, + }); + if let Err(e) = mgr.emit("webhook:response", response_data).await { + tracing::error!("[webhook] failed to emit response via socket: {}", e); + } + } else { + tracing::error!("[webhook] no socket manager available to emit response"); + } + + tracing::info!( + "[webhook] {} {} → status={}, skill={:?}, tunnel={}", + method, + path, + response.status_code, + resolved_skill_id, + tunnel_name, + ); + } +} diff --git a/src/openhuman/webhooks/mod.rs b/src/openhuman/webhooks/mod.rs index 483026ad5..5851bef0c 100644 --- a/src/openhuman/webhooks/mod.rs +++ b/src/openhuman/webhooks/mod.rs @@ -4,6 +4,7 @@ //! appropriate skill. The backend manages tunnel provisioning (ngrok, cloudflare, //! etc.); this module handles the client-side routing and skill dispatch. +pub mod bus; pub mod ops; pub mod router; mod schemas; diff --git a/src/openhuman/webhooks/router.rs b/src/openhuman/webhooks/router.rs index 326578aa0..a84473ffe 100644 --- a/src/openhuman/webhooks/router.rs +++ b/src/openhuman/webhooks/router.rs @@ -4,6 +4,7 @@ use super::types::{ TunnelRegistration, WebhookDebugEvent, WebhookDebugLogEntry, WebhookRequest, WebhookResponseData, }; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; use log::{debug, error, warn}; use once_cell::sync::Lazy; use serde::{Deserialize, Serialize}; @@ -140,6 +141,7 @@ impl WebhookRouter { tunnel_uuid, target_kind, skill_id ); + let tunnel_name_clone = tunnel_name.clone(); routes.insert( tunnel_uuid.to_string(), TunnelRegistration { @@ -154,6 +156,13 @@ impl WebhookRouter { drop(routes); self.publish_event("registration_changed", None, Some(tunnel_uuid.to_string())); self.persist(); + + publish_global(DomainEvent::WebhookRegistered { + tunnel_id: tunnel_uuid.to_string(), + skill_id: skill_id.to_string(), + tunnel_name: tunnel_name_clone, + }); + Ok(()) } @@ -183,6 +192,12 @@ impl WebhookRouter { drop(routes); self.publish_event("registration_changed", None, Some(tunnel_uuid.to_string())); self.persist(); + + publish_global(DomainEvent::WebhookUnregistered { + tunnel_id: tunnel_uuid.to_string(), + skill_id: skill_id.to_string(), + }); + Ok(()) } @@ -196,18 +211,30 @@ impl WebhookRouter { } }; - let before = routes.len(); - routes.retain(|_, reg| reg.skill_id != skill_id); - let removed = before - routes.len(); + let removed_tunnels: Vec = routes + .iter() + .filter(|(_, reg)| reg.skill_id == skill_id) + .map(|(uuid, _)| uuid.clone()) + .collect(); - if removed > 0 { + routes.retain(|_, reg| reg.skill_id != skill_id); + + if !removed_tunnels.is_empty() { debug!( "[webhooks] Unregistered {} tunnel(s) for skill '{}'", - removed, skill_id + removed_tunnels.len(), + skill_id ); drop(routes); self.publish_event("registration_changed", None, None); self.persist(); + + for tunnel_id in removed_tunnels { + publish_global(DomainEvent::WebhookUnregistered { + tunnel_id, + skill_id: skill_id.to_string(), + }); + } } }