fix(gateway/slack): send progress messages to correct thread (#3063)

Co-authored-by: Jneeee <jneeee@outlook.com>

gateway/run.py

@@ -2394,7 +2394,8 @@ class GatewayRunner:
                 history=history,
                 source=source,
                 session_id=session_entry.session_id,
-                session_key=session_key
+                session_key=session_key,
+                event_message_id=event.message_id,
             )
 
             # Stop persistent typing indicator now that the agent is done
@@ -4842,6 +4843,7 @@ class GatewayRunner:
         session_id: str,
         session_key: str = None,
         _interrupt_depth: int = 0,
+        event_message_id: Optional[str] = None,
     ) -> Dict[str, Any]:
         """
         Run the agent with the given message and context.
@@ -4978,7 +4980,12 @@ class GatewayRunner:
         
         # Background task to send progress messages
         # Accumulates tool lines into a single message that gets edited
-        _progress_metadata = {"thread_id": source.thread_id} if source.thread_id else None
+        # For DM top-level Slack messages, source.thread_id is None but the
+        # final reply will be threaded under the original message via reply_to.
+        # Use event_message_id as fallback so progress messages land in the
+        # same thread as the final response instead of going to the DM root.
+        _progress_thread_id = source.thread_id or event_message_id
+        _progress_metadata = {"thread_id": _progress_thread_id} if _progress_thread_id else None
 
         async def send_progress_messages():
             if not progress_queue:
@@ -5093,7 +5100,7 @@ class GatewayRunner:
         # Bridge sync status_callback → async adapter.send for context pressure
         _status_adapter = self.adapters.get(source.platform)
         _status_chat_id = source.chat_id
-        _status_thread_metadata = {"thread_id": source.thread_id} if source.thread_id else None
+        _status_thread_metadata = {"thread_id": _progress_thread_id} if _progress_thread_id else None
 
         def _status_callback_sync(event_type: str, message: str) -> None:
             if not _status_adapter:
@@ -5174,7 +5181,7 @@ class GatewayRunner:
                             adapter=_adapter,
                             chat_id=source.chat_id,
                             config=_consumer_cfg,
-                            metadata={"thread_id": source.thread_id} if source.thread_id else None,
+                            metadata={"thread_id": _progress_thread_id} if _progress_thread_id else None,
                         )
                         _stream_delta_cb = _stream_consumer.on_delta
                         stream_consumer_holder[0] = _stream_consumer

tests/gateway/test_slack.py

@@ -946,3 +946,100 @@ class TestFallbackPreservesThreadContext:
 
         call_kwargs = adapter._app.client.chat_postMessage.call_args.kwargs
         assert "important screenshot" in call_kwargs["text"]
+
+
+# ---------------------------------------------------------------------------
+# TestProgressMessageThread
+# ---------------------------------------------------------------------------
+
+class TestProgressMessageThread:
+    """Verify that progress messages go to the correct thread.
+
+    Issue #2954: For Slack DM top-level messages, source.thread_id is None
+    but the final reply is threaded under the user's message via reply_to.
+    Progress messages must use the same thread anchor (the original message's
+    ts) so they appear in the thread instead of the DM root.
+    """
+
+    @pytest.mark.asyncio
+    async def test_dm_toplevel_progress_uses_message_ts_as_thread(self, adapter):
+        """Progress messages for a top-level DM should go into the reply thread."""
+        # Simulate a top-level DM: no thread_ts in the event
+        event = {
+            "channel": "D_DM",
+            "channel_type": "im",
+            "user": "U_USER",
+            "text": "Hello bot",
+            "ts": "1234567890.000001",
+            # No thread_ts — this is a top-level DM
+        }
+
+        captured_events = []
+        adapter.handle_message = AsyncMock(side_effect=lambda e: captured_events.append(e))
+
+        # Patch _resolve_user_name to avoid async Slack API call
+        with patch.object(adapter, "_resolve_user_name", new=AsyncMock(return_value="testuser")):
+            await adapter._handle_slack_message(event)
+
+        assert len(captured_events) == 1
+        msg_event = captured_events[0]
+        source = msg_event.source
+
+        # For a top-level DM: source.thread_id should remain None
+        # (session keying must not be affected)
+        assert source.thread_id is None, (
+            "source.thread_id must stay None for top-level DMs "
+            "so they share one continuous session"
+        )
+
+        # The message_id should be the event's ts — this is what the gateway
+        # passes as event_message_id so progress messages can thread correctly
+        assert msg_event.message_id == "1234567890.000001", (
+            "message_id must equal the event ts so _run_agent can use it as "
+            "the fallback thread anchor for progress messages"
+        )
+
+        # Verify that the Slack send() method correctly threads a message
+        # when metadata contains thread_id equal to the original ts
+        adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "reply_ts"})
+        result = await adapter.send(
+            chat_id="D_DM",
+            content="⚙️ working...",
+            metadata={"thread_id": msg_event.message_id},
+        )
+        assert result.success
+        call_kwargs = adapter._app.client.chat_postMessage.call_args[1]
+        assert call_kwargs.get("thread_ts") == "1234567890.000001", (
+            "send() must pass thread_ts when metadata has thread_id, "
+            "ensuring progress messages land in the thread"
+        )
+
+    @pytest.mark.asyncio
+    async def test_channel_mention_progress_uses_thread_ts(self, adapter):
+        """Progress messages for a channel @mention should go into the reply thread."""
+        # Simulate an @mention in a channel: the event ts becomes the thread anchor
+        event = {
+            "channel": "C_CHAN",
+            "channel_type": "channel",
+            "user": "U_USER",
+            "text": f"<@U_BOT> help me",
+            "ts": "2000000000.000001",
+            # No thread_ts — top-level channel message
+        }
+
+        captured_events = []
+        adapter.handle_message = AsyncMock(side_effect=lambda e: captured_events.append(e))
+
+        with patch.object(adapter, "_resolve_user_name", new=AsyncMock(return_value="testuser")):
+            await adapter._handle_slack_message(event)
+
+        assert len(captured_events) == 1
+        msg_event = captured_events[0]
+        source = msg_event.source
+
+        # For channel @mention: thread_id should equal the event ts (fallback)
+        assert source.thread_id == "2000000000.000001", (
+            "source.thread_id must equal the event ts for channel messages "
+            "so each @mention starts its own thread"
+        )
+        assert msg_event.message_id == "2000000000.000001"