fix(channels): retain strong reference to Discord ack-reaction tasks (#5049)

* fix(channels): retain strong reference to Discord ack-reaction tasks

The event loop keeps only weak references to scheduled tasks, so the bare
asyncio.create_task(self._add_reaction(message)) at both message-acceptance
sites could be garbage-collected mid-flight and silently drop the ✅
acknowledgment. The module-level retention set holds each task until
completion and logs failures, mirroring the deferred subagent cleanup fix
in #4928 and the stream bridge cleanup fix in #4931.

* fix(channels): drain in-flight ack reactions on Discord channel shutdown

Adversarial follow-up to the retention fix: stop() cancelled typing tasks
but left in-flight ack-reaction tasks in the module retention set when the
channel stopped mid-HTTP-call, pinning the channel instance and the discord
Message object graph across restart cycles. _cancel_ephemeral_tasks /
_discard_ephemeral_tasks now cover both families at every stop() cleanup
point and the _run_client finally. Also tests the done-callback error-log
branch and documents the GC test as a contract pin.

* test(channels): pin Discord ack-reaction drain wiring and sharpen wording

Second adversarial pass found the drain test cancelled a never-started
task (no yield between schedule and cancel), nothing exercised the real
stop() cross-loop wiring, and the cleanup log messages still said
"typing-task" while now covering both families. The drain test now yields
so the task genuinely suspends first, a new test drives stop() through a
real background loop and fails if the wiring reverts to typing-only
cleanup, wording covers ephemeral tasks, and the discard path snapshots
the set before iterating.

* refactor(channels): scope Discord ack-reaction retention to the channel instance

The retention set was module-level, so one DiscordChannel's stop() would
cancel every other instance's in-flight ack reactions. Instance-level
matches the typing-task bookkeeping and keeps shutdown scoped; pinned by
a two-channel isolation test.

* test(channels): drive two-channel ack isolation through full stop()

Independent review of the instance-scoping refactor confirmed the P2 fix
but suggested matching the review's letter exactly: the isolation test now
shuts the first channel down via stop() instead of the drain helper, and
one stale docstring reference to the former module-level set is corrected.

backend/app/channels/discord.py

@@ -67,6 +67,15 @@ class DiscordChannel(Channel):
         # Typing indicator management
         self._typing_tasks: dict[str, asyncio.Task] = {}
 
+        # Strong references for this channel's in-flight ack-reaction tasks.
+        # The event loop keeps only weak references to scheduled tasks, so a
+        # bare ``asyncio.create_task`` could be garbage-collected mid-flight
+        # and silently drop the acknowledgment reaction (same retention
+        # pattern as the deferred subagent cleanup in #4928). Instance-level
+        # on purpose, matching ``_typing_tasks``: one channel's shutdown must
+        # not cancel another instance's in-flight reactions.
+        self._ack_reaction_tasks: set[asyncio.Task[None]] = set()
+
         self._client = None
         self._thread: threading.Thread | None = None
         self._discord_loop: asyncio.AbstractEventLoop | None = None
@@ -177,25 +186,26 @@ class DiscordChannel(Channel):
         discord_loop = self._discord_loop
         current_loop = asyncio.get_running_loop()
         if discord_loop is None or discord_loop is current_loop:
-            await self._cancel_typing_tasks()
+            await self._cancel_ephemeral_tasks()
         elif discord_loop.is_running():
-            # Serialize cleanup with _start_typing() on the owning loop.  The
+            # Serialize cleanup with _start_typing()/_schedule_ack_reaction()
+            # on the owning loop.  The
             # loop may exit between is_running() and scheduling, so neither
             # scheduling nor waiting is allowed to block the rest of stop().
-            cleanup_coro = self._cancel_typing_tasks()
+            cleanup_coro = self._cancel_ephemeral_tasks()
             try:
                 cleanup_future = asyncio.run_coroutine_threadsafe(cleanup_coro, discord_loop)
             except RuntimeError:
                 cleanup_coro.close()
-                logger.warning("[Discord] event loop stopped before typing-task cleanup could be scheduled")
+                logger.warning("[Discord] event loop stopped before ephemeral-task cleanup could be scheduled")
             else:
                 try:
                     await asyncio.wait_for(asyncio.wrap_future(cleanup_future), timeout=10)
                 except TimeoutError:
                     cleanup_future.cancel()
-                    logger.warning("[Discord] typing-task cleanup timed out after 10s")
+                    logger.warning("[Discord] ephemeral-task cleanup timed out after 10s")
                 except Exception:
-                    logger.exception("[Discord] error while cleaning up typing tasks")
+                    logger.exception("[Discord] error while cleaning up ephemeral tasks")
 
         if self._client and discord_loop and discord_loop.is_running():
             close_coro = self._client.close()
@@ -222,7 +232,7 @@ class DiscordChannel(Channel):
         # discard the stale references here; awaiting them from this loop would
         # raise a cross-loop RuntimeError.
         if discord_loop and discord_loop is not current_loop and not discord_loop.is_running():
-            self._discard_typing_tasks()
+            self._discard_ephemeral_tasks()
 
         self._client = None
         self._discord_loop = None
@@ -315,6 +325,36 @@ class DiscordChannel(Channel):
                 )
         self._typing_tasks.clear()
 
+    async def _cancel_ack_reaction_tasks(self) -> None:
+        """Cancel in-flight ack reactions so stop() does not strand them.
+
+        A task interrupted mid-HTTP-call would otherwise stay in the
+        retention set forever, pinning this channel instance and the discord
+        Message object graph after shutdown.
+        """
+        pending = [task for task in self._ack_reaction_tasks if not task.done()]
+        for task in pending:
+            task.cancel()
+        if pending:
+            await asyncio.gather(*pending, return_exceptions=True)
+
+    def _discard_ack_reaction_tasks(self) -> None:
+        """Forget stale ack-reaction tasks after the owning loop stopped."""
+        for task in list(self._ack_reaction_tasks):
+            if not task.done():
+                logger.warning("[Discord] discarding pending ack-reaction task for stopped loop")
+        self._ack_reaction_tasks.clear()
+
+    async def _cancel_ephemeral_tasks(self) -> None:
+        """Cancel typing indicators and in-flight ack reactions together."""
+        await self._cancel_typing_tasks()
+        await self._cancel_ack_reaction_tasks()
+
+    def _discard_ephemeral_tasks(self) -> None:
+        """Forget stale typing and ack-reaction tasks (stopped-loop fallback)."""
+        self._discard_typing_tasks()
+        self._discard_ack_reaction_tasks()
+
     async def _stop_typing(self, chat_id: str, thread_ts: str | None = None) -> None:
         """Stops the typing loop for a specific target."""
         target_id = thread_ts or chat_id
@@ -330,6 +370,25 @@ class DiscordChannel(Channel):
         except Exception:
             logger.debug("[Discord] failed to add reaction to message %s", message.id, exc_info=True)
 
+    def _on_ack_reaction_task_done(self, task: asyncio.Task[None]) -> None:
+        self._ack_reaction_tasks.discard(task)
+        if task.cancelled():
+            return
+        exc = task.exception()
+        if exc is not None:
+            logger.error("[Discord] ack reaction task failed: %s", exc)
+
+    def _schedule_ack_reaction(self, message) -> asyncio.Task[None]:
+        """Schedule the ack reaction with a strong task reference.
+
+        The delayed HTTP call must survive until completion; see
+        ``_ack_reaction_tasks`` for why a bare ``create_task`` is unsafe here.
+        """
+        task = asyncio.create_task(self._add_reaction(message))
+        self._ack_reaction_tasks.add(task)
+        task.add_done_callback(self._on_ack_reaction_task_done)
+        return task
+
     async def _on_message(self, message) -> None:
         if not self._running or not self._client:
             return
@@ -450,7 +509,7 @@ class DiscordChannel(Channel):
                     # Start typing indicator in the thread
                     if typing_target:
                         await self._start_typing(typing_target, chat_id, thread_id)
-                    asyncio.create_task(self._add_reaction(message))
+                    self._schedule_ack_reaction(message)
                     return
 
                 # Thread not tracked (orphaned) — create new thread and handle below
@@ -561,7 +620,7 @@ class DiscordChannel(Channel):
             # Start typing/reaction only after bounded admission succeeds.
             if typing_target:
                 await self._start_typing(typing_target, chat_id, thread_id)
-            asyncio.create_task(self._add_reaction(message))
+            self._schedule_ack_reaction(message)
         finally:
             if not reservation_transferred:
                 reservation.release()
@@ -642,9 +701,9 @@ class DiscordChannel(Channel):
                 logger.exception("Discord client error")
         finally:
             try:
-                self._discord_loop.run_until_complete(self._cancel_typing_tasks())
+                self._discord_loop.run_until_complete(self._cancel_ephemeral_tasks())
             except Exception:
-                logger.exception("Error while cleaning up Discord typing tasks")
+                logger.exception("Error while cleaning up Discord ephemeral tasks")
             try:
                 if self._client and not self._client.is_closed():
                     self._discord_loop.run_until_complete(self._client.close())

backend/tests/test_discord_channel.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 import asyncio
 import builtins
+import gc
 import threading
+import weakref
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
@@ -263,3 +265,171 @@ async def test_send_file_closes_handle_when_send_fails(tmp_path) -> None:
         assert handles[0].closed is True
     finally:
         _stop_bg_loop(bg_loop, bg_thread)
+
+
+@pytest.mark.asyncio
+async def test_ack_reaction_task_retained_under_gc() -> None:
+    """The ack-reaction task is retained and runs to completion.
+
+    This pins the retention contract (scheduled → in the set, done →
+    discarded): the fake coroutine never suspends on an unrooted future, so
+    actual mid-flight GC cannot be reproduced deterministically here — the
+    same limitation the #4928 precedent test has.
+
+    A bare ``asyncio.create_task`` holds only a weak loop reference, so the
+    ✅ acknowledgment could be garbage-collected mid-flight. The instance-level
+    retention set keeps the task strongly referenced until completion (same
+    pattern as #4928 / #4931).
+    """
+    bus = MessageBus()
+    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})
+
+    reacted = asyncio.Event()
+
+    async def _add_reaction(_emoji: str) -> None:
+        reacted.set()
+
+    message = SimpleNamespace(id=111, add_reaction=_add_reaction)
+
+    task = channel._schedule_ack_reaction(message)
+    weak_task = weakref.ref(task)
+    del task
+    gc.collect()
+
+    retained = weak_task()
+    assert retained is not None, "ack reaction task was garbage-collected mid-flight"
+    assert retained in channel._ack_reaction_tasks
+
+    await asyncio.wait_for(retained, timeout=1.0)
+    assert reacted.is_set()
+    # Completed tasks are discarded so the retention set cannot grow unboundedly.
+    assert retained not in channel._ack_reaction_tasks
+
+
+@pytest.mark.asyncio
+async def test_ack_reaction_retention_is_isolated_per_channel() -> None:
+    """One channel's shutdown must not cancel another instance's in-flight reactions.
+
+    The retention set is instance-level (matching ``_typing_tasks``): a
+    module-level set would let ``stop()`` on one channel drain every other
+    channel's pending ack tasks.
+    """
+    first = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+    second = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+
+    release = asyncio.Event()
+
+    async def _hang(_emoji: str) -> None:
+        await release.wait()
+
+    task = second._schedule_ack_reaction(SimpleNamespace(id=666, add_reaction=_hang))
+    await asyncio.sleep(0)  # let the task start and suspend
+    assert task in second._ack_reaction_tasks
+    assert task not in first._ack_reaction_tasks
+
+    # Full shutdown of the first channel must leave the second instance's
+    # in-flight reaction alone (the review asked for independent shutdown).
+    await first.stop()
+
+    assert not task.done()
+    assert task in second._ack_reaction_tasks
+
+    release.set()
+    await asyncio.wait_for(task, timeout=1.0)
+
+
+@pytest.mark.asyncio
+async def test_ack_reaction_task_survives_reaction_failure() -> None:
+    """A failing add_reaction completes quietly and is discarded from the set."""
+
+    bus = MessageBus()
+    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})
+
+    async def _boom(_emoji: str) -> None:
+        raise RuntimeError("discord api down")
+
+    message = SimpleNamespace(id=222, add_reaction=_boom)
+
+    task = channel._schedule_ack_reaction(message)
+    await asyncio.wait_for(task, timeout=1.0)
+    assert task not in channel._ack_reaction_tasks
+
+
+@pytest.mark.asyncio
+async def test_stop_drains_in_flight_ack_reaction_tasks() -> None:
+    """stop()'s cleanup path cancels in-flight ack reactions on the owning loop.
+
+    Without the drain, a task interrupted mid-HTTP-call would sit in the
+    module retention set forever, pinning the channel and Message graph.
+    """
+    release = asyncio.Event()
+
+    async def _hang(_emoji: str) -> None:
+        await release.wait()
+
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+    message = SimpleNamespace(id=333, add_reaction=_hang)
+
+    task = channel._schedule_ack_reaction(message)
+    assert task in channel._ack_reaction_tasks
+    # Yield once so the task actually starts and suspends inside _hang —
+    # cancelling a never-started task would not exercise the mid-flight path.
+    await asyncio.sleep(0)
+    assert not task.done()
+
+    await channel._cancel_ephemeral_tasks()
+
+    assert task.cancelled()
+    assert task not in channel._ack_reaction_tasks
+
+
+@pytest.mark.asyncio
+async def test_ack_reaction_task_failure_is_logged_and_discarded(caplog) -> None:
+    """An exception escaping _add_reaction is logged at error and discarded."""
+    bus = MessageBus()
+    channel = DiscordChannel(bus=bus, config={"bot_token": "token"})
+
+    async def _explode(_self, _message) -> None:
+        raise RuntimeError("unexpected boom")
+
+    with patch.object(DiscordChannel, "_add_reaction", _explode), caplog.at_level("ERROR", logger="app.channels.discord"):
+        task = channel._schedule_ack_reaction(SimpleNamespace(id=444))
+        with pytest.raises(RuntimeError, match="unexpected boom"):
+            await asyncio.wait_for(task, timeout=1.0)
+
+    assert task not in channel._ack_reaction_tasks
+    assert any("ack reaction task failed" in record.message for record in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_stop_wiring_drains_ack_tasks_across_loops() -> None:
+    """The real stop() path (cross-loop branch) drains in-flight ack reactions.
+
+    Guards the wiring itself: reverting stop() to typing-only cleanup must
+    fail here, not just at the helper level.
+    """
+    bg_loop, bg_thread = _start_bg_loop()
+    try:
+        channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+        channel._discord_loop = bg_loop
+
+        release = asyncio.Event()
+
+        async def _hang(_emoji: str) -> None:
+            await release.wait()
+
+        async def _schedule_on_bg_loop() -> asyncio.Task:
+            task = channel._schedule_ack_reaction(SimpleNamespace(id=555, add_reaction=_hang))
+            await asyncio.sleep(0.05)  # let the task start and suspend on the event
+            return task
+
+        schedule_future = asyncio.run_coroutine_threadsafe(_schedule_on_bg_loop(), bg_loop)
+        task = await asyncio.wrap_future(schedule_future)
+        assert task in channel._ack_reaction_tasks
+
+        await channel.stop()
+
+        assert task.cancelled()
+        assert not channel._ack_reaction_tasks
+    finally:
+        _stop_bg_loop(bg_loop, bg_thread)