fix(discord): advance cursors only after final delivery

Round-robin configured Discord histories under the global scan cap, but move each channel/thread cursor only when its source message reaches successful final delivery.

plugins/platforms/discord/adapter.py

@@ -2057,15 +2057,6 @@ class DiscordAdapter(BasePlatformAdapter):
                     admitted = await self._dispatch_recovered_message(message)
                     if admitted:
                         dispatched += 1
-                    channel_id = str(
-                        getattr(getattr(message, "channel", None), "id", "")
-                    )
-                    if channel_id and message_id:
-                        await asyncio.to_thread(
-                            self._advance_discord_recovery_cursor,
-                            channel_id,
-                            message_id,
-                        )
                 except asyncio.CancelledError:
                     self._dedup.discard(message_id)
                     self._record_recovery_attempt(message, status="cancelled")
@@ -2149,18 +2140,29 @@ class DiscordAdapter(BasePlatformAdapter):
                         continue
                 candidate_channels.append(channel)
 
-        yielded = 0
-        for channel in candidate_channels:
-            async for item in self._iter_channel_and_thread_messages(
+        iterators = [
+            self._iter_channel_and_thread_messages(
                 channel,
                 limit=limit,
                 after=after,
                 seen_channels=seen,
-            ):
+            ).__aiter__()
+            for channel in candidate_channels
+        ]
+        yielded = 0
+        while iterators and yielded < limit:
+            next_round = []
+            for iterator in iterators:
+                try:
+                    item = await iterator.__anext__()
+                except StopAsyncIteration:
+                    continue
                 yield item
                 yielded += 1
+                next_round.append(iterator)
                 if yielded >= limit:
                     return
+            iterators = next_round
 
     async def _iter_channel_and_thread_messages(self, channel: Any, *, limit: int, after: Any, seen_channels: set[str]):
         """Yield history from a channel plus active/recent archived child threads."""
@@ -2459,6 +2461,18 @@ class DiscordAdapter(BasePlatformAdapter):
             )
 
         self._with_discord_recovery_db(_op)
+        if completed:
+            def _channel_for_message(conn):
+                row = conn.execute(
+                    "SELECT COALESCE(thread_id, channel_id) FROM discord_messages "
+                    "WHERE message_id=?",
+                    (reply_to,),
+                ).fetchone()
+                return str(row[0]) if row and row[0] else None
+
+            channel_id = self._with_discord_recovery_db(_channel_for_message)
+            if channel_id:
+                self._advance_discord_recovery_cursor(channel_id, reply_to)
 
     def _discord_message_is_persistently_complete(self, message_id: str) -> bool:
         if not message_id:

tests/gateway/test_discord_missed_message_backfill.py

@@ -824,6 +824,30 @@ async def test_iter_candidates_applies_one_global_scan_limit(adapter, monkeypatc
     assert set(got).issubset({1, 2, 3, 4})
 
 
+@pytest.mark.asyncio
+async def test_iter_candidates_round_robins_configured_channels(adapter, monkeypatch):
+    first = FakeChannel(
+        channel_id=123,
+        history_messages=[
+            make_message(message_id=1),
+            make_message(message_id=2),
+            make_message(message_id=3),
+        ],
+    )
+    second = FakeChannel(
+        channel_id=456,
+        history_messages=[make_message(message_id=4)],
+    )
+    adapter._client.get_channel = lambda channel_id: {123: first, 456: second}[channel_id]
+    monkeypatch.setattr(adapter, "_missed_message_backfill_limit", lambda: 3)
+
+    got = []
+    async for message in adapter._iter_missed_message_backfill_candidates({"123", "456"}):
+        got.append(message.id)
+
+    assert 4 in got
+
+
 @pytest.mark.asyncio
 async def test_iter_candidates_keeps_latest_messages_when_window_exceeds_limit(adapter, monkeypatch):
     class RealisticChannel(FakeChannel):
@@ -864,6 +888,42 @@ def test_recovery_cursor_round_trip_is_channel_scoped(adapter):
     assert adapter._discord_recovery_cursor("456") == "2002"
 
 
+@pytest.mark.asyncio
+async def test_cursor_does_not_advance_past_incomplete_dispatched_message(adapter, monkeypatch):
+    channel = FakeChannel(
+        channel_id=123,
+        history_messages=[
+            make_message(message_id=1),
+            make_message(message_id=2),
+        ],
+    )
+    for message in channel._history_messages:
+        message.channel = channel
+    adapter._client.get_channel = lambda _channel_id: channel
+    monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"})
+    monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True))
+    monkeypatch.setattr(adapter, "_dispatch_recovered_message", AsyncMock(side_effect=[True, True]))
+    monkeypatch.setattr(adapter, "_missed_message_backfill_max_dispatches", lambda: 10)
+
+    await adapter._run_missed_message_backfill()
+
+    assert adapter._discord_recovery_cursor("123") is None
+
+
+def test_final_delivery_advances_channel_cursor(adapter):
+    message = make_message(message_id=103, channel=FakeChannel(channel_id=123))
+    adapter._record_discord_message_seen(message, status="processing")
+
+    adapter._record_discord_response(
+        reply_to="103",
+        result=SimpleNamespace(success=True, message_id="9010"),
+        content="done",
+        final=True,
+    )
+
+    assert adapter._discord_recovery_cursor("123") == "103"
+
+
 @pytest.mark.asyncio
 async def test_iter_candidates_uses_persisted_channel_cursor(adapter, monkeypatch):
     class CursorChannel(FakeChannel):