fix(slack): detect half-open Socket Mode via ping + idle-timeout reconnect (#1120)

* fix(slack): detect half-open Socket Mode via ping + idle-timeout reconnect

Slack's inbound Socket Mode WebSocket can go half-open (e.g. a NAT
idle-timeout silently drops inbound frames with no Close/FIN), which
leaves read.next() blocked forever so the reconnect loop never fires. The
bot then goes deaf to @mentions for hours while still logging "Slack
Socket Mode connected" (only outbound cron posts keep it looking alive).

Add a proactive WebSocket ping plus an idle-timeout that forces a
reconnect when no inbound frame (including Slack's own server pings) has
arrived within 75s, and align the reconnect delay with the gateway
adapter's capped exponential backoff (reset-on-success, interruptible by
shutdown). The idle and backoff decisions are pure functions with unit
tests.

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

* fix(slack): bound Slack Web API I/O to keep the read loop responsive

Follow-up to the heartbeat change: inline Slack Web API awaits in the read
loop (conversations.replies, bot_participated_in_thread, trusted-bot lookup)
could delay the idle-timeout watchdog tick. Keep gating serial in the read
loop (the batching ADR's canonical adapter pattern) and bound the I/O instead
of moving it off-thread:

- give the Slack Web API client a 30s request timeout (was unbounded)
- resolve bot_user_id once before the loop (no cold-cache auth.test inline)
- send the turn-limit warning off the bot_turns lock, with its result logged

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

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: ShinyChang

src/slack.rs

@@ -93,7 +93,12 @@ impl SlackAdapter {
         assistant_mode: bool,
     ) -> Self {
         Self {
-            client: reqwest::Client::new(),
+            // Bound every Slack Web API call; an unbounded inline gating call in the
+            // read loop could otherwise stall the Socket Mode idle-timeout watchdog.
+            client: reqwest::Client::builder()
+                .timeout(std::time::Duration::from_secs(30))
+                .build()
+                .unwrap_or_else(|_| reqwest::Client::new()),
             bot_token,
             bot_user_id: tokio::sync::OnceCell::new(),
             user_cache: tokio::sync::Mutex::new(HashMap::new()),
@@ -674,6 +679,28 @@ impl ChatAdapter for SlackAdapter {
 /// Hard cap on consecutive bot messages in a thread. Prevents runaway loops.
 const MAX_CONSECUTIVE_BOT_TURNS: usize = 1000;
 
+/// Socket Mode keepalive. Slack's inbound WebSocket can go half-open (e.g. a NAT
+/// idle-timeout silently drops inbound frames with no Close/FIN), which leaves
+/// `read.next()` blocked forever, so the reconnect loop never fires and the bot
+/// goes deaf while still showing as connected. We proactively ping and force a
+/// reconnect when no inbound frame (including Slack's own pings) has arrived
+/// within the idle window. Reconnect backoff mirrors the gateway adapter.
+const PING_INTERVAL_SECS: u64 = 30;
+const IDLE_TIMEOUT_SECS: u64 = 75;
+const MAX_BACKOFF_SECS: u64 = 30;
+
+/// Next reconnect delay: double, capped. Reset to 1 on a successful connect.
+fn next_backoff(cur: u64) -> u64 {
+    (cur * 2).min(MAX_BACKOFF_SECS)
+}
+
+/// The socket is considered dead (half-open) when no inbound frame has arrived
+/// within `timeout`; Slack sends periodic pings, so silence past the window
+/// means the inbound path is gone.
+fn socket_idle(since_last_inbound: std::time::Duration, timeout: std::time::Duration) -> bool {
+    since_last_inbound >= timeout
+}
+
 /// Run the Slack adapter using Socket Mode (persistent WebSocket, no public URL needed).
 /// Reconnects automatically on disconnect.
 #[allow(clippy::too_many_arguments)]
@@ -694,6 +721,10 @@ pub async fn run_slack_adapter(
 ) -> Result<()> {
     let bot_token = adapter.bot_token().to_string();
     let bot_turns = Arc::new(tokio::sync::Mutex::new(BotTurnTracker::new(max_bot_turns)));
+    // Warm the bot-user-id cache once so the per-message path never does the
+    // cold-cache `auth.test` inline in the read loop.
+    let _ = adapter.get_bot_user_id().await;
+    let mut backoff_secs = 1u64;
 
     loop {
         // Check for shutdown before (re)connecting
@@ -705,8 +736,12 @@ pub async fn run_slack_adapter(
         let ws_url = match get_socket_mode_url(&app_token).await {
             Ok(url) => url,
             Err(e) => {
-                error!("failed to get Socket Mode URL: {e}");
-                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+                error!(err = %e, backoff = backoff_secs, "failed to get Socket Mode URL, retrying");
+                tokio::select! {
+                    _ = tokio::time::sleep(std::time::Duration::from_secs(backoff_secs)) => {}
+                    _ = shutdown_rx.changed() => { return Ok(()); }
+                }
+                backoff_secs = next_backoff(backoff_secs);
                 continue;
             }
         };
@@ -715,11 +750,17 @@ pub async fn run_slack_adapter(
         match tokio_tungstenite::connect_async(&ws_url).await {
             Ok((ws_stream, _)) => {
                 info!("Slack Socket Mode connected");
+                backoff_secs = 1; // reset on success
                 let (mut write, mut read) = ws_stream.split();
+                let mut ping_interval =
+                    tokio::time::interval(std::time::Duration::from_secs(PING_INTERVAL_SECS));
+                ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+                let mut last_inbound = std::time::Instant::now();
 
                 loop {
                     tokio::select! {
                         msg_result = read.next() => {
+                            last_inbound = std::time::Instant::now();
                             let Some(msg_result) = msg_result else { break };
                             match msg_result {
                                 Ok(tungstenite::Message::Text(text)) => {
@@ -864,34 +905,47 @@ pub async fn run_slack_adapter(
                                                 } else {
                                                     format!("{}:{}", channel_id, event["ts"].as_str().unwrap_or(""))
                                                 };
-                                                {
+                                                // Classify under the lock (order-sensitive, kept in the read
+                                                // loop), but run any warning send AFTER releasing it; holding
+                                                // the tracker mutex across `chat.postMessage` would stall turn
+                                                // tracking for every thread, not just this one.
+                                                let turn_action = {
                                                     let mut tracker = bot_turns.lock().await;
                                                     if is_bot {
-                                                        match tracker.classify_bot_message(&turn_key) {
-                                                            TurnAction::Continue => {}
-                                                            TurnAction::SilentStop => continue,
-                                                            TurnAction::WarnAndStop { severity, turns, user_message } => {
-                                                                match severity {
-                                                                    TurnSeverity::Hard => warn!(channel_id, turns, "hard bot turn limit reached"),
-                                                                    TurnSeverity::Soft => info!(channel_id, turns, max = max_bot_turns, "soft bot turn limit reached"),
-                                                                }
-                                                                let channel_allowed = allow_all_channels
-                                                                    || allowed_channels.contains(channel_id);
-                                                                if !is_own_bot_msg && channel_allowed {
-                                                                    let warn_channel = ChannelRef {
-                                                                        platform: "slack".into(),
-                                                                        channel_id: channel_id.to_string(),
-                                                                        thread_id: event["thread_ts"].as_str().map(|s| s.to_string()),
-                                                                        parent_id: None,
-                                                                        origin_event_id: None,
-                                                                    };
-                                                                    let _ = adapter.send_message(&warn_channel, &user_message).await;
+                                                        tracker.classify_bot_message(&turn_key)
+                                                    } else {
+                                                        if is_plain_user_message(subtype, msg_text) {
+                                                            tracker.on_human_message(&turn_key);
+                                                        }
+                                                        TurnAction::Continue
+                                                    }
+                                                };
+                                                match turn_action {
+                                                    TurnAction::Continue => {}
+                                                    TurnAction::SilentStop => continue,
+                                                    TurnAction::WarnAndStop { severity, turns, user_message } => {
+                                                        match severity {
+                                                            TurnSeverity::Hard => warn!(channel_id, turns, "hard bot turn limit reached"),
+                                                            TurnSeverity::Soft => info!(channel_id, turns, max = max_bot_turns, "soft bot turn limit reached"),
+                                                        }
+                                                        let channel_allowed = allow_all_channels
+                                                            || allowed_channels.contains(channel_id);
+                                                        if !is_own_bot_msg && channel_allowed {
+                                                            let warn_channel = ChannelRef {
+                                                                platform: "slack".into(),
+                                                                channel_id: channel_id.to_string(),
+                                                                thread_id: event["thread_ts"].as_str().map(|s| s.to_string()),
+                                                                parent_id: None,
+                                                                origin_event_id: None,
+                                                            };
+                                                            let adapter = adapter.clone();
+                                                            tokio::spawn(async move {
+                                                                if let Err(e) = adapter.send_message(&warn_channel, &user_message).await {
+                                                                    warn!(error = %e, "failed to send bot turn limit warning");
                                                                 }
-                                                                continue;
-                                                            }
+                                                            });
                                                         }
-                                                    } else if is_plain_user_message(subtype, msg_text) {
-                                                        tracker.on_human_message(&turn_key);
+                                                        continue;
                                                     }
                                                 }
 
@@ -1058,6 +1112,22 @@ pub async fn run_slack_adapter(
                                 _ => {}
                             }
                         }
+                        _ = ping_interval.tick() => {
+                            if socket_idle(
+                                last_inbound.elapsed(),
+                                std::time::Duration::from_secs(IDLE_TIMEOUT_SECS),
+                            ) {
+                                warn!(
+                                    idle_secs = last_inbound.elapsed().as_secs(),
+                                    "Slack Socket Mode idle past timeout (likely half-open), forcing reconnect"
+                                );
+                                break;
+                            }
+                            if let Err(e) = write.send(tungstenite::Message::Ping(Vec::new())).await {
+                                warn!(error = %e, "Slack Socket Mode ping failed, reconnecting");
+                                break;
+                            }
+                        }
                         _ = shutdown_rx.changed() => {
                             info!("Slack adapter received shutdown signal");
                             let _ = write.send(tungstenite::Message::Close(None)).await;
@@ -1067,12 +1137,16 @@ pub async fn run_slack_adapter(
                 }
             }
             Err(e) => {
-                error!("failed to connect to Slack Socket Mode: {e}");
+                error!(err = %e, backoff = backoff_secs, "failed to connect to Slack Socket Mode, retrying");
             }
         }
 
-        warn!("reconnecting to Slack Socket Mode in 5s...");
-        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+        warn!(backoff = backoff_secs, "reconnecting to Slack Socket Mode");
+        tokio::select! {
+            _ = tokio::time::sleep(std::time::Duration::from_secs(backoff_secs)) => {}
+            _ = shutdown_rx.changed() => { return Ok(()); }
+        }
+        backoff_secs = next_backoff(backoff_secs);
     }
 }
 
@@ -2216,3 +2290,35 @@ mod tests {
         assert!(adapter.renders_native_tables());
     }
 }
+
+#[cfg(test)]
+mod socket_keepalive_tests {
+    use super::{next_backoff, socket_idle, IDLE_TIMEOUT_SECS, MAX_BACKOFF_SECS};
+    use std::time::Duration;
+
+    /// Backoff doubles and caps, matching the gateway adapter (1,2,4,8,16,30,30…).
+    #[test]
+    fn backoff_doubles_then_caps() {
+        let mut b = 1u64;
+        let seq: Vec<u64> = (0..8)
+            .map(|_| {
+                let cur = b;
+                b = next_backoff(b);
+                cur
+            })
+            .collect();
+        assert_eq!(seq, vec![1, 2, 4, 8, 16, MAX_BACKOFF_SECS, MAX_BACKOFF_SECS, MAX_BACKOFF_SECS]);
+        assert_eq!(next_backoff(MAX_BACKOFF_SECS), MAX_BACKOFF_SECS);
+    }
+
+    /// A half-open socket (no inbound past the window) is detected; an active one
+    /// (recent inbound, e.g. a Slack ping) is not. This is the deaf-socket guard.
+    #[test]
+    fn idle_detects_half_open_at_boundary() {
+        let timeout = Duration::from_secs(IDLE_TIMEOUT_SECS);
+        assert!(!socket_idle(Duration::from_secs(0), timeout));
+        assert!(!socket_idle(Duration::from_secs(IDLE_TIMEOUT_SECS - 1), timeout));
+        assert!(socket_idle(Duration::from_secs(IDLE_TIMEOUT_SECS), timeout));
+        assert!(socket_idle(Duration::from_secs(IDLE_TIMEOUT_SECS + 10), timeout));
+    }
+}