fix(dispatch): idle eviction, config validation, avoid clone, timestamp precision

- Add 5-min idle timeout to consumer_loop to prevent per-thread handle/task
  leak (unbounded growth from one-shot thread keys like Slack non-thread msgs)
- Validate max_buffered_messages > 0 at config load time (prevents panic from
  tokio::sync::mpsc::channel(0))
- Use into_iter() in dispatch_batch to avoid deep-copying extra_blocks
  (may contain base64 image data)
- Add TODO comment for gateway multibot detection
- Use real milliseconds in now_iso8601() via dur.subsec_millis()

Co-authored-by: 超渡法師 <chaodu@openab.dev>

Author: 超渡法師

src/config.rs

@@ -504,6 +504,18 @@ fn parse_config(raw: &str, source: &str) -> anyhow::Result<Config> {
     let expanded = expand_env_vars(raw);
     let config: Config = toml::from_str(&expanded)
         .map_err(|e| anyhow::anyhow!("failed to parse config from {source}: {e}"))?;
+
+    // Validate max_buffered_messages > 0 (tokio::sync::mpsc::channel panics on 0).
+    if let Some(ref d) = config.discord {
+        anyhow::ensure!(d.max_buffered_messages > 0, "discord.max_buffered_messages must be > 0");
+    }
+    if let Some(ref s) = config.slack {
+        anyhow::ensure!(s.max_buffered_messages > 0, "slack.max_buffered_messages must be > 0");
+    }
+    if let Some(ref g) = config.gateway {
+        anyhow::ensure!(g.max_buffered_messages > 0, "gateway.max_buffered_messages must be > 0");
+    }
+
     Ok(config)
 }
 

src/dispatch.rs

@@ -11,10 +11,10 @@
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use tracing::{error, info, info_span, warn};
+use tracing::{debug, error, info, info_span, warn};
 
 use crate::adapter::{AdapterRouter, ChannelRef, ChatAdapter, MessageRef};
 use crate::acp::ContentBlock;
@@ -255,6 +255,12 @@ impl Dispatcher {
 // consumer_loop
 // ---------------------------------------------------------------------------
 
+/// Idle timeout for per-thread consumer tasks. When no message arrives within
+/// this window the consumer exits, allowing `per_thread` map cleanup on the
+/// next `submit` (via `SendError` → `try_evict_locked`). Prevents unbounded
+/// task/memory growth from one-shot thread keys (e.g. Slack non-thread messages).
+const CONSUMER_IDLE_TIMEOUT: Duration = Duration::from_secs(300); // 5 min
+
 #[allow(clippy::too_many_arguments)]
 async fn consumer_loop(
     thread_key: String,
@@ -271,13 +277,26 @@ async fn consumer_loop(
 
     loop {
         // I1: block until at least one message arrives (zero latency for first message).
+        // Idle timeout: if no message arrives within CONSUMER_IDLE_TIMEOUT the
+        // consumer exits, freeing the task and mpsc. The next `submit` for this
+        // thread_key will observe `SendError`, evict the stale entry, and lazily
+        // spawn a fresh consumer (§2.5 generation check prevents mis-eviction).
         let first = match pending.take() {
             Some(msg) => msg,
-            None => match rx.recv().await {
-                Some(msg) => msg,
-                // All senders dropped → either shutdown() cleared the map, or
-                // cancel_buffered() removed our entry. Exit the loop.
-                None => break,
+            None => match tokio::time::timeout(CONSUMER_IDLE_TIMEOUT, rx.recv()).await {
+                Ok(Some(msg)) => msg,
+                Ok(None) => {
+                    // All senders dropped → shutdown() or cancel_buffered().
+                    break;
+                }
+                Err(_elapsed) => {
+                    debug!(
+                        thread_key = %thread_key,
+                        channel = %thread_channel.channel_id,
+                        "consumer idle timeout, exiting"
+                    );
+                    break;
+                }
             },
         };
 
@@ -340,19 +359,23 @@ async fn dispatch_batch(
     )
     .await;
 
-    // Collect per-event observability data.
+    // Collect per-event observability data (before consuming the batch).
     let tokens_per_event: Vec<usize> = batch.iter().map(|m| m.estimated_tokens).collect();
     let wait_ms: Vec<u128> = batch
         .iter()
         .map(|m| m.arrived_at.elapsed().as_millis())
         .collect();
-    let senders: Vec<&str> = batch.iter().map(|m| m.sender_name.as_str()).collect();
+    let senders: Vec<String> = batch.iter().map(|m| m.sender_name.clone()).collect();
+
+    // Anchor reactions on the last message in the batch (before consuming).
+    let trigger_msg = batch.last().unwrap().trigger_msg.clone();
 
     // Pack all arrival events into one Vec<ContentBlock> (§3.3).
+    // Uses into_iter() to avoid deep-copying extra_blocks (may contain base64 image data).
     let mut content_blocks: Vec<ContentBlock> = Vec::new();
-    for msg in &batch {
+    for msg in batch {
         let mut event_blocks =
-            AdapterRouter::pack_arrival_event(&msg.sender_json, &msg.prompt, msg.extra_blocks.clone());
+            AdapterRouter::pack_arrival_event(&msg.sender_json, &msg.prompt, msg.extra_blocks);
         content_blocks.append(&mut event_blocks);
     }
     let packed_block_count = content_blocks.len();
@@ -367,8 +390,6 @@ async fn dispatch_batch(
         return;
     }
 
-    // Anchor reactions on the last message in the batch.
-    let trigger_msg = batch.last().unwrap().trigger_msg.clone();
     let reactions_config = router.reactions_config().clone();
     let reactions = Arc::new(StatusReactionController::new(
         reactions_config.enabled,

src/gateway.rs

@@ -564,6 +564,7 @@ pub async fn run_gateway_adapter(
                                             trigger_msg,
                                             arrived_at: std::time::Instant::now(),
                                             estimated_tokens,
+                                            // TODO: implement gateway multibot detection
                                             other_bot_present: false,
                                         };
                                         if let Err(e) = dispatcher

src/timestamp.rs

@@ -42,13 +42,12 @@ pub fn slack_ts_to_iso8601(ts: &str) -> String {
     unix_to_iso8601(secs, ms)
 }
 
-/// Current wall-clock instant as ISO 8601 UTC (millisecond precision, always `.000Z`).
+/// Current wall-clock instant as ISO 8601 UTC with millisecond precision.
 pub fn now_iso8601() -> String {
-    let secs = SystemTime::now()
+    let dur = SystemTime::now()
         .duration_since(UNIX_EPOCH)
-        .unwrap_or_default()
-        .as_secs();
-    unix_to_iso8601(secs, 0)
+        .unwrap_or_default();
+    unix_to_iso8601(dur.as_secs(), (dur.subsec_millis()) as u64)
 }
 
 #[cfg(test)]
@@ -79,10 +78,11 @@ mod tests {
     #[test]
     fn now_iso8601_has_expected_shape() {
         let s = now_iso8601();
-        // YYYY-MM-DDTHH:MM:SS.000Z = 24 chars
+        // YYYY-MM-DDTHH:MM:SS.mmmZ = 24 chars
         assert_eq!(s.len(), 24);
-        assert!(s.ends_with(".000Z"));
+        assert!(s.ends_with('Z'));
         assert_eq!(&s[4..5], "-");
         assert_eq!(&s[10..11], "T");
+        assert_eq!(&s[19..20], ".");
     }
 }