feat(slack): elapsed typing heartbeat — 'still working… (2m03s)' on long turns

Salvaged from PR #45702 (heartbeat half). A multi-minute turn showed a
static 'is thinking...' assistant status that reads as stuck and provokes
mid-turn 'you there?' pings. Derive the fallback status label from the
turn's elapsed time (>=30s → 'still working… (NmSSs)'), riding the
existing _keep_typing refresh — zero extra API calls.

Ported onto the current plugin adapter: the start time rides the tracked
_active_status_threads entry (workspace-scoped key), so it shares the
existing bounds/eviction and resets when stop_typing clears the status.
Explicit live-status phrases (set_status_text) and configured
typing_status_text always win; only the built-in default label changes.

The PR's other half (top-level channel follow-up coalescing) is NOT
included — dispatch semantics changed on main (busy-input active-turn
redirect, #30170 demotion) and need a fresh design pass.

Refs #45702. Co-authored-by: MrAbsaroka <mrabsaroka@gmail.com>

plugins/platforms/slack/adapter.py

@@ -845,7 +845,7 @@ class SlackAdapter(BasePlatformAdapter):
         # Entries are popped when the status clears, but statuses abandoned
         # by an error path would accumulate — bound with oldest-thread-first
         # eviction (key[2] is the thread ts).
-        self._active_status_threads: Dict[Tuple[str, str, str], Dict[str, str]] = {}
+        self._active_status_threads: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
         self._ACTIVE_STATUS_THREADS_MAX = 1000
         # Best-effort guard so automatic Slack AI thread titles are set once
         # per visible DM thread instead of on every reply.
@@ -2596,10 +2596,24 @@ class SlackAdapter(BasePlatformAdapter):
             team_id = self._channel_team.get(chat_id, "")
 
         status_key = self._workspace_thread_key(team_id, chat_id, str(thread_ts))
+        _status_started: Optional[float] = None
         if status_key:
+            # Heartbeat (#45702): preserve the first refresh's start time
+            # across _keep_typing refreshes so a long turn surfaces elapsed
+            # time ("still working… (2m03s)") instead of a static
+            # "is thinking..." that reads as stuck — which is what provokes
+            # mid-turn "you there?" pings. Stored inside the tracked status
+            # entry so it shares the existing bounds/eviction and is dropped
+            # by stop_typing with the rest of the status state.
+            _prev_entry = self._active_status_threads.get(status_key)
+            if isinstance(_prev_entry, dict):
+                _status_started = _prev_entry.get("started")
+            if not isinstance(_status_started, (int, float)):
+                _status_started = time.monotonic()
             self._active_status_threads[status_key] = {
                 "thread_ts": str(thread_ts),
                 "team_id": str(team_id) if team_id else "",
+                "started": _status_started,
             }
             if len(self._active_status_threads) > self._ACTIVE_STATUS_THREADS_MAX:
                 # Evict abandoned statuses oldest-thread-first (key[2] is the
@@ -2618,8 +2632,23 @@ class SlackAdapter(BasePlatformAdapter):
             _status = (
                 getattr(self, "_status_text", {}).get(str(chat_id))
                 or getattr(self.config, "typing_status_text", None)
-                or "is thinking..."
             )
+            if not _status:
+                # Heartbeat (#45702): once a turn has run for 30s+, replace
+                # the static default with visible elapsed progress. Only the
+                # fallback label changes — explicit live-status phrases and
+                # configured typing_status_text always win.
+                _elapsed = (
+                    int(time.monotonic() - _status_started)
+                    if _status_started is not None
+                    else 0
+                )
+                if _elapsed >= 30:
+                    _mins, _secs = divmod(_elapsed, 60)
+                    _human = f"{_mins}m{_secs:02d}s" if _mins else f"{_secs}s"
+                    _status = f"still working… ({_human})"
+                else:
+                    _status = "is thinking..."
             await self._get_client(chat_id, team_id=team_id).assistant_threads_setStatus(
                 channel_id=chat_id,
                 thread_ts=thread_ts,

tests/gateway/test_slack.py

@@ -3382,6 +3382,68 @@ class TestSendTyping:
         await adapter.send_typing("C123")
         adapter._app.client.assistant_threads_setStatus.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_elapsed_heartbeat_after_30s(self, adapter, monkeypatch):
+        """#45702: a long-running turn surfaces elapsed time instead of a
+        static 'is thinking...' that reads as stuck."""
+        import time as _time
+
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        clock = [1000.0]
+        monkeypatch.setattr(_time, "monotonic", lambda: clock[0])
+
+        await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
+        assert (
+            adapter._app.client.assistant_threads_setStatus.call_args.kwargs["status"]
+            == "is thinking..."
+        )
+
+        # 2m03s later, the refresh loop calls send_typing again.
+        clock[0] += 123
+        await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
+        assert (
+            adapter._app.client.assistant_threads_setStatus.call_args.kwargs["status"]
+            == "still working… (2m03s)"
+        )
+
+    @pytest.mark.asyncio
+    async def test_heartbeat_resets_after_stop_typing(self, adapter, monkeypatch):
+        """stop_typing ends the turn — the next turn starts a fresh clock."""
+        import time as _time
+
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        clock = [2000.0]
+        monkeypatch.setattr(_time, "monotonic", lambda: clock[0])
+
+        await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
+        clock[0] += 90
+        await adapter.stop_typing("C123", metadata={"thread_id": "parent_ts"})
+
+        clock[0] += 5
+        await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
+        assert (
+            adapter._app.client.assistant_threads_setStatus.call_args.kwargs["status"]
+            == "is thinking..."
+        )
+
+    @pytest.mark.asyncio
+    async def test_heartbeat_never_overrides_live_status_text(self, adapter, monkeypatch):
+        """Explicit live-status phrases always win over the heartbeat label."""
+        import time as _time
+
+        adapter._app.client.assistant_threads_setStatus = AsyncMock()
+        clock = [3000.0]
+        monkeypatch.setattr(_time, "monotonic", lambda: clock[0])
+
+        await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
+        clock[0] += 120
+        adapter.set_status_text("C123", "is running pytest…")
+        await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
+        assert (
+            adapter._app.client.assistant_threads_setStatus.call_args.kwargs["status"]
+            == "is running pytest…"
+        )
+
     @pytest.mark.asyncio
     async def test_handles_missing_scope_gracefully(self, adapter):
         adapter._app.client.assistant_threads_setStatus = AsyncMock(
@@ -3632,10 +3694,11 @@ class TestSendTyping:
             call(channel_id="D123", thread_ts="thread_a", status=""),
         ]
         assert ("", "D123", "thread_a") not in adapter._active_status_threads
-        assert adapter._active_status_threads[("", "D123", "thread_b")] == {
-            "thread_ts": "thread_b",
-            "team_id": "",
-        }
+        _entry_b = adapter._active_status_threads[("", "D123", "thread_b")]
+        assert _entry_b["thread_ts"] == "thread_b"
+        assert _entry_b["team_id"] == ""
+        # Heartbeat start time rides the tracked entry (#45702).
+        assert isinstance(_entry_b.get("started"), float)
 
     @pytest.mark.asyncio
     async def test_stop_typing_with_metadata_preserves_sibling_status(self, adapter):