fix(slack): respond to thread replies without requiring @mention (#1405)

* fix(slack): respond to thread replies in channels without requiring @mention

Two fixes:

1. Host bug: `on_respond` callback never committed workspace writes or
   injected workspace reader, unlike all other WASM callbacks. Any WASM
   channel persisting state during on_respond silently lost data.

2. Slack WASM channel: track threads where the bot has participated via
   workspace storage. When a message event arrives in a channel thread
   the bot previously replied to, process it without requiring @mention.

Closes #1404

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(slack): log workspace_write error instead of silently discarding

Address code review feedback: handle the Result from workspace_write
when tracking thread participation, logging a warning on failure
instead of using `let _ =` which would silently swallow errors.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(slack): harden thread reply tracking

---------

Co-authored-by: synner88 <29090601+synner88@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Firat Sertgoz <f@nuff.tech>
Co-authored-by: firat.sertgoz <firat.sertgoz@near.ai>

FEATURE_PARITY.md

@@ -112,7 +112,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O
 |---------|----------|----------|-------|
 | Streaming draft replies | ✅ | ❌ | Partial replies via draft message updates |
 | Configurable stream modes | ✅ | ❌ | Per-channel stream behavior |
-| Thread ownership | ✅ | ❌ | Thread-level ownership tracking plus reply participation memory |
+| Thread ownership | ✅ | 🚧 | Reply participation memory now persists with TTL-bounded tracking; full thread-level ownership tracking is still missing |
 | Download-file action | ✅ | ❌ | On-demand attachment downloads via message actions |
 
 ### Mattermost-Specific Features (since Mar 2026)

channels-src/slack/src/lib.rs

@@ -23,6 +23,7 @@ wit_bindgen::generate!({
 });
 
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
 
 // Re-export generated types
 use exports::near::agent::channel::{
@@ -129,9 +130,17 @@ const OWNER_ID_PATH: &str = "state/owner_id";
 const DM_POLICY_PATH: &str = "state/dm_policy";
 /// Workspace path for persisting allow_from (JSON array) across WASM callbacks.
 const ALLOW_FROM_PATH: &str = "state/allow_from";
+/// Workspace path for tracking recently active Slack threads.
+const ACTIVE_THREADS_PATH: &str = "state/active_threads.json";
+/// Recently active threads expire after 24 hours to avoid reviving stale threads forever.
+const ACTIVE_THREAD_TTL_MS: u64 = 24 * 60 * 60 * 1000;
+/// Cap stored thread markers so the workspace state stays bounded.
+const ACTIVE_THREAD_MAX_ENTRIES: usize = 256;
 /// Channel name for pairing store (used by pairing host APIs).
 const CHANNEL_NAME: &str = "slack";
 
+type ActiveThreads = BTreeMap<String, u64>;
+
 /// Channel configuration from capabilities file.
 #[derive(Debug, Deserialize)]
 struct SlackConfig {
@@ -263,9 +272,9 @@ impl Guest for SlackChannel {
             "text": response.content,
         });
 
-        // Add thread_ts for threaded replies
-        if let Some(thread_ts) = response.thread_id.or(metadata.thread_ts) {
-            payload["thread_ts"] = serde_json::Value::String(thread_ts);
+        let thread_ts = response.thread_id.or(metadata.thread_ts);
+        if let Some(ref thread_ts) = thread_ts {
+            payload["thread_ts"] = serde_json::Value::String(thread_ts.clone());
         }
 
         let payload_bytes = serde_json::to_vec(&payload)
@@ -308,6 +317,10 @@ impl Guest for SlackChannel {
                     ));
                 }
 
+                if let Some(thread_ts) = thread_ts {
+                    track_active_thread(&metadata.channel, &thread_ts)?;
+                }
+
                 channel_host::log(
                     channel_host::LogLevel::Debug,
                     &format!(
@@ -452,13 +465,14 @@ fn download_and_store_slack_files(attachments: &[InboundAttachment]) {
     }
 }
 
+fn prepare_inbound_attachments(files: &Option<Vec<SlackFile>>) -> Vec<InboundAttachment> {
+    let attachments = extract_slack_attachments(files);
+    download_and_store_slack_files(&attachments);
+    attachments
+}
+
 /// Handle a Slack event and emit message if applicable.
 fn handle_slack_event(event: SlackEvent, team_id: Option<String>, _event_id: Option<String>) {
-    let attachments = extract_slack_attachments(&event.files);
-
-    // Download and store file attachments for host-side processing
-    download_and_store_slack_files(&attachments);
-
     match event.event_type.as_str() {
         // Direct mention of the bot (always in a channel, not a DM)
         "app_mention" => {
@@ -472,6 +486,7 @@ fn handle_slack_event(event: SlackEvent, team_id: Option<String>, _event_id: Opt
                 if !check_sender_permission(&user, &channel, false) {
                     return;
                 }
+                let attachments = prepare_inbound_attachments(&event.files);
                 emit_message(
                     user,
                     text,
@@ -483,7 +498,7 @@ fn handle_slack_event(event: SlackEvent, team_id: Option<String>, _event_id: Opt
             }
         }
 
-        // Direct message to the bot
+        // Direct message or thread follow-up to the bot
         "message" => {
             // Skip messages from bots (including ourselves)
             if event.bot_id.is_some() || event.subtype.is_some() {
@@ -496,11 +511,20 @@ fn handle_slack_event(event: SlackEvent, team_id: Option<String>, _event_id: Opt
                 event.text,
                 event.ts.clone(),
             ) {
-                // Only process DMs (channel IDs starting with D)
-                if channel.starts_with('D') {
-                    if !check_sender_permission(&user, &channel, true) {
+                let is_dm = channel.starts_with('D');
+
+                // Check if this is a reply in a thread where we previously participated
+                let is_active_thread = !is_dm
+                    && event
+                        .thread_ts
+                        .as_ref()
+                        .is_some_and(|thread_ts| is_active_thread(&channel, thread_ts));
+
+                if is_dm || is_active_thread {
+                    if !check_sender_permission(&user, &channel, is_dm) {
                         return;
                     }
+                    let attachments = prepare_inbound_attachments(&event.files);
                     emit_message(
                         user,
                         text,
@@ -522,6 +546,93 @@ fn handle_slack_event(event: SlackEvent, team_id: Option<String>, _event_id: Opt
     }
 }
 
+fn active_thread_key(channel: &str, thread_ts: &str) -> String {
+    format!("{channel}/{thread_ts}")
+}
+
+fn is_thread_marker_fresh(last_seen_millis: u64, now_millis: u64) -> bool {
+    now_millis.saturating_sub(last_seen_millis) <= ACTIVE_THREAD_TTL_MS
+}
+
+fn prune_active_threads(active_threads: &mut ActiveThreads, now_millis: u64) -> bool {
+    let mut changed = false;
+    active_threads.retain(|_, last_seen_millis| {
+        let keep = is_thread_marker_fresh(*last_seen_millis, now_millis);
+        if !keep {
+            changed = true;
+        }
+        keep
+    });
+
+    if active_threads.len() > ACTIVE_THREAD_MAX_ENTRIES {
+        let mut oldest_first: Vec<_> = active_threads
+            .iter()
+            .map(|(key, last_seen_millis)| (key.clone(), *last_seen_millis))
+            .collect();
+        oldest_first.sort_by_key(|(_, last_seen_millis)| *last_seen_millis);
+
+        for (key, _) in oldest_first
+            .into_iter()
+            .take(active_threads.len() - ACTIVE_THREAD_MAX_ENTRIES)
+        {
+            active_threads.remove(&key);
+            changed = true;
+        }
+    }
+
+    changed
+}
+
+fn load_active_threads() -> ActiveThreads {
+    let Some(raw) = channel_host::workspace_read(ACTIVE_THREADS_PATH) else {
+        return ActiveThreads::new();
+    };
+
+    match serde_json::from_str(&raw) {
+        Ok(active_threads) => active_threads,
+        Err(e) => {
+            channel_host::log(
+                channel_host::LogLevel::Warn,
+                &format!("Failed to parse active thread state: {e}"),
+            );
+            ActiveThreads::new()
+        }
+    }
+}
+
+fn persist_active_threads(active_threads: &ActiveThreads) -> Result<(), String> {
+    let serialized = serde_json::to_string(active_threads)
+        .map_err(|e| format!("Failed to serialize active thread state: {e}"))?;
+    channel_host::workspace_write(ACTIVE_THREADS_PATH, &serialized)
+        .map_err(|e| format!("Failed to persist active thread state: {e}"))
+}
+
+fn track_active_thread(channel: &str, thread_ts: &str) -> Result<(), String> {
+    let now_millis = channel_host::now_millis();
+    let mut active_threads = load_active_threads();
+    prune_active_threads(&mut active_threads, now_millis);
+    active_threads.insert(active_thread_key(channel, thread_ts), now_millis);
+    prune_active_threads(&mut active_threads, now_millis);
+    persist_active_threads(&active_threads)
+}
+
+fn is_active_thread(channel: &str, thread_ts: &str) -> bool {
+    let now_millis = channel_host::now_millis();
+    let mut active_threads = load_active_threads();
+    let changed = prune_active_threads(&mut active_threads, now_millis);
+
+    if changed {
+        if let Err(e) = persist_active_threads(&active_threads) {
+            channel_host::log(
+                channel_host::LogLevel::Warn,
+                &format!("Failed to prune active thread state: {e}"),
+            );
+        }
+    }
+
+    active_threads.contains_key(&active_thread_key(channel, thread_ts))
+}
+
 /// Emit a message to the agent.
 fn emit_message(
     user_id: String,
@@ -606,8 +717,7 @@ fn check_sender_permission(user_id: &str, channel_id: &str, is_dm: bool) -> bool
     }
 
     // 4. Check sender (Slack events only have user ID, not username)
-    let is_allowed =
-        allowed.contains(&"*".to_string()) || allowed.contains(&user_id.to_string());
+    let is_allowed = allowed.contains(&"*".to_string()) || allowed.contains(&user_id.to_string());
 
     if is_allowed {
         return true;
@@ -625,10 +735,7 @@ fn check_sender_permission(user_id: &str, channel_id: &str, is_dm: bool) -> bool
             Ok(result) => {
                 channel_host::log(
                     channel_host::LogLevel::Info,
-                    &format!(
-                        "Pairing request for user {}: code {}",
-                        user_id, result.code
-                    ),
+                    &format!("Pairing request for user {}: code {}", user_id, result.code),
                 );
                 if result.created {
                     let _ = send_pairing_reply(channel_id, &result.code);
@@ -826,4 +933,63 @@ mod tests {
         // Verify the constant is 20 MB
         assert_eq!(MAX_DOWNLOAD_SIZE_BYTES, 20 * 1024 * 1024);
     }
+
+    #[test]
+    fn test_active_thread_key_scopes_by_channel_and_thread() {
+        assert_eq!(
+            active_thread_key("C123", "1742486400.000100"),
+            "C123/1742486400.000100"
+        );
+    }
+
+    #[test]
+    fn test_prune_active_threads_removes_expired_entries() {
+        let now_millis = ACTIVE_THREAD_TTL_MS + 1_000;
+        let mut active_threads = ActiveThreads::from([
+            (
+                "C1/expired".to_string(),
+                now_millis - ACTIVE_THREAD_TTL_MS - 1,
+            ),
+            ("C1/fresh".to_string(), now_millis - ACTIVE_THREAD_TTL_MS),
+        ]);
+
+        let changed = prune_active_threads(&mut active_threads, now_millis);
+
+        assert!(changed);
+        assert!(!active_threads.contains_key("C1/expired"));
+        assert!(active_threads.contains_key("C1/fresh"));
+    }
+
+    #[test]
+    fn test_prune_active_threads_trims_oldest_entries_when_over_limit() {
+        let now_millis = ACTIVE_THREAD_TTL_MS + 1_000;
+        let mut active_threads = ActiveThreads::new();
+
+        for i in 0..=ACTIVE_THREAD_MAX_ENTRIES {
+            active_threads.insert(format!("C1/{i}"), now_millis + i as u64);
+        }
+
+        let changed = prune_active_threads(
+            &mut active_threads,
+            now_millis + ACTIVE_THREAD_MAX_ENTRIES as u64,
+        );
+
+        assert!(changed);
+        assert_eq!(active_threads.len(), ACTIVE_THREAD_MAX_ENTRIES);
+        assert!(!active_threads.contains_key("C1/0"));
+        assert!(active_threads.contains_key(&format!("C1/{ACTIVE_THREAD_MAX_ENTRIES}")));
+    }
+
+    #[test]
+    fn test_is_thread_marker_fresh_respects_ttl_boundary() {
+        let now_millis = ACTIVE_THREAD_TTL_MS + 1_000;
+        assert!(is_thread_marker_fresh(
+            now_millis - ACTIVE_THREAD_TTL_MS,
+            now_millis
+        ));
+        assert!(!is_thread_marker_fresh(
+            now_millis - ACTIVE_THREAD_TTL_MS - 1,
+            now_millis
+        ));
+    }
 }

src/channels/wasm/wrapper.rs

@@ -1800,7 +1800,7 @@ impl WasmChannel {
 
         let runtime = Arc::clone(&self.runtime);
         let prepared = Arc::clone(&self.prepared);
-        let capabilities = self.capabilities.clone();
+        let capabilities = Self::inject_workspace_reader(&self.capabilities, &self.workspace_store);
         let timeout = self.runtime.config().callback_timeout;
         let channel_name = self.name.clone();
         let credentials = self.get_credentials().await;
@@ -1811,6 +1811,7 @@ impl WasmChannel {
         )
         .await;
         let pairing_store = self.pairing_store.clone();
+        let workspace_store = self.workspace_store.clone();
 
         // Prepare response data
         let message_id_str = message_id.to_string();
@@ -1881,8 +1882,10 @@ impl WasmChannel {
                     });
                 }
 
-                let host_state =
+                let mut host_state =
                     Self::extract_host_state(&mut store, &prepared.name, &capabilities);
+                let pending_writes = host_state.take_pending_writes();
+                workspace_store.commit_writes(&pending_writes);
                 tracing::info!("on_respond WASM execution completed successfully");
                 Ok(((), host_state))
             })
@@ -1944,7 +1947,7 @@ impl WasmChannel {
 
         let runtime = Arc::clone(&self.runtime);
         let prepared = Arc::clone(&self.prepared);
-        let capabilities = self.capabilities.clone();
+        let capabilities = Self::inject_workspace_reader(&self.capabilities, &self.workspace_store);
         let timeout = self.runtime.config().callback_timeout;
         let channel_name = self.name.clone();
         let credentials = self.get_credentials().await;
@@ -1955,6 +1958,7 @@ impl WasmChannel {
         )
         .await;
         let pairing_store = self.pairing_store.clone();
+        let workspace_store = self.workspace_store.clone();
 
         let user_id = user_id.to_string();
         let content = content.to_string();
@@ -2006,8 +2010,10 @@ impl WasmChannel {
                     });
                 }
 
-                let host_state =
+                let mut host_state =
                     Self::extract_host_state(&mut store, &prepared.name, &capabilities);
+                let pending_writes = host_state.take_pending_writes();
+                workspace_store.commit_writes(&pending_writes);
                 tracing::info!("on_broadcast WASM execution completed successfully");
                 Ok(((), host_state))
             })
@@ -2051,7 +2057,7 @@ impl WasmChannel {
 
         let runtime = Arc::clone(&self.runtime);
         let prepared = Arc::clone(&self.prepared);
-        let capabilities = self.capabilities.clone();
+        let capabilities = Self::inject_workspace_reader(&self.capabilities, &self.workspace_store);
         let timeout = self.runtime.config().callback_timeout;
         let channel_name = self.name.clone();
         let credentials = self.get_credentials().await;
@@ -2062,6 +2068,7 @@ impl WasmChannel {
         )
         .await;
         let pairing_store = self.pairing_store.clone();
+        let workspace_store = self.workspace_store.clone();
 
         let Some(wit_update) = status_to_wit(status, metadata) else {
             return Ok(());
@@ -2084,6 +2091,11 @@ impl WasmChannel {
                     .call_on_status(&mut store, &wit_update)
                     .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?;
 
+                let mut host_state =
+                    Self::extract_host_state(&mut store, &prepared.name, &capabilities);
+                let pending_writes = host_state.take_pending_writes();
+                workspace_store.commit_writes(&pending_writes);
+
                 Ok(())
             })
             .await
@@ -2124,6 +2136,7 @@ impl WasmChannel {
         host_credentials: Vec<ResolvedHostCredential>,
         pairing_store: Arc<PairingStore>,
         timeout: Duration,
+        workspace_store: &Arc<ChannelWorkspaceStore>,
         wit_update: wit_channel::StatusUpdate,
     ) -> Result<(), WasmChannelError> {
         if prepared.component().is_none() {
@@ -2132,9 +2145,10 @@ impl WasmChannel {
 
         let runtime = Arc::clone(runtime);
         let prepared = Arc::clone(prepared);
-        let capabilities = capabilities.clone();
+        let capabilities = Self::inject_workspace_reader(capabilities, workspace_store);
         let credentials_snapshot = credentials.read().await.clone();
         let channel_name_owned = channel_name.to_string();
+        let workspace_store = Arc::clone(workspace_store);
 
         let result = tokio::time::timeout(timeout, async move {
             tokio::task::spawn_blocking(move || {
@@ -2153,6 +2167,11 @@ impl WasmChannel {
                     .call_on_status(&mut store, &wit_update)
                     .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?;
 
+                let mut host_state =
+                    Self::extract_host_state(&mut store, &prepared.name, &capabilities);
+                let pending_writes = host_state.take_pending_writes();
+                workspace_store.commit_writes(&pending_writes);
+
                 Ok(())
             })
             .await
@@ -2224,6 +2243,7 @@ impl WasmChannel {
                 let runtime = Arc::clone(&self.runtime);
                 let prepared = Arc::clone(&self.prepared);
                 let capabilities = self.capabilities.clone();
+                let workspace_store = self.workspace_store.clone();
                 let credentials = self.credentials.clone();
                 // Pre-resolve host credentials once for the lifetime of the repeater.
                 // Channels tokens rarely change, so a snapshot per-repeater is correct.
@@ -2259,6 +2279,7 @@ impl WasmChannel {
                             hc,
                             pairing_store.clone(),
                             callback_timeout,
+                            &workspace_store,
                             wit_update_clone,
                         )
                         .await
@@ -4430,6 +4451,47 @@ mod tests {
         assert_eq!(response.body, b"Bad request");
     }
 
+    #[test]
+    fn test_inject_workspace_reader_adds_missing_reader() {
+        let capabilities = ChannelCapabilities::for_channel("test");
+        assert!(capabilities.tool_capabilities.workspace_read.is_none());
+
+        let workspace_store = Arc::new(crate::channels::wasm::host::ChannelWorkspaceStore::new());
+        let injected = WasmChannel::inject_workspace_reader(&capabilities, &workspace_store);
+
+        assert!(injected.tool_capabilities.workspace_read.is_some());
+        assert!(
+            injected
+                .tool_capabilities
+                .workspace_read
+                .as_ref()
+                .and_then(|cap| cap.reader.as_ref())
+                .is_some()
+        );
+    }
+
+    #[test]
+    fn test_inject_workspace_reader_preserves_allowed_prefixes() {
+        let tool_capabilities = crate::tools::wasm::Capabilities::default()
+            .with_workspace_read(vec!["state/".to_string(), "context/".to_string()]);
+        let capabilities =
+            ChannelCapabilities::for_channel("test").with_tool_capabilities(tool_capabilities);
+        let workspace_store = Arc::new(crate::channels::wasm::host::ChannelWorkspaceStore::new());
+
+        let injected = WasmChannel::inject_workspace_reader(&capabilities, &workspace_store);
+
+        let workspace_read = injected
+            .tool_capabilities
+            .workspace_read
+            .as_ref()
+            .expect("workspace_read capability should exist");
+        assert_eq!(
+            workspace_read.allowed_prefixes,
+            vec!["state/".to_string(), "context/".to_string()]
+        );
+        assert!(workspace_read.reader.is_some());
+    }
+
     #[tokio::test]
     async fn test_channel_start_and_shutdown() {
         let channel = create_test_channel();