fix(gateway): label redelivery after mid-recovery crash

gateway/delivery_ledger.py

@@ -241,11 +241,13 @@ def sweep_recoverable(
     """Claim undelivered rows owned by dead processes; return them for
     redelivery.
 
-    Claiming atomically re-stamps the owner to THIS process and increments
-    ``attempts``, so a second gateway racing the same sweep cannot
-    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.
+    Claiming atomically re-stamps the owner to THIS process, sets
+    ``state='attempting'``, and increments ``attempts``, so a second
+    gateway racing the same sweep cannot double-claim (the UPDATE is
+    guarded on the previous owner stamp) and a crash mid-redelivery is
+    treated as ambiguous on the next boot. 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
@@ -285,7 +287,8 @@ def sweep_recoverable(
                 continue
             cursor = conn.execute(
                 """UPDATE delivery_obligations
-                   SET owner_pid=?, owner_started_at=?, attempts=attempts+1,
+                   SET state='attempting',
+                       owner_pid=?, owner_started_at=?, attempts=attempts+1,
                        updated_at=?
                    WHERE obligation_id=? AND (owner_pid IS ? OR owner_pid=?)""",
                 (pid, started, now, oid, owner_pid, owner_pid),
@@ -298,9 +301,12 @@ def sweep_recoverable(
                     "chat_id": chat_id,
                     "thread_id": thread_id,
                     "content": content,
-                    # pending = send never started, redeliver plainly;
-                    # attempting/failed = ambiguous or rejected, carry marker.
-                    "needs_marker": state != "pending",
+                    # pending + attempts==0 = send never started, redeliver
+                    # plainly. attempting/failed, or a prior reclaim that
+                    # already spent an attempt (stuck pending rows from
+                    # older builds), carry the recovered-reply marker so a
+                    # possible platform-side duplicate is labeled.
+                    "needs_marker": state != "pending" or attempts >= 1,
                     "attempts": attempts + 1,
                 })
     return claimed

gateway/run.py

@@ -10242,6 +10242,7 @@ async def _redeliver_pending_obligations(self) -> int:
             from gateway.delivery_ledger import (
                 RECOVERED_MARKER,
                 ledger_enabled,
+                mark_attempting,
                 mark_delivered,
                 mark_failed,
                 sweep_recoverable,
@@ -10285,6 +10286,15 @@ async def _redeliver_pending_obligations(self) -> int:
             metadata = (
                 {"thread_id": row["thread_id"]} if row.get("thread_id") else None
             )
+            # Mirror the producer path: claim already stamps attempting, but
+            # stamp again immediately before await so any crash mid-send
+            # leaves an unambiguous ledger state for the next boot.
+            try:
+                await asyncio.to_thread(mark_attempting, row["obligation_id"])
+            except Exception:
+                logger.debug(
+                    "delivery ledger mark_attempting failed", exc_info=True
+                )
             try:
                 result = await adapter.send(
                     chat_id=row["chat_id"],

tests/gateway/test_delivery_ledger.py

@@ -118,10 +118,47 @@ def test_dead_owner_pending_claimed_without_marker(self):
         assert len(claimed) == 1
         assert claimed[0]["needs_marker"] is False
         assert claimed[0]["attempts"] == 1
+        # Claim starts the recovery send: state must leave pending so a
+        # crash mid-redelivery is treated as ambiguous on the next boot.
+        assert _row("ob-1")["state"] == "attempting"
         # Claim re-stamps ownership: a second sweep in the same (live)
         # process must not double-claim.
         assert dl.sweep_recoverable() == []
 
+    def test_pending_reclaim_after_prior_attempt_needs_marker(self):
+        """Crash mid-redelivery of a pending row must not silent-duplicate.
+
+        First reclaim of orphaned pending redelivers plainly (send never
+        started). If the process dies after that claim/send starts but
+        before mark_delivered, the next reclaim must carry the recovered
+        marker — even if state was stuck at pending historically.
+        """
+        _record()
+        _orphan("ob-1")
+        first = dl.sweep_recoverable()
+        assert first[0]["needs_marker"] is False
+        assert _row("ob-1")["state"] == "attempting"
+        assert _row("ob-1")["attempts"] == 1
+
+        _orphan("ob-1")
+        second = dl.sweep_recoverable()
+        assert len(second) == 1
+        assert second[0]["needs_marker"] is True
+        assert second[0]["attempts"] == 2
+
+    def test_stuck_pending_with_prior_attempts_needs_marker(self):
+        """Pre-fix rows: pending + attempts>=1 still need the marker."""
+        _record()
+        with dl._connect() as conn:
+            conn.execute(
+                "UPDATE delivery_obligations SET attempts=1, "
+                "owner_pid=999999999, owner_started_at=1 WHERE obligation_id=?",
+                ("ob-1",),
+            )
+        claimed = dl.sweep_recoverable()
+        assert claimed[0]["needs_marker"] is True
+        assert claimed[0]["attempts"] == 2
+
 
 class TestPrune:
     def test_old_delivered_rows_pruned(self):
@@ -199,6 +236,30 @@ async def test_attempting_redelivers_with_marker(self):
         assert sent["content"].startswith(dl.RECOVERED_MARKER)
         assert sent["content"].endswith("the final answer")
 
+    @pytest.mark.asyncio
+    async def test_second_boot_pending_recovery_carries_marker(self):
+        """Simulate crash after first pending reclaim started send."""
+        _record()
+        _orphan("ob-1")
+        first = self._adapter()
+        n1 = await self._runner(first)._redeliver_pending_obligations()
+        assert n1 == 1
+        assert not first.send.call_args.kwargs["content"].startswith(
+            dl.RECOVERED_MARKER
+        )
+        # Pretend the first recovery send never reached mark_delivered.
+        with dl._connect() as conn:
+            conn.execute(
+                "UPDATE delivery_obligations SET state='attempting', "
+                "owner_pid=999999999, owner_started_at=1 WHERE obligation_id=?",
+                ("ob-1",),
+            )
+        second = self._adapter()
+        await self._runner(second)._redeliver_pending_obligations()
+        sent = second.send.call_args.kwargs["content"]
+        assert sent.startswith(dl.RECOVERED_MARKER)
+        assert sent.endswith("the final answer")
+
     @pytest.mark.parametrize(
         ("send_success", "ledger_method"),
         [(True, "mark_delivered"), (False, "mark_failed")],