fix(slack): progress/status card lifecycle — edit in place, always clear, survive transients

contributors/emails/15167896+2001Y@users.noreply.github.com

@@ -0,0 +1 @@
+2001Y

contributors/emails/boumagent@gmail.com

@@ -0,0 +1 @@
+patp

contributors/emails/hello@jeromeiveson.com

@@ -0,0 +1 @@
+Trantor-develops

contributors/emails/rt.cms012@gmail.com

@@ -0,0 +1 @@
+trac3r00

gateway/run.py

@@ -529,10 +529,31 @@ async def _send_or_update_status_coro(adapter, chat_id, status_key, content, met
     return await adapter.send(chat_id, content, metadata=metadata)
 
 
-def _resolve_progress_thread_id(platform: Any, source_thread_id: Any, event_message_id: Any) -> Optional[str]:
-    """Return thread/root ID that progress/status bubbles should target."""
+def _resolve_progress_thread_id(
+    platform: Any,
+    source_thread_id: Any,
+    event_message_id: Any,
+    *,
+    reply_in_thread: bool = True,
+) -> Optional[str]:
+    """Return thread/root ID that progress/status bubbles should target.
+
+    ``reply_in_thread=False`` (Slack ``platforms.slack.extra.reply_in_thread``)
+    disables the synthetic-thread fallback: progress messages must not create
+    a thread the final flat reply would then inherit. A source.thread_id equal
+    to the event's own message id is the adapter's synthetic session-keying
+    thread, not a real thread — treat it as "no thread" too (#18859).
+    """
     platform_value = getattr(platform, "value", platform)
     platform_key = str(platform_value or "").lower()
+    if not reply_in_thread:
+        if (
+            source_thread_id
+            and event_message_id
+            and str(source_thread_id) == str(event_message_id)
+        ):
+            return None
+        return str(source_thread_id) if source_thread_id else None
     if source_thread_id:
         return str(source_thread_id)
     if platform_key in {"slack", "mattermost"} and event_message_id:
@@ -16285,6 +16306,10 @@ def _thread_metadata_for_target(
                 metadata["direct_messages_topic_id"] = tid
             if reply_to_message_id is not None:
                 metadata["telegram_reply_to_message_id"] = str(reply_to_message_id)
+        if platform == Platform.SLACK and reply_to_message_id is not None:
+            # Slack's reply_in_thread=false path uses message_id to distinguish
+            # real existing threads from synthetic top-level session keys.
+            metadata["message_id"] = str(reply_to_message_id)
         return metadata
 
     @staticmethod
@@ -20042,13 +20067,38 @@ def progress_callback(event_type: str, tool_name: str = None, preview: str = Non
         # - Feishu only honors reply_in_thread when sending a reply, so topic
         #   progress uses the triggering event message as the reply target
         # - Other platforms should use explicit source.thread_id only
+        #
+        # Slack honours platforms.slack.extra.reply_in_thread=false: if the
+        # user has opted out of threaded replies, don't synthesise a thread
+        # for progress messages either — the very first progress message
+        # would otherwise create a thread that all subsequent replies
+        # (including the final answer) would inherit (#18859).
+        _progress_reply_in_thread = True
+        if source.platform == Platform.SLACK:
+            _slack_adapter_for_progress = self._adapter_for_source(source)
+            if _slack_adapter_for_progress is not None:
+                try:
+                    _progress_reply_in_thread = bool(
+                        _slack_adapter_for_progress.config.extra.get(
+                            "reply_in_thread", True
+                        )
+                    )
+                except Exception:
+                    _progress_reply_in_thread = True
         _progress_thread_id = _resolve_progress_thread_id(
             source.platform, source.thread_id, event_message_id,
+            reply_in_thread=_progress_reply_in_thread,
         )
         _progress_metadata = (
             self._thread_metadata_for_source(source, event_message_id)
             if _progress_thread_id == source.thread_id
-            else {"thread_id": _progress_thread_id}
+            else self._thread_metadata_for_target(
+                source.platform,
+                source.chat_id,
+                _progress_thread_id,
+                chat_type=getattr(source, "chat_type", None),
+                reply_to_message_id=event_message_id,
+            )
         ) if _progress_thread_id else None
         _progress_metadata = _non_conversational_metadata(_progress_metadata, platform=source.platform)
         _progress_reply_to = (
@@ -20220,8 +20270,10 @@ async def _send_progress_text(text: str):
             async def _roll_progress_overflow_if_needed() -> bool:
                 """Start fresh editable progress bubbles before a bubble exceeds limit.
 
-                Returns True when it delivered/split the current buffer and the
-                caller should skip the normal send/edit path for this tick.
+                Returns True when it delivered/split the current buffer, or when
+                a transient edit failure left the buffer and message identity
+                intact for a later retry.  In either case the caller should skip
+                the normal send/edit path for this tick.
                 """
                 nonlocal progress_msg_id, progress_lines, can_edit
                 if not progress_lines or not can_edit:
@@ -20234,6 +20286,12 @@ async def _roll_progress_overflow_if_needed() -> bool:
                 if progress_msg_id is not None:
                     result = await _edit_progress_message(progress_msg_id, first_text)
                     if not result.success:
+                        if getattr(result, "retryable", False):
+                            logger.debug(
+                                "[%s] Transient overflow edit failure — keeping can_edit=True",
+                                adapter.name,
+                            )
+                            return True
                         can_edit = False
                         # Fall back to the existing non-edit behavior below.
                         return False
@@ -20503,7 +20561,17 @@ def _event_callback_sync(event_type: str, context: dict) -> None:
                 "reply_to_message_id": event_message_id,
             }
         else:
-            _status_thread_metadata = self._thread_metadata_for_source(source, event_message_id) if _progress_thread_id else None
+            _status_thread_metadata = (
+                self._thread_metadata_for_source(source, event_message_id)
+                if _progress_thread_id == source.thread_id
+                else self._thread_metadata_for_target(
+                    source.platform,
+                    source.chat_id,
+                    _progress_thread_id,
+                    chat_type=getattr(source, "chat_type", None),
+                    reply_to_message_id=event_message_id,
+                )
+            ) if _progress_thread_id else None
 
         def _status_callback_sync(event_type: str, message: str) -> None:
             if not _status_adapter or not _run_still_current():

plugins/platforms/slack/adapter.py

@@ -788,6 +788,14 @@ def __init__(self, config: PlatformConfig):
         # cache bridges lifecycle and message delivery ordering.
         self._agent_view_contexts: Dict[Tuple[str, str], Dict[str, str]] = {}
         self._AGENT_VIEW_CONTEXTS_MAX = 5000
+        # Status-bubble dedup (issue #30045, extended to Slack): remember the
+        # message ts of the last status bubble per (channel, thread, status
+        # key) so repeated progress callbacks (compression retries, fallback
+        # switches, ...) edit ONE message in place instead of appending a new
+        # bubble per event — long retry loops used to spam threads with
+        # dozens of out-of-order status messages.
+        self._status_message_ids: Dict[Tuple[str, str, str], str] = {}
+        self._STATUS_MESSAGE_IDS_MAX = 2000
         # Cache for _fetch_thread_context results: cache_key → _ThreadContextCache
         self._thread_context_cache: Dict[str, _ThreadContextCache] = {}
         self._THREAD_CACHE_TTL = 60.0
@@ -2043,6 +2051,26 @@ async def _ensure_dm_conversation(
             )
         return chat_id
 
+    async def _clear_thread_status_quietly(
+        self, chat_id: str, metadata: Optional[Dict[str, Any]] = None
+    ) -> None:
+        """Best-effort assistant-status clear for send() paths that bypass
+        the normal post-delivery clear.
+
+        Issue #24117: the assistant thread can stay stuck "is thinking..."
+        when a turn ends through a path that never reaches the regular
+        ``if thread_ts: stop_typing`` clear — an empty final response, a
+        slash-command ephemeral reply, or an exception raised before
+        ``thread_ts`` was resolved. ``stop_typing`` is already idempotent
+        (clearing an unset status is a no-op on Slack's side), so this just
+        guarantees it runs without letting a cleanup error mask the caller's
+        SendResult.
+        """
+        try:
+            await self.stop_typing(chat_id, metadata=metadata)
+        except Exception as e:  # pragma: no cover - defensive cleanup
+            logger.debug("[Slack] status cleanup failed: %s", e)
+
     async def send(
         self,
         chat_id: str,
@@ -2071,6 +2099,11 @@ async def send(
                     content,
                 )
                 if ephemeral_result.success:
+                    # Ephemeral replies do not count as thread replies, so
+                    # Slack never auto-clears the Assistant status for them.
+                    # Clear it explicitly or a command run inside an
+                    # assistant thread leaves "is thinking..." forever.
+                    await self._clear_thread_status_quietly(chat_id, metadata)
                     return ephemeral_result
                 # response_url delivery failed (#19688): fall back to
                 # chat.postEphemeral — an independent API path that keeps
@@ -2089,6 +2122,7 @@ async def send(
                     content,
                 )
                 if fallback_result.success:
+                    await self._clear_thread_status_quietly(chat_id, metadata)
                     return fallback_result
                 # Both ephemeral paths failed — surface the failure instead
                 # of leaking the reply publicly. The user still has the
@@ -2108,6 +2142,11 @@ async def send(
             # Guard against empty/whitespace-only messages — Slack API
             # returns ``no_text`` for chat.postMessage with blank text.
             if not formatted or not formatted.strip():
+                # This is still the end of a delivery attempt: if the turn
+                # produced no visible text (e.g. "(empty)" final responses
+                # are filtered upstream), the assistant thread status must
+                # not stay stuck on "is thinking..." (#24117).
+                await self._clear_thread_status_quietly(chat_id, metadata)
                 return SendResult(success=True)
 
             # Split long messages, preserving code block boundaries
@@ -2180,8 +2219,12 @@ async def send(
             )
 
         except Exception as e:  # pragma: no cover - defensive logging
-            if thread_ts:
-                await self.stop_typing(chat_id, metadata=metadata)
+            # Clear the assistant status even when the failure happened
+            # BEFORE thread_ts was resolved (formatting, slash-context, DM
+            # resolution): stop_typing falls back to metadata / the uniquely
+            # tracked status for this channel, so a failed turn cannot leave
+            # "is thinking..." visible (#24117).
+            await self._clear_thread_status_quietly(chat_id, metadata)
             logger.error("[Slack] Send error: %s", e, exc_info=True)
             _retryable = self._is_retryable_upload_error(e)
             _retry_after = None
@@ -2239,6 +2282,49 @@ async def send_private_notice(
             logger.error("[Slack] Ephemeral send error: %s", e, exc_info=True)
             return SendResult(success=False, error=str(e))
 
+    async def send_or_update_status(
+        self,
+        chat_id: str,
+        status_key: str,
+        content: str,
+        *,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Send a status message, or edit the previous one with the same key.
+
+        Issue #30045 (Telegram) extended to Slack: progress/status callbacks
+        (context-pressure, compression retries, model fallback, lifecycle)
+        used to append a fresh bubble on every call, spamming threads during
+        long retry loops. The first call posts and the message ts is
+        remembered; subsequent calls with the same (channel, thread,
+        status_key) edit that message in place via ``chat.update``. If the
+        edit fails (message deleted, too old, ...) the cached ts is dropped
+        and a fresh message is sent.
+        """
+        thread_ts = self._resolve_thread_ts(None, metadata) or ""
+        key = (str(chat_id), str(thread_ts), str(status_key))
+        cached_id = self._status_message_ids.get(key)
+        if cached_id is not None:
+            result = await self.edit_message(
+                chat_id, cached_id, content, finalize=False, metadata=metadata,
+            )
+            if result.success:
+                if result.message_id:
+                    self._status_message_ids[key] = str(result.message_id)
+                return result
+            # Edit failed — clear the cached ts and fall through to a fresh send.
+            self._status_message_ids.pop(key, None)
+        result = await self.send(chat_id, content, metadata=metadata)
+        if result.success and result.message_id:
+            if len(self._status_message_ids) >= self._STATUS_MESSAGE_IDS_MAX:
+                # Simple FIFO trim: drop the oldest half to bound memory.
+                for stale in list(self._status_message_ids)[
+                    : self._STATUS_MESSAGE_IDS_MAX // 2
+                ]:
+                    self._status_message_ids.pop(stale, None)
+            self._status_message_ids[key] = str(result.message_id)
+        return result
+
     async def edit_message(
         self,
         chat_id: str,
@@ -2293,11 +2379,48 @@ async def edit_message(
                 else:
                     raise
             if finalize:
-                await self.stop_typing(chat_id, metadata=metadata)
+                await self._clear_thread_status_quietly(chat_id, metadata)
             return SendResult(success=True, message_id=message_id)
         except Exception as e:  # pragma: no cover - defensive logging
             if finalize:
-                await self.stop_typing(chat_id, metadata=metadata)
+                await self._clear_thread_status_quietly(chat_id, metadata)
+            aiohttp_module = globals().get("aiohttp")
+            connection_error_type = getattr(
+                aiohttp_module, "ClientConnectionError", None
+            )
+            permanent_tls_error_types = tuple(
+                error_type
+                for error_type in (
+                    getattr(aiohttp_module, "ClientSSLError", None),
+                    getattr(aiohttp_module, "ServerFingerprintMismatch", None),
+                )
+                if isinstance(error_type, type)
+            )
+            is_permanent_tls_error = bool(permanent_tls_error_types) and isinstance(
+                e, permanent_tls_error_types
+            )
+            is_transient_transport_error = isinstance(e, TimeoutError) or (
+                isinstance(connection_error_type, type)
+                and isinstance(e, connection_error_type)
+                and not is_permanent_tls_error
+            )
+            if is_transient_transport_error:
+                # chat.update is idempotent: keep this message ID after a
+                # transport failure so a later edit can catch up. Treating the
+                # failure as permanent makes every later tool update a new post.
+                logger.error(
+                    "[Slack] transient chat.update failure on message %s in channel %s: %s",
+                    message_id,
+                    chat_id,
+                    e,
+                    exc_info=True,
+                )
+                return SendResult(
+                    success=False,
+                    error=str(e),
+                    retryable=True,
+                    error_kind="transient",
+                )
             logger.error(
                 "[Slack] Failed to edit message %s in channel %s: %s",
                 message_id,
@@ -2307,6 +2430,34 @@ async def edit_message(
             )
             return SendResult(success=False, error=str(e))
 
+    async def delete_message(self, chat_id: str, message_id: str) -> bool:
+        """Delete a Slack message previously sent by this bot.
+
+        Used by gateway progress cleanup so temporary "Working"/tool-progress
+        bubbles do not remain after a successful final response.
+        """
+        if not self._app:
+            return False
+        try:
+            response = await self._get_client(chat_id).chat_delete(channel=chat_id, ts=message_id)
+            if hasattr(response, "get") and response.get("ok") is False:
+                logger.debug(
+                    "[Slack] chat.delete returned ok=false for message %s in channel %s: %s",
+                    message_id,
+                    chat_id,
+                    response.get("error", "unknown"),
+                )
+                return False
+            return True
+        except Exception as e:  # pragma: no cover - best-effort cleanup
+            logger.debug(
+                "[Slack] Failed to delete message %s in channel %s: %s",
+                message_id,
+                chat_id,
+                e,
+            )
+            return False
+
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         """Show a typing/status indicator using assistant.threads.setStatus.
 
@@ -2320,7 +2471,15 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
 
         thread_ts = None
         if metadata:
-            thread_ts = metadata.get("thread_id") or metadata.get("thread_ts")
+            # Reuse the same synthetic-thread guard as message sending. When
+            # reply_in_thread=false, top-level channel events carry their own
+            # message ts as metadata.thread_id for session keying. Calling
+            # assistant_threads_setStatus on that ts activates a Slack assistant
+            # thread before the actual response is sent.
+            thread_ts = self._resolve_thread_ts(
+                reply_to=metadata.get("message_id"),
+                metadata=metadata,
+            )
 
         if not thread_ts:
             return  # Can only set status in a thread context