Let an engaged Slack channel keep listening, and read the thread it joined

crates/daemon/src/service.rs

@@ -161,6 +161,33 @@ impl SlackProgress {
     }
 }
 
+/// Where a Slack channel keeps listening after it has been addressed.
+///
+/// A bot that must be `@`-mentioned for every turn cannot hold a conversation:
+/// the person asking has to keep re-addressing an participant that is visibly
+/// already in the room. Once engaged, the bot behaves like a participant —
+/// within a boundary the operator sets, because "answers everything in this
+/// channel" is right for a dedicated channel and wrong for a busy shared one.
+///
+/// Anything past `Off` needs the `message.channels` event subscription (plus
+/// `message.groups` for private channels). Without it Slack never sends the
+/// untagged messages, and every mode behaves like `Off`.
+#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+pub enum SlackFollowUp {
+    /// Only direct mentions and DMs.
+    Off,
+    /// Keep answering inside a thread the bot was mentioned in.
+    #[default]
+    Thread,
+    /// Keep answering anywhere in a channel the bot has been mentioned in.
+    Channel,
+}
+
+fn default_thread_context() -> usize {
+    50
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ServiceChannelConfig {
     #[serde(default)]
@@ -180,6 +207,13 @@ pub struct ServiceChannelConfig {
     /// Slack only. Omitted definitions keep the default affordance.
     #[serde(default)]
     pub progress: SlackProgress,
+    /// Slack only. Where the bot keeps answering once it has been addressed.
+    #[serde(default)]
+    pub follow_up: SlackFollowUp,
+    /// Slack only. How many earlier messages of a thread to read when first
+    /// pulled into one. `0` reads none. Needs `channels:history`.
+    #[serde(default = "default_thread_context")]
+    pub thread_context: usize,
 }
 
 fn default_channel_enabled() -> bool {
@@ -509,6 +543,14 @@ pub fn put_channel(
             .as_ref()
             .map(|channel| channel.progress)
             .unwrap_or_default(),
+        follow_up: existing
+            .as_ref()
+            .map(|channel| channel.follow_up)
+            .unwrap_or_default(),
+        thread_context: existing
+            .as_ref()
+            .map(|channel| channel.thread_context)
+            .unwrap_or_else(default_thread_context),
     };
     service
         .channels
@@ -922,6 +964,8 @@ pub(crate) fn slack_config(
         allowed_workspaces: channel.allowed_workspaces.clone(),
         allowed_channels: channel.allowed_channels.clone(),
         progress: channel.progress,
+        follow_up: channel.follow_up,
+        thread_context: channel.thread_context,
     })
 }
 
@@ -986,6 +1030,8 @@ mod tests {
                     allowed_workspaces: Vec::new(),
                     allowed_channels: Vec::new(),
                     progress: Default::default(),
+                    follow_up: Default::default(),
+                    thread_context: default_thread_context(),
                 },
             )]),
         }
@@ -1470,6 +1516,28 @@ mod tests {
         assert_eq!(channels["c"].progress, SlackProgress::Placeholder);
     }
 
+    #[test]
+    fn a_slack_channel_chooses_where_it_keeps_listening() {
+        let dir = tempfile::tempdir().unwrap();
+        std::fs::write(
+            dir.path().join("bot.toml"),
+            "harness = \"codex\"\n\
+             [channels.a]\nkind = \"slack\"\nfollow_up = \"channel\"\nthread_context = 10\n\
+             [channels.b]\nkind = \"slack\"\nfollow_up = \"off\"\nthread_context = 0\n\
+             [channels.c]\nkind = \"slack\"\n",
+        )
+        .unwrap();
+
+        let channels = &load_definitions(dir.path()).unwrap()["bot"].channels;
+        assert_eq!(channels["a"].follow_up, SlackFollowUp::Channel);
+        assert_eq!(channels["a"].thread_context, 10);
+        assert_eq!(channels["b"].follow_up, SlackFollowUp::Off);
+        assert_eq!(channels["b"].thread_context, 0);
+        // A definition written before these options existed keeps working.
+        assert_eq!(channels["c"].follow_up, SlackFollowUp::Thread);
+        assert_eq!(channels["c"].thread_context, default_thread_context());
+    }
+
     #[test]
     fn service_put_preserves_channels_and_channel_crud_rotates_credentials() {
         let config = tempfile::tempdir().unwrap();
@@ -1836,6 +1904,7 @@ mod tests {
             services.join("chat.toml"),
             "harness = \"codex\"\n\
              [channels.bot]\nkind = \"slack\"\nprogress = \"reaction\"\n\
+             follow_up = \"channel\"\nthread_context = 7\n\
              app_token = \"xapp-1\"\nbot_token = \"xoxb-1\"\n",
         )
         .unwrap();
@@ -1861,6 +1930,8 @@ mod tests {
 
         let stored = &load_definitions(&services).unwrap()["chat"].channels["bot"];
         assert_eq!(stored.progress, SlackProgress::Reaction);
+        assert_eq!(stored.follow_up, SlackFollowUp::Channel);
+        assert_eq!(stored.thread_context, 7);
         assert_eq!(stored.allowed_workspaces, vec!["T9".to_string()]);
     }
 

crates/daemon/src/service/ingress.rs

@@ -168,6 +168,33 @@ impl ServiceIngress {
         Ok(self.submit_tracked(request).await?.session)
     }
 
+    /// Whether this channel already routes a conversation for `session_key`.
+    ///
+    /// A channel asks this to decide whether it is *already engaged* — which
+    /// is what lets it answer a message that did not address it directly.
+    /// Engagement is exactly "a session exists for this key", so it needs no
+    /// second record that could disagree with the routing table.
+    pub(super) async fn has_session(&self, session_key: &str) -> bool {
+        let lookup_key = format!("{}:{session_key}", self.channel_id);
+        let state = self.shared.state.lock().await;
+        state.sessions.contains_key(&lookup_key)
+            || (self.channel_id == "http" && state.sessions.contains_key(session_key))
+    }
+
+    /// Whether this channel routes any conversation whose key starts with
+    /// `prefix` — "have I been engaged anywhere in this Slack channel", where
+    /// the caller's keys nest the channel above the thread.
+    pub(super) async fn has_session_under(&self, prefix: &str) -> bool {
+        let lookup_prefix = format!("{}:{prefix}", self.channel_id);
+        self.shared
+            .state
+            .lock()
+            .await
+            .sessions
+            .keys()
+            .any(|key| key.starts_with(&lookup_prefix))
+    }
+
     /// Submit a native channel delivery and retain the transcript position at
     /// which its turn began. Long-lived adapters use this cursor to avoid
     /// mistaking the previous turn's final answer for the new one.
@@ -936,6 +963,32 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn engagement_is_having_a_session_for_the_thread_or_the_channel() {
+        let shared = shared_for_delivery_tests().await;
+        let ingress = ServiceIngress::new("bot".to_string(), shared.clone());
+        shared
+            .state
+            .lock()
+            .await
+            .sessions
+            .insert("bot:T1:C1:111.11".to_string(), "s1".to_string());
+
+        // Thread follow-up: only the thread that already has a conversation.
+        assert!(ingress.has_session("T1:C1:111.11").await);
+        assert!(!ingress.has_session("T1:C1:999.99").await);
+
+        // Channel follow-up: any thread in the same Slack channel counts.
+        assert!(ingress.has_session_under("T1:C1:").await);
+        assert!(!ingress.has_session_under("T1:C2:").await);
+
+        // A key belonging to another channel of the same service must not
+        // read as engagement here — channels own their own conversations.
+        let other = ServiceIngress::new("other-bot".to_string(), shared);
+        assert!(!other.has_session("T1:C1:111.11").await);
+        assert!(!other.has_session_under("T1:C1:").await);
+    }
+
     #[test]
     fn progress_reports_the_approval_a_turn_is_stopped_at() {
         let at = chrono::Utc::now();