fix(slack): respect reply_in_thread and reply_to_mode for channel replies

gateway/platforms/slack.py

@@ -9,6 +9,7 @@
 """
 
 import asyncio
+import contextvars
 import json
 import logging
 import os
@@ -17,6 +18,15 @@
 from dataclasses import dataclass, field
 from typing import Dict, Optional, Any, Tuple
 
+# Per-task flag: True when the current inbound message is a top-level
+# channel message that should NOT create a thread reply.  Set by
+# _handle_slack_message, read by _resolve_thread_ts and send_typing.
+# Using ContextVar (not an instance variable) so concurrent asyncio
+# tasks each get their own value without cross-contamination.
+_force_channel_reply: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "_force_channel_reply", default=False
+)
+
 try:
     from slack_bolt.async_app import AsyncApp
     from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
@@ -355,6 +365,13 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
         if not thread_ts:
             return  # Can only set status in a thread context
 
+        # Skip assistant thread status for top-level channel messages when
+        # reply_in_thread is disabled.  The setStatus API activates an
+        # assistant thread, which would force replies into a thread even
+        # when _resolve_thread_ts returns None.
+        if _force_channel_reply.get():
+            return
+
         try:
             await self._get_client(chat_id).assistant_threads_setStatus(
                 channel_id=chat_id,
@@ -381,9 +398,13 @@ def _resolve_thread_ts(
         thread replies.  Messages that originate inside an existing thread are
         always replied to in-thread to preserve conversation context.
         """
-        # When reply_in_thread is disabled (default: True for backward compat),
-        # only thread messages that are already part of an existing thread.
-        if not self.config.extra.get("reply_in_thread", True):
+        # When reply_in_thread is disabled (default: True for backward compat)
+        # or reply_to_mode is "off", suppress threading for top-level messages.
+        # _force_channel_reply is set per-message by _handle_slack_message
+        # (True for top-level, False for genuine thread replies).
+        if not self.config.extra.get("reply_in_thread", True) or getattr(self.config, "reply_to_mode", None) == "off":
+            if _force_channel_reply.get():
+                return None
             existing_thread = (metadata or {}).get("thread_id") or (metadata or {}).get("thread_ts")
             return existing_thread or None
 
@@ -1157,6 +1178,18 @@ async def _handle_slack_message(self, event: dict) -> None:
         # Resolve user display name (cached after first lookup)
         user_name = await self._resolve_user_name(user_id, chat_id=channel_id)
 
+        # When reply_in_thread is disabled or reply_to_mode is off, mark
+        # top-level messages so _resolve_thread_ts and send_typing suppress
+        # threading.  The flag is per-message, reset on each inbound event.
+        # Safe in single-threaded asyncio.
+        _force_channel_reply.set(
+            not is_thread_reply
+            and (
+                not self.config.extra.get("reply_in_thread", True)
+                or getattr(self.config, "reply_to_mode", None) == "off"
+            )
+        )
+
         # Build source
         source = self.build_source(
             chat_id=channel_id,

tests/gateway/test_slack.py

@@ -59,7 +59,7 @@ def _ensure_slack_mock():
 import gateway.platforms.slack as _slack_mod
 _slack_mod.SLACK_AVAILABLE = True
 
-from gateway.platforms.slack import SlackAdapter  # noqa: E402
+from gateway.platforms.slack import SlackAdapter, _force_channel_reply  # noqa: E402
 
 
 # ---------------------------------------------------------------------------
@@ -590,6 +590,63 @@ async def test_uses_thread_ts_fallback(self, adapter):
             status="is thinking...",
         )
 
+    @pytest.mark.asyncio
+    async def test_skips_status_when_force_channel_reply(self, adapter):
+        """When _force_channel_reply is set, send_typing skips setStatus."""
+        _force_channel_reply.set(True)
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        await adapter.send_typing("C123", metadata={"thread_id": "ts1"})
+        adapter._app.client.assistant_threads_setStatus.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# TestResolveThreadTs — thread routing logic
+# ---------------------------------------------------------------------------
+
+
+class TestResolveThreadTs:
+    """Test _resolve_thread_ts with _force_channel_reply flag."""
+
+    def test_default_returns_thread_id(self, adapter):
+        """Default config: returns thread_id from metadata."""
+        result = adapter._resolve_thread_ts("reply_ts", {"thread_id": "parent_ts"})
+        assert result == "parent_ts"
+
+    def test_force_channel_reply_returns_none(self, adapter):
+        """Top-level with reply_in_thread=false: returns None via flag."""
+        adapter.config.extra["reply_in_thread"] = False
+        _force_channel_reply.set(True)
+        result = adapter._resolve_thread_ts(None, {"thread_id": "ts123"})
+        assert result is None
+
+    def test_thread_reply_stays_in_thread_when_disabled(self, adapter):
+        """Genuine thread reply with reply_in_thread=false: stays in thread."""
+        adapter.config.extra["reply_in_thread"] = False
+        _force_channel_reply.set(False)
+        result = adapter._resolve_thread_ts(None, {"thread_id": "parent_ts"})
+        assert result == "parent_ts"
+
+    def test_reply_to_mode_off_with_flag(self, adapter):
+        """reply_to_mode=off + _force_channel_reply: returns None."""
+        adapter.config.reply_to_mode = "off"
+        _force_channel_reply.set(True)
+        result = adapter._resolve_thread_ts(None, {"thread_id": "ts123"})
+        assert result is None
+
+    def test_no_flag_preserves_thread(self, adapter):
+        """Without _force_channel_reply: thread_id preserved (genuine thread)."""
+        adapter.config.extra["reply_in_thread"] = False
+        _force_channel_reply.set(False)
+        result = adapter._resolve_thread_ts(None, {"thread_id": "parent_ts"})
+        assert result == "parent_ts"
+
+    def test_default_config_fully_backward_compatible(self, adapter):
+        """Default config (reply_in_thread=True): entire block skipped, thread preserved."""
+        # Default adapter has reply_in_thread not set (defaults True)
+        # ContextVar defaults False — should not matter, block is skipped
+        result = adapter._resolve_thread_ts("reply_ts", {"thread_id": "parent_ts"})
+        assert result == "parent_ts"
+
 
 # ---------------------------------------------------------------------------
 # TestFormatMessage — Markdown → mrkdwn conversion