fix(slack): keep transient status in assistant UI

gateway/platforms/slack.py

@@ -1270,34 +1270,112 @@ async def edit_message(
             )
             return SendResult(success=False, error=str(e))
 
-    async def send_typing(self, chat_id: str, metadata=None) -> None:
-        """Show a typing/status indicator using assistant.threads.setStatus.
-
-        Displays "is thinking..." next to the bot name in a thread.
-        Requires the assistant:write or chat:write scope.
-        Auto-clears when the bot sends a reply to the thread.
-        """
-        if not self._app:
-            return
-
-        thread_ts = None
-        if metadata:
-            thread_ts = metadata.get("thread_id") or metadata.get("thread_ts")
-
-        if not thread_ts:
-            return  # Can only set status in a thread context
-
+    @staticmethod
+    def _status_thread_ts(metadata: Optional[Dict[str, Any]] = None) -> Optional[str]:
+        if not metadata:
+            return None
+        return metadata.get("thread_id") or metadata.get("thread_ts")
+
+    @staticmethod
+    def _assistant_activity_status(text: str) -> str:
+        """Turn heartbeat detail into a truthful Slack Assistant status."""
+        lowered = text.lower()
+        activity_map = [
+            (("session_search", "recall"), "is recalling prior context"),
+            (("search_files", "grep", "find"), "is searching files"),
+            (("read_file",), "is reading files"),
+            (("web_search",), "is searching the web"),
+            (("web_extract",), "is reading web pages"),
+            (("browser", "mcp_chrome_devtools"), "is using the browser"),
+            (("terminal", "process"), "is running a command"),
+            (("delegate_task",), "is coordinating agents"),
+            (("todo",), "is updating the task list"),
+            (("memory", "skill_manage", "skill_view"), "is checking operating memory"),
+            (("mcp_zoblin_connections_close", "close_"), "is checking Close"),
+            (("mcp_zoblin_connections_growth_n8n", "n8n"), "is checking n8n"),
+            (("mcp_zoblin_connections_railway", "railway"), "is checking Railway"),
+            (("mcp_google_sheets", "sheet"), "is checking Sheets"),
+            (("mcp_ga4", "analytics"), "is checking analytics"),
+            (("waiting for non-streaming", "waiting for model"), "is waiting on the model"),
+            (("receiving stream", "stream response"), "is drafting the reply"),
+        ]
+        for needles, status in activity_map:
+            if any(needle in lowered for needle in needles):
+                return status
+        return "is working through the request"
+
+    @staticmethod
+    def _assistant_status_text(content: str) -> str:
+        """Convert gateway status text into Slack Assistant status grammar."""
+        text = str(content or "").strip()
+        text = re.sub(r"^[\s:!\w-]+:\s*", "", text) if text.startswith(":") else text
+        text = re.sub(r"^[\s📦🗜️⏳⚠️✅✓•*-]+", "", text).strip()
+        text = re.sub(r"\s+", " ", text)
+        lowered = text.lower()
+
+        if lowered.startswith("preflight compression") or lowered.startswith("compacting context"):
+            return "is compacting context..."
+        if lowered.startswith("gateway restarting"):
+            return ("is restarting" + text[len("Gateway restarting"):])[:96]
+        if lowered.startswith("gateway shutting down"):
+            return ("is shutting down" + text[len("Gateway shutting down"):])[:96]
+        if lowered.startswith("working") or lowered.startswith("still working"):
+            return SlackAdapter._assistant_activity_status(text)[:96]
+
+        if text and not text.lower().startswith(("is ", "has ", "was ")):
+            text = f"is {text[0].lower()}{text[1:]}"
+        return text[:96]
+
+    async def _set_assistant_thread_status(
+        self,
+        chat_id: str,
+        thread_ts: Optional[str],
+        status: str,
+    ) -> bool:
+        if not self._app or not thread_ts:
+            return False
         self._active_status_threads[chat_id] = thread_ts
         try:
             await self._get_client(chat_id).assistant_threads_setStatus(
                 channel_id=chat_id,
                 thread_ts=thread_ts,
-                status="is thinking...",
+                status=status,
             )
+            return True
         except Exception as e:
-            # Silently ignore — may lack assistant:write scope or not be
-            # in an assistant-enabled context. Falls back to reactions.
             logger.debug("[Slack] assistant.threads.setStatus failed: %s", e)
+            return False
+
+    async def send_or_update_status(
+        self,
+        chat_id: str,
+        status_key: str,
+        content: str,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Use Slack Assistant thread status for transient gateway state.
+
+        This intentionally never falls back to chat_postMessage: a permanent
+        Slack message would recreate the thread spam this path exists to avoid.
+        """
+        thread_ts = self._status_thread_ts(metadata)
+        status = self._assistant_status_text(content)
+        ok = await self._set_assistant_thread_status(chat_id, thread_ts, status)
+        return SendResult(
+            success=ok or thread_ts is None,
+            message_id=None,
+            error=None if ok or thread_ts is None else "assistant_status_failed",
+        )
+
+    async def send_typing(self, chat_id: str, metadata=None) -> None:
+        """Show a typing/status indicator using assistant.threads.setStatus.
+
+        Displays "is thinking..." next to the bot name in a thread.
+        Requires the assistant:write or chat:write scope.
+        Auto-clears when the bot sends a reply to the thread.
+        """
+        thread_ts = self._status_thread_ts(metadata)
+        await self._set_assistant_thread_status(chat_id, thread_ts, "is thinking...")
 
     async def stop_typing(self, chat_id: str, metadata=None) -> None:
         """Clear the assistant thread status indicator."""

gateway/run.py

@@ -4031,7 +4031,16 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                     adapter=adapter,
                 )
 
-                result = await adapter.send(chat_id, msg, metadata=metadata)
+                if platform == Platform.SLACK:
+                    result = await _send_or_update_status_coro(
+                        adapter,
+                        chat_id,
+                        "gateway_restart" if self._restart_requested else "gateway_shutdown",
+                        msg,
+                        metadata,
+                    )
+                else:
+                    result = await adapter.send(chat_id, msg, metadata=metadata)
                 if result is not None and getattr(result, "success", True) is False:
                     logger.debug(
                         "Failed to send shutdown notification to %s:%s: %s",
@@ -4085,7 +4094,15 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                     home.thread_id,
                     adapter=adapter,
                 )
-                if metadata:
+                if platform == Platform.SLACK:
+                    result = await _send_or_update_status_coro(
+                        adapter,
+                        str(home.chat_id),
+                        "gateway_restart" if self._restart_requested else "gateway_shutdown",
+                        msg,
+                        metadata,
+                    )
+                elif metadata:
                     result = await adapter.send(str(home.chat_id), msg, metadata=metadata)
                 else:
                     result = await adapter.send(str(home.chat_id), msg)
@@ -14926,28 +14943,40 @@ async def _notify_long_running():
                 _heartbeat_text = f"⏳ Working — {_elapsed_mins} min{_status_detail}"
                 try:
                     _notify_res = None
-                    if _heartbeat_msg_id:
-                        try:
-                            _notify_res = await _notify_adapter.edit_message(
-                                source.chat_id,
-                                _heartbeat_msg_id,
-                                _heartbeat_text,
-                            )
-                        except Exception as _ee:
-                            logger.debug("Heartbeat edit failed: %s", _ee)
-                            _notify_res = None
-                    if not (_notify_res and getattr(_notify_res, "success", False)):
-                        _notify_res = await _notify_adapter.send(
+                    if source.platform == Platform.SLACK:
+                        # Slack has a native non-permanent Assistant thread
+                        # status. Use it for heartbeat state instead of posting
+                        # durable "Working" messages into the thread.
+                        _notify_res = await _send_or_update_status_coro(
+                            _notify_adapter,
                             source.chat_id,
+                            "long_running",
                             _heartbeat_text,
-                            metadata=_status_thread_metadata,
+                            _status_thread_metadata,
                         )
-                        if getattr(_notify_res, "success", False) and getattr(
-                            _notify_res, "message_id", None
-                        ):
-                            _heartbeat_msg_id = str(_notify_res.message_id)
-                            if _cleanup_progress:
-                                _cleanup_msg_ids.append(_heartbeat_msg_id)
+                    else:
+                        if _heartbeat_msg_id:
+                            try:
+                                _notify_res = await _notify_adapter.edit_message(
+                                    source.chat_id,
+                                    _heartbeat_msg_id,
+                                    _heartbeat_text,
+                                )
+                            except Exception as _ee:
+                                logger.debug("Heartbeat edit failed: %s", _ee)
+                                _notify_res = None
+                        if not (_notify_res and getattr(_notify_res, "success", False)):
+                            _notify_res = await _notify_adapter.send(
+                                source.chat_id,
+                                _heartbeat_text,
+                                metadata=_status_thread_metadata,
+                            )
+                            if getattr(_notify_res, "success", False) and getattr(
+                                _notify_res, "message_id", None
+                            ):
+                                _heartbeat_msg_id = str(_notify_res.message_id)
+                                if _cleanup_progress:
+                                    _cleanup_msg_ids.append(_heartbeat_msg_id)
                 except Exception as _ne:
                     logger.debug("Long-running notification error: %s", _ne)
 

tests/gateway/test_slack.py

@@ -1755,6 +1755,113 @@ async def test_message_edits_ignored(self, adapter):
 class TestSendTyping:
     """Test typing indicator via assistant.threads.setStatus."""
 
+    @pytest.mark.asyncio
+    async def test_send_or_update_status_sets_assistant_status_without_posting(self, adapter):
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "msg"})
+
+        result = await adapter.send_or_update_status(
+            "C123",
+            "compression",
+            "📦 Preflight compression: ~264,618 tokens >= 217,600 threshold. This may take a moment.",
+            metadata={"thread_id": "parent_ts"},
+        )
+
+        assert result.success
+        assert result.message_id is None
+        adapter._app.client.assistant_threads_setStatus.assert_called_once_with(
+            channel_id="C123",
+            thread_ts="parent_ts",
+            status="is compacting context...",
+        )
+        adapter._app.client.chat_postMessage.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_send_or_update_status_noops_without_thread(self, adapter):
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "msg"})
+
+        result = await adapter.send_or_update_status(
+            "C123",
+            "compression",
+            "🗜️ Compacting context — summarizing earlier conversation so I can continue...",
+        )
+
+        assert result.success
+        assert result.message_id is None
+        adapter._app.client.assistant_threads_setStatus.assert_not_called()
+        adapter._app.client.chat_postMessage.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_send_or_update_status_handles_api_error_without_posting(self, adapter):
+        adapter._app.client.assistant_threads_setStatus = AsyncMock(
+            side_effect=Exception("missing_scope")
+        )
+        adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "msg"})
+
+        result = await adapter.send_or_update_status(
+            "C123",
+            "long_running",
+            "⏳ Working — 3 min — iteration 7/200",
+            metadata={"thread_ts": "parent_ts"},
+        )
+
+        assert not result.success
+        assert result.error == "assistant_status_failed"
+        adapter._app.client.chat_postMessage.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_send_or_update_status_uses_real_activity_words(self, adapter):
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "msg"})
+
+        await adapter.send_or_update_status(
+            "C123",
+            "long_running",
+            "⏳ Working — 3 min — iteration 7/200, session_search",
+            metadata={"thread_id": "parent_ts"},
+        )
+        await adapter.send_or_update_status(
+            "C123",
+            "long_running",
+            "⏳ Working — 6 min — iteration 12/200, read_file",
+            metadata={"thread_id": "parent_ts"},
+        )
+        await adapter.send_or_update_status(
+            "C123",
+            "long_running",
+            "⏳ Working — 9 min — mcp_zoblin_connections_close_fetch_lead_context",
+            metadata={"thread_id": "parent_ts"},
+        )
+
+        statuses = [call.kwargs["status"] for call in adapter._app.client.assistant_threads_setStatus.call_args_list]
+        assert statuses == [
+            "is recalling prior context",
+            "is reading files",
+            "is checking Close",
+        ]
+        adapter._app.client.chat_postMessage.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_gateway_restart_status_wording_is_natural(self, adapter):
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "msg"})
+
+        result = await adapter.send_or_update_status(
+            "C123",
+            "gateway_restart",
+            "⚠️ Gateway restarting — Your current task will be interrupted.",
+            metadata={"thread_id": "parent_ts"},
+        )
+
+        assert result.success
+        adapter._app.client.assistant_threads_setStatus.assert_called_once_with(
+            channel_id="C123",
+            thread_ts="parent_ts",
+            status="is restarting — Your current task will be interrupted.",
+        )
+        adapter._app.client.chat_postMessage.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_sets_status_in_thread(self, adapter):
         adapter._app.client.assistant_threads_setStatus = AsyncMock()