fix(slack): clear stuck assistant status on /stop and via explicit metadata

Salvaged from #32340 by @LeonSGP43, adapted to the workspace-scoped
status tracking that landed in #63709:

- /stop with no running agent now best-effort clears the platform
  status indicator, so a phantom 'is thinking...' left by a gateway
  restart or a turn that died without a final send can always be
  dismissed (#32295).
- SlackAdapter.stop_typing clears an untracked thread when the caller
  names it explicitly in metadata — clearing an unset status is a
  harmless no-op on Slack's side. The fallback is skipped when multiple
  Slack Connect workspaces track the same channel+thread and no team_id
  is given, preserving #63709's cross-workspace safety guarantee.

gateway/slash_commands.py

@@ -1112,6 +1112,26 @@ class GatewaySlashCommandsMixin:
             )
             return EphemeralReply(t("gateway.stop.stopped"))
 
+        # No running agent anywhere for this scope. A platform status
+        # indicator can still be stuck — e.g. Slack's persistent
+        # assistant.threads.setStatus survives a gateway restart or a turn
+        # that died without a final send (#32295). Best-effort clear so
+        # /stop always dismisses a phantom "is thinking...".
+        adapter = getattr(self, "adapters", {}).get(source.platform)
+        if adapter and hasattr(adapter, "_stop_typing_with_metadata"):
+            try:
+                await adapter._stop_typing_with_metadata(
+                    source.chat_id,
+                    self._thread_metadata_for_source(
+                        source, self._reply_anchor_for_event(event)
+                    ),
+                )
+            except Exception:
+                logger.debug(
+                    "Failed to clear typing on /stop with no active agent",
+                    exc_info=True,
+                )
+
         return t("gateway.stop.no_active")
 
     async def _handle_platform_command(self, event: MessageEvent) -> str:

plugins/platforms/slack/adapter.py

@@ -1609,6 +1609,7 @@ class SlackAdapter(BasePlatformAdapter):
             )
         requested_team_id = self._metadata_team_id(metadata)
         active = None
+        ambiguous_tracked = False
         if requested_thread_ts:
             if requested_team_id:
                 active_key = self._workspace_thread_key(
@@ -1628,6 +1629,7 @@ class SlackAdapter(BasePlatformAdapter):
                 ]
                 if len(matching_keys) == 1:
                     active = self._active_status_threads.pop(matching_keys[0], None)
+                ambiguous_tracked = len(matching_keys) > 1
         else:
             # Metadata-free cleanup is safe only if exactly one status exists
             # for this channel; otherwise it may clear another Slack Connect
@@ -1648,6 +1650,18 @@ class SlackAdapter(BasePlatformAdapter):
             team_id = active.get("team_id", "")
         if metadata:
             team_id = self._metadata_team_id(metadata) or team_id
+        if not thread_ts and requested_thread_ts and not ambiguous_tracked:
+            # No tracked entry (gateway restart, eviction, or a status set
+            # before this process started) but the caller identified the exact
+            # thread to clear. Issue the clear anyway so a stuck "is
+            # thinking..." can always be dismissed — clearing an unset status
+            # is a harmless no-op on Slack's side. Skipped when MULTIPLE
+            # workspaces track this channel+thread (ambiguous_tracked): a
+            # team-less clear there could hit the wrong Slack Connect
+            # workspace. Client routing uses the caller's team when given,
+            # else the channel→team fallback.
+            thread_ts = requested_thread_ts
+            team_id = requested_team_id or team_id
         if not thread_ts:
             return
         try:

tests/gateway/test_slack.py

@@ -2107,6 +2107,45 @@ class TestSendTyping:
 
         adapter._app.client.assistant_threads_setStatus.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_stop_typing_clears_untracked_thread_from_metadata(self, adapter):
+        """Explicit thread metadata clears a status the map no longer tracks.
+
+        A gateway restart (or cache eviction) wipes _active_status_threads
+        while Slack's persistent assistant status stays visible. A caller
+        that names the exact thread must still be able to dismiss it (#32295).
+        """
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        assert adapter._active_status_threads == {}
+
+        await adapter.stop_typing("C123", metadata={"thread_id": "stuck_ts"})
+
+        adapter._app.client.assistant_threads_setStatus.assert_called_once_with(
+            channel_id="C123",
+            thread_ts="stuck_ts",
+            status="",
+        )
+
+    @pytest.mark.asyncio
+    async def test_stop_typing_untracked_fallback_respects_ambiguous_workspaces(
+        self, adapter
+    ):
+        """Team-less clear must NOT fire when multiple workspaces track the thread."""
+        team_one, team_two = AsyncMock(), AsyncMock()
+        adapter._team_clients.update({"T_ONE": team_one, "T_TWO": team_two})
+        for team_id in ("T_ONE", "T_TWO"):
+            await adapter.send_typing(
+                "D_SHARED",
+                metadata={"thread_id": "171.000", "slack_team_id": team_id},
+            )
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+
+        await adapter.stop_typing("D_SHARED", metadata={"thread_id": "171.000"})
+
+        adapter._app.client.assistant_threads_setStatus.assert_not_called()
+        assert ("T_ONE", "D_SHARED", "171.000") in adapter._active_status_threads
+        assert ("T_TWO", "D_SHARED", "171.000") in adapter._active_status_threads
+
     @pytest.mark.asyncio
     async def test_stop_typing_handles_api_error_gracefully(self, adapter):
         adapter._active_status_threads[("", "C123", "parent_ts")] = {

tests/gateway/test_stop_thread_sibling.py

@@ -156,3 +156,67 @@ async def test_stop_does_not_interrupt_sibling_when_unauthorized(monkeypatch):
 
     assert interrupted == []
     assert "no active" in str(getattr(result, "text", result)).lower()
+
+
+# ---------------------------------------------------------------------------
+# /stop with no active agent still clears a stuck platform status (#32295)
+# ---------------------------------------------------------------------------
+
+
+class _FakeStatusAdapter:
+    def __init__(self):
+        self.cleared = []
+
+    async def _stop_typing_with_metadata(self, chat_id, metadata=None):
+        self.cleared.append((chat_id, metadata))
+
+
+@pytest.mark.asyncio
+async def test_stop_no_active_agent_clears_stuck_status():
+    runner = object.__new__(GatewayRunner)
+    runner._running_agents = {}
+    key = _per_user_key("userA")
+    runner.session_store = _FakeStore(key)
+    runner._is_user_authorized = lambda source: True
+
+    adapter = _FakeStatusAdapter()
+    runner.adapters = {Platform.DISCORD: adapter}
+    runner._thread_metadata_for_source = (
+        lambda source, reply_to_message_id=None: {"thread_id": source.thread_id}
+    )
+    runner._reply_anchor_for_event = lambda event: None
+
+    event = MessageEvent(
+        text="/stop", message_type=MessageType.TEXT, source=_thread_source("userA")
+    )
+    result = await runner._handle_stop_command(event)
+
+    assert "no active" in str(getattr(result, "text", result)).lower()
+    assert adapter.cleared == [("chan1", {"thread_id": "thr1"})]
+
+
+@pytest.mark.asyncio
+async def test_stop_no_active_agent_survives_status_clear_failure():
+    """A failing adapter clear must not break the /stop reply."""
+    runner = object.__new__(GatewayRunner)
+    runner._running_agents = {}
+    key = _per_user_key("userA")
+    runner.session_store = _FakeStore(key)
+    runner._is_user_authorized = lambda source: True
+
+    class _BoomAdapter:
+        async def _stop_typing_with_metadata(self, chat_id, metadata=None):
+            raise RuntimeError("boom")
+
+    runner.adapters = {Platform.DISCORD: _BoomAdapter()}
+    runner._thread_metadata_for_source = (
+        lambda source, reply_to_message_id=None: None
+    )
+    runner._reply_anchor_for_event = lambda event: None
+
+    event = MessageEvent(
+        text="/stop", message_type=MessageType.TEXT, source=_thread_source("userA")
+    )
+    result = await runner._handle_stop_command(event)
+
+    assert "no active" in str(getattr(result, "text", result)).lower()