From 601d73d16b4faa8d2126d648fa8cba8d7d681b06 Mon Sep 17 00:00:00 2001 From: Henry Park Date: Fri, 27 Feb 2026 12:01:32 -0800 Subject: [PATCH] feat(routines): deliver notifications to all installed channels (#398) * feat(routines): deliver notifications to all installed channels Routine notifications were silently lost because the forwarder didn't use NotifyConfig fields and WASM channels (Telegram, Slack) had broadcast() as a no-op. This fixes three issues: 1. send_notification() now includes notify_user/notify_channel in metadata so the forwarder can route to specific channels 2. The routine forwarder mirrors the heartbeat pattern: try targeted channel first, fall back to broadcast_all 3. WasmChannel implements broadcast() using last-seen message metadata (chat_id), with persistence to the settings table so it survives restarts. Only writes to DB when the value actually changes. Heartbeat notifications also benefit from the WASM broadcast fix. Co-Authored-By: Claude Sonnet 4.6 * refactor(wasm): extract do_update_broadcast_metadata to eliminate duplication The inline metadata-update block in `dispatch_emitted_messages` was identical to the `update_broadcast_metadata` instance method. Extract the shared logic into a private free function `do_update_broadcast_metadata` that both call, so the persistence logic lives in one place. Addresses Gemini code review comment on PR #398. Co-Authored-By: Claude Sonnet 4.6 --------- Co-authored-by: Claude Sonnet 4.6 --- src/agent/agent_loop.rs | 37 ++++++-- src/agent/routine_engine.rs | 2 + src/channels/wasm/loader.rs | 12 ++- src/channels/wasm/router.rs | 1 + src/channels/wasm/wrapper.rs | 147 ++++++++++++++++++++++++++++++ src/extensions/manager.rs | 9 +- src/main.rs | 10 +- tests/wasm_channel_integration.rs | 1 + 8 files changed, 205 insertions(+), 14 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 28a2cfc2..418b623d 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -413,7 +413,7 @@ impl Agent { // Load initial event cache engine.refresh_event_cache().await; - // Spawn notification forwarder + // Spawn notification forwarder (mirrors heartbeat pattern) let channels = self.channels.clone(); tokio::spawn(async move { while let Some(response) = notify_rx.recv().await { @@ -423,14 +423,33 @@ impl Agent { .and_then(|v| v.as_str()) .unwrap_or("default") .to_string(); - let results = channels.broadcast_all(&user, response).await; - for (ch, result) in results { - if let Err(e) = result { - tracing::warn!( - "Failed to broadcast routine notification to {}: {}", - ch, - e - ); + let notify_channel = response + .metadata + .get("notify_channel") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + // Try the configured channel first, fall back to + // broadcasting on all channels. + let targeted_ok = if let Some(ref channel) = notify_channel { + channels + .broadcast(channel, &user, response.clone()) + .await + .is_ok() + } else { + false + }; + + if !targeted_ok { + let results = channels.broadcast_all(&user, response).await; + for (ch, result) in results { + if let Err(e) = result { + tracing::warn!( + "Failed to broadcast routine notification to {}: {}", + ch, + e + ); + } } } } diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index 5598434c..75970c4f 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -605,6 +605,8 @@ async fn send_notification( "source": "routine", "routine_name": routine_name, "status": status.to_string(), + "notify_user": notify.user, + "notify_channel": notify.channel, }), }; diff --git a/src/channels/wasm/loader.rs b/src/channels/wasm/loader.rs index e597fc32..57372c2e 100644 --- a/src/channels/wasm/loader.rs +++ b/src/channels/wasm/loader.rs @@ -17,20 +17,27 @@ use crate::channels::wasm::error::WasmChannelError; use crate::channels::wasm::runtime::WasmChannelRuntime; use crate::channels::wasm::schema::ChannelCapabilitiesFile; use crate::channels::wasm::wrapper::WasmChannel; +use crate::db::SettingsStore; use crate::pairing::PairingStore; /// Loads WASM channels from the filesystem. pub struct WasmChannelLoader { runtime: Arc, pairing_store: Arc, + settings_store: Option>, } impl WasmChannelLoader { /// Create a new loader with the given runtime and pairing store. - pub fn new(runtime: Arc, pairing_store: Arc) -> Self { + pub fn new( + runtime: Arc, + pairing_store: Arc, + settings_store: Option>, + ) -> Self { Self { runtime, pairing_store, + settings_store, } } @@ -126,6 +133,7 @@ impl WasmChannelLoader { capabilities, config_json, self.pairing_store.clone(), + self.settings_store.clone(), ); tracing::info!( @@ -437,7 +445,7 @@ mod tests { async fn test_loader_invalid_name() { let config = WasmChannelRuntimeConfig::for_testing(); let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap()); - let loader = WasmChannelLoader::new(runtime, Arc::new(PairingStore::new())); + let loader = WasmChannelLoader::new(runtime, Arc::new(PairingStore::new()), None); let dir = TempDir::new().unwrap(); let wasm_path = dir.path().join("test.wasm"); diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs index 99309c36..870bfc37 100644 --- a/src/channels/wasm/router.rs +++ b/src/channels/wasm/router.rs @@ -601,6 +601,7 @@ mod tests { capabilities, "{}".to_string(), Arc::new(PairingStore::new()), + None, )) } diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index 3b8e5759..a29da1e7 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -553,6 +553,40 @@ pub struct WasmChannel { /// In-memory workspace store persisting writes across callback invocations. /// Ensures WASM channels can maintain state (e.g., polling offsets) between ticks. workspace_store: Arc, + + /// Last-seen message metadata (contains chat_id for broadcast routing). + /// Populated from incoming messages so `broadcast()` knows where to send. + last_broadcast_metadata: Arc>>, + + /// Settings store for persisting broadcast metadata across restarts. + settings_store: Option>, +} + +/// Update broadcast metadata in memory and persist to the settings store when +/// it changes. Extracted as a free function so both the `WasmChannel` instance +/// method and the static polling helper share one implementation. +async fn do_update_broadcast_metadata( + channel_name: &str, + metadata: &str, + last_broadcast_metadata: &tokio::sync::RwLock>, + settings_store: Option<&Arc>, +) { + let mut guard = last_broadcast_metadata.write().await; + let changed = guard.as_deref() != Some(metadata); + *guard = Some(metadata.to_string()); + drop(guard); + + if changed && let Some(store) = settings_store { + let key = format!("channel_broadcast_metadata_{}", channel_name); + let value = serde_json::Value::String(metadata.to_string()); + if let Err(e) = store.set_setting("default", &key, &value).await { + tracing::warn!( + channel = %channel_name, + "Failed to persist broadcast metadata: {}", + e + ); + } + } } impl WasmChannel { @@ -563,6 +597,7 @@ impl WasmChannel { capabilities: ChannelCapabilities, config_json: String, pairing_store: Arc, + settings_store: Option>, ) -> Self { let name = prepared.name.clone(); let rate_limiter = ChannelEmitRateLimiter::new(capabilities.emit_rate_limit.clone()); @@ -584,6 +619,8 @@ impl WasmChannel { typing_task: RwLock::new(None), pairing_store, workspace_store: Arc::new(ChannelWorkspaceStore::new()), + last_broadcast_metadata: Arc::new(tokio::sync::RwLock::new(None)), + settings_store, } } @@ -631,6 +668,51 @@ impl WasmChannel { &self.name } + /// Settings key for persisted broadcast metadata. + fn broadcast_metadata_key(&self) -> String { + format!("channel_broadcast_metadata_{}", self.name) + } + + /// Update broadcast metadata in memory and persist if changed (best-effort). + /// + /// Compares with the current value to avoid redundant DB writes on every + /// incoming message (the chat_id rarely changes). + async fn update_broadcast_metadata(&self, metadata: &str) { + do_update_broadcast_metadata( + &self.name, + metadata, + &self.last_broadcast_metadata, + self.settings_store.as_ref(), + ) + .await; + } + + /// Load broadcast metadata from settings store on startup. + async fn load_broadcast_metadata(&self) { + if let Some(ref store) = self.settings_store { + match store + .get_setting("default", &self.broadcast_metadata_key()) + .await + { + Ok(Some(serde_json::Value::String(meta))) => { + *self.last_broadcast_metadata.write().await = Some(meta); + tracing::debug!( + channel = %self.name, + "Restored broadcast metadata from settings" + ); + } + Ok(_) => {} + Err(e) => { + tracing::warn!( + channel = %self.name, + "Failed to load broadcast metadata: {}", + e + ); + } + } + } + } + /// Get the channel capabilities. pub fn capabilities(&self) -> &ChannelCapabilities { &self.capabilities @@ -1613,6 +1695,8 @@ impl WasmChannel { // Parse metadata JSON if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) { msg = msg.with_metadata(metadata); + // Store for broadcast routing (chat_id etc.) + self.update_broadcast_metadata(&emitted.metadata_json).await; } // Send to stream @@ -1656,6 +1740,8 @@ impl WasmChannel { let pairing_store = self.pairing_store.clone(); let callback_timeout = self.runtime.config().callback_timeout; let workspace_store = self.workspace_store.clone(); + let last_broadcast_metadata = self.last_broadcast_metadata.clone(); + let settings_store = self.settings_store.clone(); tokio::spawn(async move { let mut interval_timer = tokio::time::interval(interval); @@ -1690,6 +1776,8 @@ impl WasmChannel { emitted_messages, &message_tx, &rate_limiter, + &last_broadcast_metadata, + settings_store.as_ref(), ).await { tracing::warn!( channel = %channel_name, @@ -1813,6 +1901,8 @@ impl WasmChannel { messages: Vec, message_tx: &RwLock>>, rate_limiter: &RwLock, + last_broadcast_metadata: &tokio::sync::RwLock>, + settings_store: Option<&Arc>, ) -> Result<(), WasmChannelError> { tracing::info!( channel = %channel_name, @@ -1858,6 +1948,14 @@ impl WasmChannel { // Parse metadata JSON if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) { msg = msg.with_metadata(metadata); + // Store for broadcast routing (chat_id etc.) + do_update_broadcast_metadata( + channel_name, + &emitted.metadata_json, + last_broadcast_metadata, + settings_store, + ) + .await; } // Send to stream @@ -1893,6 +1991,9 @@ impl Channel for WasmChannel { } async fn start(&self) -> Result { + // Restore broadcast metadata from settings (survives restarts) + self.load_broadcast_metadata().await; + // Create message channel let (tx, rx) = mpsc::channel(256); *self.message_tx.write().await = Some(tx); @@ -1982,6 +2083,8 @@ impl Channel for WasmChannel { // The original metadata contains channel-specific routing info (e.g., Telegram chat_id) // that the WASM channel needs to send the reply to the correct destination. let metadata_json = serde_json::to_string(&msg.metadata).unwrap_or_default(); + // Store for broadcast routing (chat_id etc.) + self.update_broadcast_metadata(&metadata_json).await; self.call_on_respond( msg.id, &response.content, @@ -1997,6 +2100,34 @@ impl Channel for WasmChannel { Ok(()) } + async fn broadcast( + &self, + _user_id: &str, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + let metadata_json = self + .last_broadcast_metadata + .read() + .await + .clone() + .ok_or_else(|| ChannelError::SendFailed { + name: self.name.clone(), + reason: "No messages received yet — no chat_id available for broadcast".into(), + })?; + + self.call_on_respond( + uuid::Uuid::new_v4(), + &response.content, + response.thread_id.as_deref(), + &metadata_json, + ) + .await + .map_err(|e| ChannelError::SendFailed { + name: self.name.clone(), + reason: e.to_string(), + }) + } + async fn send_status( &self, status: StatusUpdate, @@ -2101,6 +2232,14 @@ impl Channel for SharedWasmChannel { self.inner.respond(msg, response).await } + async fn broadcast( + &self, + user_id: &str, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + self.inner.broadcast(user_id, response).await + } + async fn send_status( &self, status: StatusUpdate, @@ -2384,6 +2523,7 @@ mod tests { capabilities, "{}".to_string(), Arc::new(PairingStore::new()), + None, ) } @@ -2489,11 +2629,14 @@ mod tests { EmittedMessage::new("user2", "Another message"), ]; + let last_broadcast_metadata = Arc::new(tokio::sync::RwLock::new(None)); let result = WasmChannel::dispatch_emitted_messages( "test-channel", messages, &message_tx, &rate_limiter, + &last_broadcast_metadata, + None, ) .await; @@ -2527,11 +2670,14 @@ mod tests { let messages = vec![EmittedMessage::new("user1", "Hello!")]; // Should return Ok even without a sender (logs warning but doesn't fail) + let last_broadcast_metadata = Arc::new(tokio::sync::RwLock::new(None)); let result = WasmChannel::dispatch_emitted_messages( "test-channel", messages, &message_tx, &rate_limiter, + &last_broadcast_metadata, + None, ) .await; @@ -2562,6 +2708,7 @@ mod tests { capabilities, "{}".to_string(), Arc::new(PairingStore::new()), + None, ); // Start the channel diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 1adbca18..d4bfbb92 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -1786,8 +1786,13 @@ impl ExtensionManager { None }; - let loader = - WasmChannelLoader::new(Arc::clone(&channel_runtime), Arc::clone(&pairing_store)); + let settings_store: Option> = + self.store.as_ref().map(|db| Arc::clone(db) as _); + let loader = WasmChannelLoader::new( + Arc::clone(&channel_runtime), + Arc::clone(&pairing_store), + settings_store, + ); let loaded = loader .load_from_files(name, &wasm_path, cap_path_option) .await diff --git a/src/main.rs b/src/main.rs index 1b3877c5..ed5d9dee 100644 --- a/src/main.rs +++ b/src/main.rs @@ -348,6 +348,7 @@ async fn main() -> anyhow::Result<()> { &config, &components.secrets_store, components.extension_manager.as_ref(), + components.db.as_ref(), ) .await; @@ -863,6 +864,7 @@ async fn setup_wasm_channels( config: &ironclaw::config::Config, secrets_store: &Option>, extension_manager: Option<&Arc>, + database: Option<&Arc>, ) -> Option { let runtime = match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) { Ok(r) => Arc::new(r), @@ -873,7 +875,13 @@ async fn setup_wasm_channels( }; let pairing_store = Arc::new(PairingStore::new()); - let loader = WasmChannelLoader::new(Arc::clone(&runtime), Arc::clone(&pairing_store)); + let settings_store: Option> = + database.map(|db| Arc::clone(db) as Arc); + let loader = WasmChannelLoader::new( + Arc::clone(&runtime), + Arc::clone(&pairing_store), + settings_store, + ); let results = match loader .load_from_dir(&config.channels.wasm_channels_dir) diff --git a/tests/wasm_channel_integration.rs b/tests/wasm_channel_integration.rs index 5d1fdf58..b5d1785b 100644 --- a/tests/wasm_channel_integration.rs +++ b/tests/wasm_channel_integration.rs @@ -45,6 +45,7 @@ fn create_test_channel( capabilities, "{}".to_string(), Arc::new(PairingStore::new()), + None, ) }