fix(gateway): don't spend a redelivery attempt when the platform is down

The delivery ledger durably records a final response before the send so a
crash between finalize and platform ACK can redeliver it on the next boot.
attempts is that redelivery budget, capped at MAX_ATTEMPTS=3.

sweep_recoverable() claims every dead-owner row and increments attempts
before the caller knows whether it can send. self.adapters only holds a
platform after its connect() succeeded, so when the platform failed to
connect this boot _redeliver_pending_obligations() hits its "adapter is
None" branch and continues WITHOUT sending — but the attempt is already
spent. Three such boots and the row abandons, having never been sent once.

That is the loss the ledger exists to prevent, and the trigger correlates
with the crash that created the obligation: the network trouble that killed
the send tends to still be there on the next boot. Worse, the message stays
lost — once abandoned it is never retried even after the platform recovers.

Reproduced against the real runner with an unconnected adapter:

    boot 1: claimed=1 state='attempting' attempts=1  (0 sends attempted)
    boot 2: claimed=1 state='attempting' attempts=2  (0 sends attempted)
    boot 3: claimed=1 state='attempting' attempts=3  (0 sends attempted)
    boot 4: claimed=0 state='abandoned'  attempts=3  (0 sends attempted)

Let the caller declare which platforms it can send on, and skip claiming
rows for the others. attempts then only ever buys a real send. Rows for a
platform that never returns are still bounded by the stale cutoff, so
nothing accumulates. The parameter is keyword-only and optional — omitting
it keeps the previous claim-everything behaviour for other callers.

gateway/delivery_ledger.py

@@ -200,7 +200,11 @@ def _update_state(obligation_id: str, state: str, error: str = "") -> None:
         )
 
 
-def sweep_recoverable(now: Optional[float] = None) -> List[Dict[str, Any]]:
+def sweep_recoverable(
+    now: Optional[float] = None,
+    *,
+    deliverable_platforms: Optional[set] = None,
+) -> List[Dict[str, Any]]:
     """Claim undelivered rows owned by dead processes; return them for
     redelivery.
 
@@ -209,6 +213,13 @@ def sweep_recoverable(now: Optional[float] = None) -> List[Dict[str, Any]]:
     double-claim (the UPDATE is guarded on the previous owner stamp).
     Rows over the attempts cap or older than the stale cutoff transition to
     'abandoned' instead of being returned.
+
+    ``deliverable_platforms`` (platform value strings) restricts claiming to
+    platforms the caller can actually send on this boot.  ``attempts`` is the
+    redelivery budget, so it must only be spent on a real send: a platform
+    that failed to connect would otherwise burn one attempt per boot and hit
+    the cap having never been sent once.  Rows for absent platforms are left
+    untouched for a later boot; the stale cutoff still bounds them.
     """
     now = now if now is not None else time.time()
     pid, started = _owner_stamp()
@@ -232,6 +243,13 @@ def sweep_recoverable(now: Optional[float] = None) -> List[Dict[str, Any]]:
                     (now, oid),
                 )
                 continue
+            if (
+                deliverable_platforms is not None
+                and platform not in deliverable_platforms
+            ):
+                # No adapter for this platform this boot — the caller cannot
+                # send, so claiming would spend an attempt on a no-op.
+                continue
             cursor = conn.execute(
                 """UPDATE delivery_obligations
                    SET owner_pid=?, owner_started_at=?, attempts=attempts+1,

gateway/run.py

@@ -6981,7 +6981,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
             if not ledger_enabled():
                 return 0
-            claimed = await asyncio.to_thread(sweep_recoverable)
+            # Only claim rows we can actually send this boot: self.adapters
+            # holds a platform only after its connect() succeeded, and each
+            # claim spends one of the row's three redelivery attempts.
+            _deliverable = {
+                getattr(p, "value", str(p)) for p in self.adapters
+            }
+            claimed = await asyncio.to_thread(
+                sweep_recoverable, None, deliverable_platforms=_deliverable
+            )
         except Exception:
             logger.debug("delivery ledger sweep failed", exc_info=True)
             return 0

tests/gateway/test_delivery_ledger.py

@@ -281,3 +281,124 @@ class TestGatewayRedeliverySweep:
             n = await runner._redeliver_pending_obligations()
         assert n == 0
         adapter.send.assert_not_awaited()
+
+
+class TestAttemptsOnlySpentOnRealSends:
+    """``attempts`` is the redelivery budget — it must buy a send.
+
+    ``self.adapters`` only holds a platform after its ``connect()`` succeeded,
+    and the sweep claimed every dead-owner row regardless. A platform that
+    failed to connect this boot therefore burned one attempt per boot while
+    the caller's ``adapter is None`` branch skipped it without sending — so
+    after MAX_ATTEMPTS boots the row abandoned having never been sent once,
+    losing exactly the response the ledger exists to guarantee. That failure
+    correlates with the crash that created the obligation: the network
+    trouble that killed the send tends to still be there on the next boot.
+    """
+
+    def test_absent_platform_does_not_burn_attempts(self):
+        _record(platform="telegram")
+        dl.mark_attempting("ob-1")
+
+        for _ in range(dl.MAX_ATTEMPTS + 2):
+            _orphan("ob-1")
+            assert dl.sweep_recoverable(deliverable_platforms={"discord"}) == []
+
+        row = dl.debug_rows()
+        assert "abandoned" not in row
+        with dl._connect() as conn:
+            state, attempts = conn.execute(
+                "SELECT state, attempts FROM delivery_obligations "
+                "WHERE obligation_id=?", ("ob-1",),
+            ).fetchone()
+        assert attempts == 0, "an unsendable boot must not spend the budget"
+        assert state == "attempting"
+
+    def test_row_still_delivers_once_its_platform_returns(self):
+        _record(platform="telegram")
+        for _ in range(dl.MAX_ATTEMPTS + 2):
+            _orphan("ob-1")
+            dl.sweep_recoverable(deliverable_platforms={"discord"})
+
+        _orphan("ob-1")
+        claimed = dl.sweep_recoverable(deliverable_platforms={"telegram"})
+        assert len(claimed) == 1
+        assert claimed[0]["attempts"] == 1
+
+    def test_present_platform_still_claims(self):
+        _record(platform="slack")
+        _orphan("ob-1")
+        claimed = dl.sweep_recoverable(deliverable_platforms={"slack"})
+        assert len(claimed) == 1
+
+    def test_omitting_the_filter_claims_everything(self):
+        """Back-compat: existing callers pass no platform set."""
+        _record(platform="telegram")
+        _orphan("ob-1")
+        assert len(dl.sweep_recoverable()) == 1
+
+    def test_stale_rows_abandon_even_when_undeliverable(self):
+        """The cutoff still bounds rows whose platform never returns."""
+        _record(platform="telegram")
+        _orphan("ob-1")
+        future = time.time() + dl.STALE_AFTER_SECONDS + 10
+        assert dl.sweep_recoverable(
+            now=future, deliverable_platforms={"discord"}
+        ) == []
+        with dl._connect() as conn:
+            state = conn.execute(
+                "SELECT state FROM delivery_obligations WHERE obligation_id=?",
+                ("ob-1",),
+            ).fetchone()[0]
+        assert state == "abandoned"
+
+
+class TestUnconnectedPlatformKeepsItsBudget:
+    """End-to-end through the real runner: boots where the platform failed to
+    connect must not consume the row's redelivery budget."""
+
+    @staticmethod
+    def _runner_without_slack():
+        from gateway.run import GatewayRunner
+
+        runner = object.__new__(GatewayRunner)
+        runner.adapters = {}  # slack failed to connect this boot
+        _store = MagicMock()
+        _store.clear_resume_pending = AsyncMock()
+        _store._store = None
+        runner.session_store = None
+        runner._async_session_store = _store
+        return runner
+
+    @pytest.mark.asyncio
+    async def test_row_survives_boots_where_its_platform_is_down(self):
+        _record(platform="slack")
+        dl.mark_attempting("ob-1")
+
+        for _ in range(dl.MAX_ATTEMPTS + 1):
+            _orphan("ob-1")
+            runner = self._runner_without_slack()
+            assert await runner._redeliver_pending_obligations() == 0
+
+        assert _row("ob-1")["state"] != "abandoned", (
+            "the obligation was abandoned without a single send being attempted"
+        )
+        assert _row("ob-1")["attempts"] == 0
+
+    @pytest.mark.asyncio
+    async def test_delivers_when_the_platform_comes_back(self):
+        from gateway.config import Platform
+
+        _record(platform="slack")
+        for _ in range(dl.MAX_ATTEMPTS + 1):
+            _orphan("ob-1")
+            await self._runner_without_slack()._redeliver_pending_obligations()
+
+        _orphan("ob-1")
+        adapter = MagicMock()
+        adapter.send = AsyncMock(return_value=MagicMock(success=True, error=""))
+        runner = self._runner_without_slack()
+        runner.adapters = {Platform.SLACK: adapter}
+
+        assert await runner._redeliver_pending_obligations() == 1
+        assert _row("ob-1")["state"] == "delivered"