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 <[email protected]> * 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 <[email protected]> --------- Co-authored-by: Claude Sonnet 4.6 <[email protected]>
2026-08-25 14:53:34 +00:00 · 2026-02-27 12:01:32 -08:00
parent a65b282066
commit 601d73d16b
8 changed files with 205 additions and 14 deletions
@@ -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
+                                        );
+                                    }
                                }
                            }
                        }
@@ -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,
        }),
    };

@@ -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<WasmChannelRuntime>,
    pairing_store: Arc<PairingStore>,
+    settings_store: Option<Arc<dyn SettingsStore>>,
 }

 impl WasmChannelLoader {
    /// Create a new loader with the given runtime and pairing store.
-    pub fn new(runtime: Arc<WasmChannelRuntime>, pairing_store: Arc<PairingStore>) -> Self {
+    pub fn new(
+        runtime: Arc<WasmChannelRuntime>,
+        pairing_store: Arc<PairingStore>,
+        settings_store: Option<Arc<dyn SettingsStore>>,
+    ) -> 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");
@@ -601,6 +601,7 @@ mod tests {
            capabilities,
            "{}".to_string(),
            Arc::new(PairingStore::new()),
+            None,
        ))
    }

@@ -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<ChannelWorkspaceStore>,
+
+    /// Last-seen message metadata (contains chat_id for broadcast routing).
+    /// Populated from incoming messages so `broadcast()` knows where to send.
+    last_broadcast_metadata: Arc<tokio::sync::RwLock<Option<String>>>,
+
+    /// Settings store for persisting broadcast metadata across restarts.
+    settings_store: Option<Arc<dyn crate::db::SettingsStore>>,
+}
+
+/// 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<Option<String>>,
+    settings_store: Option<&Arc<dyn crate::db::SettingsStore>>,
+) {
+    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<PairingStore>,
+        settings_store: Option<Arc<dyn crate::db::SettingsStore>>,
    ) -> 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<EmittedMessage>,
        message_tx: &RwLock<Option<mpsc::Sender<IncomingMessage>>>,
        rate_limiter: &RwLock<ChannelEmitRateLimiter>,
+        last_broadcast_metadata: &tokio::sync::RwLock<Option<String>>,
+        settings_store: Option<&Arc<dyn crate::db::SettingsStore>>,
    ) -> 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<MessageStream, ChannelError> {
+        // 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
@@ -1786,8 +1786,13 @@ impl ExtensionManager {
            None
        };

-        let loader =
-            WasmChannelLoader::new(Arc::clone(&channel_runtime), Arc::clone(&pairing_store));
+        let settings_store: Option<Arc<dyn crate::db::SettingsStore>> =
+            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
@@ -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<Arc<dyn SecretsStore + Send + Sync>>,
    extension_manager: Option<&Arc<ironclaw::extensions::ExtensionManager>>,
+    database: Option<&Arc<dyn ironclaw::db::Database>>,
 ) -> Option<WasmChannelSetup> {
    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<Arc<dyn ironclaw::db::SettingsStore>> =
+        database.map(|db| Arc::clone(db) as Arc<dyn ironclaw::db::SettingsStore>);
+    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)
@@ -45,6 +45,7 @@ fn create_test_channel(
        capabilities,
        "{}".to_string(),
        Arc::new(PairingStore::new()),
+        None,
    )
 }