[withdrawn]

gateway/platforms/base.py

@@ -3199,6 +3199,10 @@ def __init__(self, config: PlatformConfig, platform: Platform):
         # Chats where typing indicator is paused (e.g. during approval waits).
         # _keep_typing skips send_typing when the chat_id is in this set.
         self._typing_paused: set = set()
+        # Chats whose turn has finished. Discord send_typing starts a persistent
+        # loop; a late progress-path send_typing after stop_typing can recreate
+        # it forever (see #85427). Closed until the next _keep_typing session.
+        self._typing_closed: set = set()
         # Dynamic working-state status text per chat (chat_id -> phrase).
         # Set by the gateway on tool starts ("is running pytest…") and read
         # by adapters whose typing indicator renders text (Slack's
@@ -5320,6 +5324,7 @@ async def _keep_typing(
         # gated on network health.  Must stay below ``interval`` so a slow
         # call gets abandoned before the next scheduled tick.
         _send_typing_timeout = max(0.25, min(1.5, interval - 0.25))
+        self._typing_closed.discard(chat_id)
         try:
             while True:
                 if stop_event is not None and stop_event.is_set():
@@ -5385,6 +5390,7 @@ async def _stop_typing_refresh(
         stop_attempts: int = 2,
     ) -> None:
         """Stop the refresh task and platform typing state as one operation."""
+        self._typing_closed.add(chat_id)
         self._typing_paused.add(chat_id)
         try:
             if typing_task is not None and not typing_task.done():

plugins/platforms/discord/adapter.py

@@ -5600,6 +5600,11 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
         """
         if not self._client:
             return
+        # Turn already finished — refuse to recreate the persistent loop.
+        # gateway/run.py progress restore and _keep_typing finally-unpause
+        # both call send_typing after stop_typing (#85427).
+        if chat_id in getattr(self, "_typing_closed", ()):
+            return
         # Don't start a duplicate loop
         if chat_id in self._typing_tasks:
             return
@@ -5635,18 +5640,22 @@ async def _typing_loop() -> None:
             except asyncio.CancelledError:
                 pass
             finally:
-                self._typing_tasks.pop(chat_id, None)
+                # Only evict *this* loop. An unconditional pop orphans a newer
+                # loop registered by a concurrent send_typing (#85425).
+                if self._typing_tasks.get(chat_id) is asyncio.current_task():
+                    self._typing_tasks.pop(chat_id, None)
 
         self._typing_tasks[chat_id] = asyncio.create_task(_typing_loop())
 
     async def stop_typing(self, chat_id: str) -> None:
         """Stop the persistent typing indicator for a channel."""
+        getattr(self, "_typing_closed", set()).add(chat_id)
         task = self._typing_tasks.pop(chat_id, None)
         if task:
             task.cancel()
             try:
-                await task
-            except (asyncio.CancelledError, Exception):
+                await asyncio.wait_for(asyncio.shield(task), timeout=0.5)
+            except (asyncio.CancelledError, asyncio.TimeoutError, Exception):
                 pass
 
     async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:

tests/gateway/test_discord_typing_stuck.py

@@ -0,0 +1,91 @@
+"""Discord typing-loop races: closed-gate, finally-guard, bounded stop.
+
+Tracks #85427 (class) / #85425 (orphan pop) and the late progress-path
+send_typing recreate we hit on Hermes 0.20.x after ✅.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _ensure_discord_mock():
+    if "discord" in sys.modules and hasattr(sys.modules["discord"], "__file__"):
+        return
+    discord_mod = MagicMock()
+    discord_mod.Intents.default.return_value = MagicMock()
+    discord_mod.Client = MagicMock
+    discord_mod.http = SimpleNamespace(Route=MagicMock)
+    ext_mod = MagicMock()
+    commands_mod = MagicMock()
+    commands_mod.Bot = MagicMock
+    ext_mod.commands = commands_mod
+    sys.modules.setdefault("discord", discord_mod)
+    sys.modules.setdefault("discord.ext", ext_mod)
+    sys.modules.setdefault("discord.ext.commands", commands_mod)
+
+
+_ensure_discord_mock()
+
+from plugins.platforms.discord.adapter import DiscordAdapter  # noqa: E402
+
+
+def _adapter() -> DiscordAdapter:
+    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="***"))
+    adapter._client = MagicMock()
+    adapter._client.http = MagicMock()
+    adapter._client.http.request = AsyncMock()
+    adapter._typing_tasks = {}
+    return adapter
+
+
+@pytest.mark.asyncio
+async def test_send_typing_after_stop_does_not_recreate_loop():
+    """Late progress send_typing must not start a new loop after stop_typing."""
+    adapter = _adapter()
+    await adapter.send_typing("chan")
+    assert "chan" in adapter._typing_tasks
+    await adapter.stop_typing("chan")
+    assert "chan" not in adapter._typing_tasks
+    assert "chan" in adapter._typing_closed
+
+    await adapter.send_typing("chan")
+    assert "chan" not in adapter._typing_tasks
+
+
+@pytest.mark.asyncio
+async def test_keep_typing_reopens_closed_gate():
+    adapter = _adapter()
+    await adapter.stop_typing("chan")
+    assert "chan" in adapter._typing_closed
+    adapter._typing_closed.discard("chan")
+    await adapter.send_typing("chan")
+    assert "chan" in adapter._typing_tasks
+    await adapter.stop_typing("chan")
+
+
+@pytest.mark.asyncio
+async def test_typing_finally_does_not_orphan_newer_loop():
+    """Stale loop finally must not pop a replacement registered for the same chat."""
+    adapter = _adapter()
+    await adapter.send_typing("chan")
+    loop_a = adapter._typing_tasks["chan"]
+    adapter._typing_tasks.pop("chan")
+    adapter._typing_closed.discard("chan")
+    await adapter.send_typing("chan")
+    loop_b = adapter._typing_tasks["chan"]
+    assert loop_b is not loop_a
+    loop_a.cancel()
+    try:
+        await asyncio.wait_for(asyncio.shield(loop_a), timeout=0.5)
+    except (asyncio.CancelledError, asyncio.TimeoutError):
+        pass
+    assert adapter._typing_tasks.get("chan") is loop_b
+    await adapter.stop_typing("chan")