Typing indicators linger or get stuck after replies

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #3003 fix(discord): stop phantom typing indicator after agent turn completes

  merged to main

  files: gateway/platforms/base.py

  Show the diff

  • gateway/platforms/base.py

    @@ -819,6 +819,16 @@ async def _keep_typing(self, chat_id: str, interval: float = 2.0, metadata=None)
                     await asyncio.sleep(interval)
             except asyncio.CancelledError:
                 pass  # Normal cancellation when handler completes
    +        finally:
    +            # Ensure the underlying platform typing loop is stopped.
    +            # _keep_typing may have called send_typing() after an outer
    +            # stop_typing() cleared the task dict, recreating the loop.
    +            # Cancelling _keep_typing alone won't clean that up.
    +            if hasattr(self, "stop_typing"):
    +                try:
    +                    await self.stop_typing(chat_id)
    +                except Exception:
    +                    pass
         
         async def handle_message(self, event: MessageEvent) -> None:
             """
    @@ -1130,6 +1140,13 @@ async def _process_message_background(self, event: MessageEvent, session_key: st
                     await typing_task
                 except asyncio.CancelledError:
                     pass
    +            # Also cancel any platform-level persistent typing tasks (e.g. Discord)
    +            # that may have been recreated by _keep_typing after the last stop_typing()
    +            try:
    +                if hasattr(self, "stop_typing"):
    +                    await self.stop_typing(event.source.chat_id)
    +            except Exception:
    +                pass
                 # Clean up session tracking
                 if session_key in self._active_sessions:
                     del self._active_sessions[session_key]

• #10239 fix(gateway): fix matrix lingering typing indicator

  merged to main

  files: gateway/platforms/matrix.py, tests/gateway/test_matrix.py

  Show the diff

  • gateway/platforms/matrix.py

    @@ -729,6 +729,14 @@ async def send_typing(
                 except Exception:
                     pass
     
    +    async def stop_typing(self, chat_id: str) -> None:
    +        """Stop the Matrix typing indicator."""
    +        if self._client:
    +            try:
    +                await self._client.set_typing(RoomID(chat_id), timeout=0)
    +            except Exception:
    +                pass
    +
         async def edit_message(
             self, chat_id: str, message_id: str, content: str
         ) -> SendResult:

  • tests/gateway/test_matrix.py

    @@ -335,6 +335,29 @@ def _make_adapter():
         return adapter
     
     
    +# ---------------------------------------------------------------------------
    +# Typing indicator
    +# ---------------------------------------------------------------------------
    +
    +class TestMatrixTypingIndicator:
    +    def setup_method(self):
    +        self.adapter = _make_adapter()
    +        self.adapter._client = MagicMock()
    +        self.adapter._client.set_typing = AsyncMock()
    +
    +    @pytest.mark.asyncio
    +    async def test_stop_typing_clears_matrix_typing_state(self):
    +        """stop_typing() should send typing=false instead of waiting for timeout expiry."""
    +        from gateway.platforms.matrix import RoomID
    +
    +        await self.adapter.stop_typing("!room:example.org")
    +
    +        self.adapter._client.set_typing.assert_awaited_once_with(
    +            RoomID("!room:example.org"),
    +            timeout=0,
    +        )
    +
    +
     # ---------------------------------------------------------------------------
     # mxc:// URL conversion
     # ---------------------------------------------------------------------------
    @@ -1831,4 +1854,3 @@ async def test_set_presence_no_client(self):
             assert result is False
     
     
    -

• #44836 fix: stop Discord typing after replies

  merged to main

  files: gateway/platforms/base.py, scripts/release.py, tests/gateway/test_keep_typing_timeout.py, tests/gateway/test_run_progress_topics.py

  Show the diff

  • gateway/platforms/base.py

    @@ -3163,6 +3163,38 @@ async def _keep_typing(
                         pass
                 self._typing_paused.discard(chat_id)
     
    +    async def _stop_typing_refresh(
    +        self,
    +        chat_id: str,
    +        typing_task: asyncio.Task | None = None,
    +        *,
    +        timeout: float = 0.5,
    +        stop_attempts: int = 2,
    +    ) -> None:
    +        """Stop the refresh task and platform typing state as one operation."""
    +        self._typing_paused.add(chat_id)
    +        try:
    +            if typing_task is not None and not typing_task.done():
    +                typing_task.cancel()
    +                try:
    +                    await asyncio.wait_for(asyncio.shield(typing_task), timeout=timeout)
    +                except (asyncio.CancelledError, asyncio.TimeoutError):
    +                    # The task is cancelled; don't let a slow adapter-specific
    +                    # cleanup block response delivery or shutdown.
    +                    pass
    +            if not hasattr(self, "stop_typing"):
    +                return
    +            attempts = max(1, stop_attempts)
    +            for attempt in range(attempts):
    +                try:
    +                    await self.stop_typing(chat_id)
    +                except Exception:
    +                    pass
    +                if attempt < attempts - 1:
    +                    await asyncio.sleep(0)
    +        finally:
    +            self._typing_paused.discard(chat_id)
    +
         def pause_typing_for_chat(self, chat_id: str) -> None:
             """Pause typing indicator for a chat (e.g. during approval waits).
     
    @@ -4092,14 +4124,10 @@ def _record_delivery(result):
             )
     
             async def _stop_typing_task() -> None:
    -            typing_task.cancel()
    -            try:
    -                await asyncio.wait_for(asyncio.shield(typing_task), timeout=0.5)
    -            except (asyncio.CancelledError, asyncio.TimeoutError):
    -                # Cancellation cleanup must not block adapter shutdown.  The
    -                # typing task is already cancelled; if the parent task is also
    -                # cancelling, let this message-processing task unwind now.
    -                pass
    +            await self._stop_typing_refresh(
    +                event.source.chat_id,
    +                typing_task,
    +            )
             
             try:
                 await self._run_processing_hook("on_processing_start", event)
    @@ -4486,11 +4514,6 @@ async def _stop_typing_task() -> None:
                 # callbacks may perform platform I/O; a stuck callback must not
                 # leave the typing refresh task running indefinitely.
                 await _stop_typing_task()
    -            try:
    -                if hasattr(self, "stop_typing"):
    -                    await self.stop_typing(event.source.chat_id)
    -            except Exception:
    -                pass
                 # Fire any one-shot post-delivery callback registered for this
                 # session (e.g. deferred background-review notifications).
                 #
    @@ -4524,13 +4547,14 @@ async def _stop_typing_task() -> None:
                             )
                     except (asyncio.TimeoutError, Exception):
                         pass
    -            # Also cancel any platform-level persistent typing tasks (e.g. Discord)
    -            # that may have been recreated by _keep_typing after the last stop_typing()
    -            try:
    -                if hasattr(self, "stop_typing"):
    -                    await self.stop_typing(event.source.chat_id)
    -            except Exception:
    -                pass
    +            # Some adapters keep platform-level typing tasks.  If callback
    +            # work or a late refresh recreated one, make one final bounded stop
    +            # before releasing the session guard.
    +            await self._stop_typing_refresh(
    +                event.source.chat_id,
    +                None,
    +                stop_attempts=1,
    +            )
                 # Final drain/release boundary: force-flush any timer that missed
                 # the in-band drain before deciding whether the guard can clear.
                 await self._flush_text_debounce_now(session_key)

  • scripts/release.py

    @@ -75,6 +75,7 @@
         "[redacted:email]": "harjothkhara",
         "[redacted:email]": "HeLLGURD",
         "[redacted:email]": "synapsesx",
    +    "[redacted:email]": "itsflownium",
         "[redacted:email]": "dirtyren",
         "[redacted:email]": "kdunn926",
         "[redacted:email]": "mvanhorn",

  • tests/gateway/test_keep_typing_timeout.py

    @@ -198,3 +198,39 @@ async def recording_send_typing(chat_id, metadata=None):
             assert calls == [], (
                 f"send_typing was called on a paused chat: {calls}"
             )
    +
    +    @pytest.mark.asyncio
    +    async def test_stop_typing_refresh_blocks_late_cancel_tick(self, monkeypatch):
    +        """Final cleanup must not let a cancelled refresh loop send typing again."""
    +        adapter = _StubAdapter()
    +        late_sends = []
    +        stop_calls = []
    +
    +        async def send_typing(chat_id, metadata=None):
    +            late_sends.append(chat_id)
    +
    +        async def stop_typing(chat_id):
    +            stop_calls.append((chat_id, chat_id in adapter._typing_paused))
    +
    +        monkeypatch.setattr(adapter, "send_typing", send_typing)
    +        monkeypatch.setattr(adapter, "stop_typing", stop_typing)
    +
    +        async def late_refresh_after_cancel():
    +            try:
    +                await asyncio.sleep(10)
    +            except asyncio.CancelledError:
    +                if "discord-chat" not in adapter._typing_paused:
    +                    await adapter.send_typing("discord-chat")
    +                raise
    +
    +        task = asyncio.create_task(late_refresh_after_cancel())
    +        await asyncio.sleep(0)
    +
    +        await adapter._stop_typing_refresh("discord-chat", task, timeout=1.0)
    +
    +        assert late_sends == []
    +        assert stop_calls == [
    +            ("discord-chat", True),
    +            ("discord-chat", True),
    +        ]
    +        assert "discord-chat" not in adapter._typing_paused

  • tests/gateway/test_run_progress_topics.py

    @@ -1121,7 +1121,15 @@ async def _stop_typing(chat_id):
         )
     
         assert [call["content"] for call in adapter.sent] == ["done"]
    -    assert events[:2] == ["typing-stopped", "callback-start"]
    +    # Invariant: typing must stop before the (hung) post-delivery callback
    +    # starts.  Don't pin the exact stop_typing call count — the shared
    +    # cleanup path may make more than one bounded stop attempt.
    +    assert "typing-stopped" in events
    +    assert "callback-start" in events
    +    assert events.index("typing-stopped") < events.index("callback-start")
    +    assert events[: events.index("callback-start")] == (
    +        ["typing-stopped"] * events.index("callback-start")
    +    )
         assert any(call["metadata"] == {"stopped": True} for call in adapter.typing)
     
     

• #50378 fix(gateway): pause Telegram typing before stream finalize (#49712)

  merged to main

  files: gateway/run.py, gateway/stream_consumer.py, tests/gateway/test_stream_consumer.py

  Show the diff

  • gateway/run.py

    @@ -14133,6 +14133,13 @@ def _run_still_current() -> bool:
                     from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
                     _adapter = self.adapters.get(source.platform)
                     if _adapter:
    +                    _pause_typing_before_finalize = None
    +                    if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"):
    +                        def _pause_typing_before_finalize(
    +                            _adapter=_adapter,
    +                            _chat_id=source.chat_id,
    +                        ) -> None:
    +                            _adapter.pause_typing_for_chat(_chat_id)
                         _adapter_supports_edit = getattr(_adapter, "SUPPORTS_MESSAGE_EDITING", True)
                         _effective_cursor = _scfg.cursor if _adapter_supports_edit else ""
                         _buffer_only = False
    @@ -14162,6 +14169,7 @@ def _run_still_current() -> bool:
                             chat_id=source.chat_id,
                             config=_consumer_cfg,
                             metadata=_thread_metadata,
    +                        on_before_finalize=_pause_typing_before_finalize,
                             initial_reply_to_id=event_message_id,
                         )
                 except Exception as _sc_err:
    @@ -15290,6 +15298,13 @@ def run_sync():
                         from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
                         _adapter = self.adapters.get(source.platform)
                         if _adapter:
    +                        _pause_typing_before_finalize = None
    +                        if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"):
    +                            def _pause_typing_before_finalize(
    +                                _adapter=_adapter,
    +                                _chat_id=source.chat_id,
    +                            ) -> None:
    +                                _adapter.pause_typing_for_chat(_chat_id)
                             # Platforms that don't support editing sent messages
                             # (e.g. QQ, WeChat) should skip streaming entirely —
                             # without edit support, the consumer sends a partial
    @@ -15334,6 +15349,7 @@ def run_sync():
                                     if progress_queue is not None
                                     else None
                                 ),
    +                            on_before_finalize=_pause_typing_before_finalize,
                                 initial_reply_to_id=event_message_id,
                             )
                             if _want_stream_deltas:

  • gateway/stream_consumer.py

    @@ -119,6 +119,7 @@ def __init__(
             config: Optional[StreamConsumerConfig] = None,
             metadata: Optional[dict] = None,
             on_new_message: Optional[callable] = None,
    +        on_before_finalize: Optional[Callable[[], Any]] = None,
             initial_reply_to_id: Optional[str] = None,
         ):
             self.adapter = adapter
    @@ -133,6 +134,10 @@ def __init__(
             # the content, not edit the old bubble above it.
             # Called with no arguments. Exceptions are swallowed.
             self._on_new_message = on_new_message
    +        # Fired once when the stream transitions into its finalization path.
    +        # Gateway callers use this to pause typing refreshes before a slow
    +        # final rich-text edit (Telegram MarkdownV2 finalize, etc.).
    +        self._on_before_finalize = on_before_finalize
             self._initial_reply_to_id = initial_reply_to_id
             self._queue: queue.Queue = queue.Queue()
             self._accumulated = ""
    @@ -196,6 +201,7 @@ def __init__(
             # first failure we permanently disable drafts for the remainder of
             # this response and route through edit-based for graceful degradation.
             self._draft_failures = 0
    +        self._before_finalize_notified = False
     
         def _metadata_for_send(
             self,
    @@ -242,6 +248,20 @@ def final_content_delivered(self) -> bool:
             the subsequent cosmetic edit (cursor removal) failed."""
             return self._final_content_delivered
     
    +    async def _notify_before_finalize(self) -> None:
    +        """Run the pre-finalize hook exactly once, swallowing hook errors."""
    +        if self._before_finalize_notified:
    +            return
    +        self._before_finalize_notified = True
    +        if self._on_before_finalize is None:
    +            return
    +        try:
    +            result = self._on_before_finalize()
    +            if inspect.isawaitable(result):
    +                await result
    +        except Exception:
    +            pass
    +
         async def _edit_message(
             self,
             *,
    @@ -620,6 +640,8 @@ async def run(self) -> None:
                         self._last_edit_time = time.monotonic()
     
                     if got_done:
    +                    if self._accumulated or self._message_id is not None or self._already_sent:
    +                        await self._notify_before_finalize()
                         # Final edit without cursor. If progressive editing failed
                         # mid-stream, send a single continuation/fallback message
                         # here instead of letting the base gateway path send the

  • tests/gateway/test_stream_consumer.py

    @@ -361,6 +361,67 @@ async def test_stream_with_media_tag(self):
             assert consumer.already_sent
     
     
    +class TestBeforeFinalizeHook:
    +    """Verify the optional pre-finalize hook fires at the right time."""
    +
    +    @pytest.mark.asyncio
    +    async def test_hook_runs_before_finalize_edit(self):
    +        """Adapters that require finalize should pause typing before the edit."""
    +        events = []
    +        adapter = MagicMock()
    +        adapter.REQUIRES_EDIT_FINALIZE = True
    +        adapter.send = AsyncMock(
    +            side_effect=lambda **_kw: (
    +                events.append("send"),
    +                SimpleNamespace(success=True, message_id="msg_1"),
    +            )[1]
    +        )
    +        adapter.edit_message = AsyncMock(
    +            side_effect=lambda **_kw: (
    +                events.append("edit"),
    +                SimpleNamespace(success=True, message_id="msg_1"),
    +            )[1]
    +        )
    +        adapter.MAX_MESSAGE_LENGTH = 4096
    +
    +        consumer = GatewayStreamConsumer(
    +            adapter,
    +            "chat_123",
    +            StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5),
    +            on_before_finalize=lambda: events.append("pause"),
    +        )
    +        consumer.on_delta("Hello")
    +        consumer.finish()
    +
    +        await consumer.run()
    +
    +        assert events == ["send", "pause", "edit"]
    +
    +    @pytest.mark.asyncio
    +    async def test_hook_runs_once_when_final_text_already_visible(self):
    +        """The hook still fires once even when no final edit is required."""
    +        events = []
    +        adapter = MagicMock()
    +        adapter.REQUIRES_EDIT_FINALIZE = False
    +        adapter.send = AsyncMock(return_value=SimpleNamespace(success=True, message_id="msg_1"))
    +        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=True, message_id="msg_1"))
    +        adapter.MAX_MESSAGE_LENGTH = 4096
    +
    +        consumer = GatewayStreamConsumer(
    +            adapter,
    +            "chat_123",
    +            StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5),
    +            on_before_finalize=lambda: events.append("pause"),
    +        )
    +        consumer.on_delta("Hello")
    +        consumer.finish()
    +
    +        await consumer.run()
    +
    +        assert events == ["pause"]
    +        adapter.edit_message.assert_not_called()
    +
    +
     # ── Segment break (tool boundary) tests ──────────────────────────────────