fix(discord): claim dedup ID in missed-message backfill dispatch

plugins/platforms/discord/adapter.py

@@ -2129,9 +2129,11 @@ async def _run_missed_message_backfill(self) -> None:
                 scanned += 1
                 message_id = str(getattr(message, "id", ""))
                 self._record_discord_message_seen(message, status="discovered")
-                # A live gateway event may race this REST scan. Check without
-                # claiming the ID; the shared ingress helper owns the dedup
-                # write immediately before normal auth/filter dispatch.
+                # A live gateway event may race this REST scan. Cheap
+                # pre-filter only: the actual claim happens atomically in
+                # _dispatch_recovered_message via claim=True, so whichever
+                # path (live or backfill) claims first wins and the other
+                # drops the duplicate.
                 if self._dedup.contains(message_id):
                     continue
                 if not await self._should_backfill_discord_message(message):
@@ -2215,7 +2217,12 @@ async def _dispatch_recovered_message(self, message: Any) -> bool:
             ):
                 return False
         admitted, role_authorized = self._discord_message_admission(
-            message, claim=False,
+            # Claim the ID: without this the message never enters the dedup
+            # cache, so a live gateway replay of the same event (Discord
+            # replays missed events on resume, racing this REST scan) is
+            # admitted again and the user gets two runs / two replies.
+            # Failure paths in the caller release the claim via discard().
+            message, claim=True,
         )
         if not admitted:
             return False

tests/gateway/test_discord_missed_message_backfill.py

@@ -458,3 +458,81 @@ async def _gen():
     assert got == [2, 3, 4]
 
 
+
+
+@pytest.mark.asyncio
+async def test_recovered_dispatch_claims_dedup_id(adapter):
+    """Backfilled messages must enter the dedup cache.
+
+    Regression: the backfill path admitted with claim=False, so a recovered
+    message never entered the dedup cache — a live gateway replay of the same
+    event (Discord replays missed events on resume) was admitted again and
+    the user got two runs / two replies.
+    """
+    bot_user = adapter._client.user
+    message = make_message(
+        message_id=555,
+        content=f"<@{bot_user.id}> please ingest",
+        mentions=[bot_user],
+    )
+
+    assert await adapter._dispatch_recovered_message(message) is True
+    assert adapter._handle_message.call_count == 1
+    assert adapter._dedup.contains("555") is True
+
+    # A live replay of the same message must now be dropped.
+    admitted, _ = adapter._discord_message_admission(message, claim=True)
+    assert admitted is False
+    assert adapter._handle_message.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_recovered_dispatch_second_attempt_dropped(adapter):
+    """A second backfill dispatch of the same message is a no-op."""
+    bot_user = adapter._client.user
+    message = make_message(
+        message_id=777,
+        content=f"<@{bot_user.id}> please ingest",
+        mentions=[bot_user],
+    )
+
+    assert await adapter._dispatch_recovered_message(message) is True
+    assert await adapter._dispatch_recovered_message(message) is False
+    assert adapter._handle_message.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_failed_recovered_dispatch_releases_claim_for_retry(adapter, monkeypatch):
+    """A dispatch that raises must release the claimed dedup ID.
+
+    The backfill now claims the ID at dispatch time; the scan loop's error
+    path calls discard() so a transient failure doesn't permanently suppress
+    the message from a later retry (or the live replay).
+    """
+    bot_user = adapter._client.user
+    message = make_message(
+        message_id=888,
+        content=f"<@{bot_user.id}> please ingest",
+        mentions=[bot_user],
+    )
+
+    async def fake_candidates(_channels):
+        yield message
+
+    monkeypatch.setenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", "123")
+    monkeypatch.setattr(adapter, "_iter_missed_message_backfill_candidates", fake_candidates)
+    monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True))
+    monkeypatch.setattr(adapter, "_missed_message_backfill_max_dispatches", lambda: 10)
+    monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"})
+    monkeypatch.setattr("asyncio.sleep", AsyncMock())
+
+    # First pass: dispatch fails -> claim must be released by the error path.
+    adapter._handle_message = AsyncMock(side_effect=RuntimeError("transient"))
+    await adapter._run_missed_message_backfill()
+    assert adapter._dedup.contains("888") is False
+
+    # Retry with a healthy handler: the message dispatches and stays claimed.
+    adapter._handle_message = AsyncMock(return_value=True)
+    await adapter._run_missed_message_backfill()
+    assert adapter._handle_message.call_count == 1
+    assert adapter._dedup.contains("888") is True