fix(discord): prevent typing indicator sticking when _typing_loop cleanup races a concurrent send_typing

[handnewb]

Summary

Fixes a race in the Discord adapter's typing loop that leaves the "is typing…" indicator stuck permanently, even after the agent has finished and the message has been delivered.

Root cause

send_typing() registers a persistent per-chat loop that POSTs /channels/{id}/typing every 12s; stop_typing() cancels it. The loop's finally popped the registry entry unconditionally:

finally:
    self._typing_tasks.pop(chat_id, None)

The race:

1. stop_typing() pops the entry for loop A and cancels it.
2. Before A's finally unwinds, a concurrent send_typing() re-registers a fresh loop B for the same chat. This happens routinely: the tool-progress path in gateway/run.py and the base _keep_typing refresh both call send_typing() during a turn.
3. A's finally runs pop(chat_id), which removes B — not A — from the registry.
4. B is now orphaned: it keeps POSTing /typing every 12s forever, and stop_typing() can never cancel it again because it is no longer in the dict.

Result: Discord keeps showing "is typing…" indefinitely until the gateway is restarted.

This line was introduced by the adapter refactor cc8e5ec2a (Discord → bundled plugin) and is still present on main (line 5594).

Fix

Only clear the entry when it still points at this loop:

finally:
    if self._typing_tasks.get(chat_id) is asyncio.current_task():
        self._typing_tasks.pop(chat_id, None)

If the entry has been replaced by a newer loop, the old loop leaves it alone.

Test

tests/gateway/test_discord_typing_stuck.py reproduces the exact interleaving deterministically:

• register loop A → stop_typing pops A → re-register loop B → cancel A → assert B is still tracked (was None before the fix) → assert stop_typing can still cancel B cleanly.

Verified: the test fails on the buggy code and passes with the fix (2/2).

Relationship to existing issues

This is a complementary fix, not a duplicate of the open work:

• [Bug]: Discord typing indicator gets stuck permanently when _typing_loop HTTP request hangs #64874 (open, P2) — _typing_loop HTTP request hangs with no timeout. Different mechanism (needs a timeout on the HTTP call; PR fix(discord): add timeout to typing indicator HTTP request to prevent stuck loop #64910 addresses that).
• Discord: typing indicator lingers after response is sent #49290 (open, P3) — indicator lingers ~10s after the message (Discord has no early-clear API; unrelated to orphaning).
• Discord typing indicator stuck indefinitely due to race condition in _keep_typing cleanup #26854 / [Bug]: Discord typing indicator can get stuck forever during/after long replies #26728 (closed) — earlier _keep_typing/base.py races, fixed separately.

This PR closes the orphaned-re-registered-loop variant, which is not yet covered by any open PR.

[Enough1122]

AI code review — automated review for reference, author can ignore or act on any point.

fix(discord): prevent typing indicator sticking when _typing_loop cleanup races a concurrent send_typing

The identity-guarded pop is the right fix: the get() → is-check → pop() sequence contains no await, so it cannot interleave with a concurrent send_typing() re-registration, and during cancellation unwinding asyncio.current_task() still returns the loop's own task.

• plugins/platforms/discord/adapter.py (_typing_loop finally block): the same orphaning race also applies when the loop exits via the non-429 return in the inner except (not only via cancellation). The guard covers that path too, but the regression test only reproduces the cancel interleaving — a second test for the early-exit path would lock it in.
• Consider documenting the invariant next to _typing_tasks ("the dict entry for a chat always points at the live loop for that chat") so a future refactor does not reintroduce the unconditional pop.
• The normal-path test (test_stop_typing_clears_registry_in_normal_path) is a good complement — both the raced and un-raced paths are covered.

[omgbabyweb]

Adopted this commit on a downstream candidate; the identity guard works, and the two PR tests plus the existing send suite and an additional real-stop regression pass (14 tests).

One useful test improvement: exercise asyncio.create_task(adapter.stop_typing(chat_id)) instead of manually popping the registry. After yielding to the stopper, assert the slot is empty and the old task is not done, start the replacement, await the stopper, and assert the replacement remains tracked. Then call stop_typing and assert the replacement task itself is done, not just absent from the registry. Cancel/gather captured tasks in a finally so an assertion failure does not leak a typing loop.

This tests the actual stop/replacement interleaving and makes an unexpected scheduler ordering fail explicitly rather than silently bypassing the race. No production change suggested.

[omgbabyweb]

Apply-ready version of the real-stop regression suggested above. This replaces the manual registry pop, checks the old task has not finished before replacement, verifies cancellation of the replacement itself, and cleans up captured tasks on failure.

Suggested change

	@pytest.mark.asyncio
	async def test_cancelled_typing_loop_does_not_orphan_newer_loop() -> None:
	"""A cancelled loop's ``finally`` must not pop a re-registered sibling.
	Reproduces the interleaving directly: stop_typing's pop happens first,
	a new send_typing re-registers loop B, then loop A's cancellation unwinds
	through its ``finally``. With the bug, A's ``finally`` pops B and B is
	orphaned (still tracked nowhere, still POSTing /typing, unstoppable).
	"""
	adapter = _make_adapter()
	chat_id = "123456789012345678"
	# Signal once loop A has actually started POSTing, so the later cancel()
	# lands while A is inside its 12s sleep — not before the coroutine body
	# (and therefore its finally) has even begun.
	a_started = asyncio.Event()
	async def _request(_route) -> None:
	a_started.set()
	adapter._client.http.request = _request
	# Turn 1: send_typing registers loop A.
	await adapter.send_typing(chat_id)
	loop_a = adapter._typing_tasks[chat_id]
	assert loop_a is not None
	# Let loop A start and reach its first POST, then settle into sleep(12).
	await asyncio.wait_for(a_started.wait(), timeout=1.0)
	await asyncio.sleep(0)
	# stop_typing pops the entry, then cancels. We emulate the pop explicitly
	# so we can interleave a re-registration the way the real code does.
	popped = adapter._typing_tasks.pop(chat_id)
	assert popped is loop_a
	# A fresh send_typing re-registers loop B before A's finally runs.
	await adapter.send_typing(chat_id)
	loop_b = adapter._typing_tasks[chat_id]
	assert loop_b is not None and loop_b is not loop_a
	# Now A's cancellation unwinds through its finally.
	loop_a.cancel()
	try:
	await loop_a
	except asyncio.CancelledError:
	pass
	# B must still be tracked — otherwise it is orphaned and unstoppable.
	assert adapter._typing_tasks.get(chat_id) is loop_b
	# And stop_typing can still cancel B cleanly.
	await adapter.stop_typing(chat_id)
	assert adapter._typing_tasks.get(chat_id) is None
	@pytest.mark.asyncio
	async def test_cancelled_typing_loop_does_not_orphan_newer_loop() -> None:
	adapter = DiscordAdapter(PlatformConfig(enabled=True, token="***"))
	entered = asyncio.Event()
	parked = asyncio.Event()
	async def request(*args, **kwargs):
	entered.set()
	await parked.wait()
	adapter._client = SimpleNamespace(http=SimpleNamespace(request=request))
	tasks = []
	try:
	await adapter.send_typing("12345")
	old = adapter._typing_tasks["12345"]
	tasks.append(old)
	await asyncio.wait_for(entered.wait(), timeout=1)
	stopping = asyncio.create_task(adapter.stop_typing("12345"))
	tasks.append(stopping)
	# stop pops/cancels old; this turn resumes before old's finalizer.
	await asyncio.sleep(0)
	assert "12345" not in adapter._typing_tasks
	assert not old.done()
	await adapter.send_typing("12345")
	replacement = adapter._typing_tasks["12345"]
	tasks.append(replacement)
	await stopping
	assert old.done()
	assert adapter._typing_tasks.get("12345") is replacement
	assert not replacement.done()
	await adapter.send_typing("12345")
	assert adapter._typing_tasks["12345"] is replacement
	await adapter.stop_typing("12345")
	assert replacement.done()
	assert "12345" not in adapter._typing_tasks
	await adapter.stop_typing("12345")
	finally:
	for task in tasks:
	task.cancel()
	await asyncio.gather(*tasks, return_exceptions=True)

[yowmamasita]

I hit this in production on a Discord forum thread. It happened twice in four days. The bot kept showing "is typing" for hours after its last reply. The gateway was idle apart from one POST to /channels/{id}/typing every 12 seconds. A later turn in the same thread completed normally and did not clear it. Only a restart did. That fits this race. The orphaned loop is no longer in _typing_tasks so no stop_typing call can reach it.

The Normal final-send NOT suppressed warning from #85427 was not the cause for me. It fired on 28 of the 50 turns in that thread. The second leak came on a turn that never logged it.

The race is reachable from the real callers. It does not need a hand built interleaving. The base _keep_typing refresh is still running when _hmwa_stop_typing_for_turn stops typing at the end of a turn. If the event loop stalls long enough for that stop and the next refresh tick to come due together then both run in the same loop iteration. The tick registers a new loop before the cancelled one unwinds. The second test below drives exactly that path. It runs _keep_typing and the turn end stop and then the _stop_typing_refresh cleanup from the base message task. Without the 0.3 second stall it passes on main. That is why the bug only shows up now and then. The first test stops through asyncio.create_task(adapter.stop_typing(...)) as @omgbabyweb suggested above.

Tested against main at c62bd9f207.

	main	main + this PR
test_discord_typing_stuck.py + the two tests below	3 failed, 1 passed	4 passed
tests/gateway -k "discord or typing"	12 failed	9 failed

The 9 are the same tests on both sides and unrelated to typing. The adapter hunk no longer applies to main because the file was reorganized. The conflict is only the surrounding context and the change ports as is. I have been running it in production since this morning.

tests/gateway/test_discord_typing_orphan.py

"""A Discord typing loop must never outlive every stop_typing() call.

stop_typing() pops the loop, cancels it and awaits it. A send_typing() that
lands before the cancelled loop unwinds registers a successor, and the old
loop's finally used to pop that successor out of _typing_tasks, orphaning it.
"""
import asyncio
import time
from unittest.mock import AsyncMock, MagicMock

import pytest

from gateway.config import PlatformConfig
from plugins.platforms.discord.adapter import DiscordAdapter

CHAT = "12345"


def _make_adapter():
    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="***"))
    adapter._client = MagicMock()
    adapter._client.http = MagicMock()
    adapter._client.http.request = AsyncMock()
    adapter._typing_tasks = {}
    return adapter


def _live_typing_loops():
    return [
        t for t in asyncio.all_tasks()
        if not t.done() and t.get_coro().__qualname__.endswith("._typing_loop")
    ]


async def _cancel_leftovers():
    leftovers = _live_typing_loops()
    for t in leftovers:
        t.cancel()
    await asyncio.gather(*leftovers, return_exceptions=True)


@pytest.mark.asyncio
async def test_send_typing_inside_stop_window_is_still_stoppable():
    adapter = _make_adapter()
    try:
        await adapter.send_typing(CHAT)
        await asyncio.sleep(0)  # first POST, loop parks in sleep(12)

        stopper = asyncio.create_task(adapter.stop_typing(CHAT))
        await asyncio.sleep(0)  # stopper popped + cancelled the loop, now awaits it
        await adapter.send_typing(CHAT)  # a refresh tick lands in that window
        await stopper

        # Any later stop (e.g. the end of the next turn) must reach it.
        await adapter.stop_typing(CHAT)
        await asyncio.sleep(0)
        assert _live_typing_loops() == []
    finally:
        await _cancel_leftovers()


@pytest.mark.asyncio
async def test_turn_end_stop_racing_keep_typing_tick_leaks_nothing():
    """Real callers: the base _keep_typing refresh, the gateway's turn-end
    stop_typing, then the base message task's _stop_typing_refresh cleanup."""
    adapter = _make_adapter()
    try:
        stop_event = asyncio.Event()
        keep = asyncio.create_task(
            adapter._keep_typing(CHAT, interval=0.5, stop_event=stop_event)
        )

        async def gateway_turn_end_stop():
            # gateway/run_turn.py _hmwa_stop_typing_for_turn runs while the
            # base refresh task is still alive.
            await asyncio.sleep(0.4)
            await adapter.stop_typing(CHAT)

        async def loop_stall():
            # Synchronous work on the event loop. The stop (due 0.4s) and the
            # next refresh tick (due 0.5s) both come due during it and run
            # back to back in one loop iteration. Without the stall this
            # test passes on unfixed code, which is why the bug is sporadic.
            await asyncio.sleep(0.3)
            time.sleep(0.3)

        await asyncio.gather(gateway_turn_end_stop(), loop_stall())
        await asyncio.sleep(0.05)

        # Base message task finally: stop the refresh, then one bounded stop.
        await adapter._stop_typing_refresh(CHAT, keep)
        await adapter._stop_typing_refresh(CHAT, None, stop_attempts=1)
        await asyncio.sleep(0)
        assert _live_typing_loops() == []
    finally:
        await _cancel_leftovers()