fix(notifications): exponential backoff + CAS claim for expired-unacked re-escalation (#652)

The sweep re-escalated every expired unacked ack-required notification
on every ~60s tick, forever — the live incident: 3 fresh blocker
escalations + Telegram DMs per minute from a static stale pile. Now
each notification carries reescalation_count / last_reescalated_at /
reescalation_delivered_count (migration 079): first fire at expiry,
then doubling intervals from 1h capped at 24h, hard stop after
ROBOCO_NOTIFICATION_MAX_REESCALATIONS (default 5) with one permanent
log carrying attempts-vs-delivered so 'seen and ignored' is
distinguishable from 'route never worked'. The due/wait/capped decision
is a pure function in foundation/policy/communications.py.

Per adversarial review, the attempt slot is claimed by compare-and-set
(UPDATE ... WHERE reescalation_count = :n) BEFORE delivery — the
previous draft leaned on the 60s dedup window, which never engages for
BLOCKER_ESCALATION (_LOOP_PRONE_TYPES excludes it), so concurrent
sweeps would have double-delivered. A lost claim skips delivery
outright. Legacy rows read as count=0 and keep today's first-fire
semantics. 61 tests incl. a two-session CAS race and a real alembic
upgrade/downgrade round trip.

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

alembic/versions/079_notification_backoff.py

@@ -0,0 +1,64 @@
+"""Add notifications.reescalation_count/_delivered_count + last_reescalated_at.
+
+`sweep_expired_notifications` used to re-escalate EVERY ack-required
+notification past `expires_at` on every ~1min sweep tick, forever — a static
+pile of stale notifications produced a fresh blocker_escalation row (+
+Telegram DM) per row per tick. These three columns back a per-notification
+exponential backoff (first re-escalation at expiry, then doubling from
+`notification_reescalation_base_seconds`, capped at 24h, hard-stopped past
+`notification_max_reescalations`): `reescalation_count` is the attempt
+counter (claimed via compare-and-set even when delivery then fails),
+`reescalation_delivered_count` is how many of those attempts actually
+reached a recipient, `last_reescalated_at` anchors the backoff interval.
+Additive and default-`0`/`NULL`: existing rows read as `count=0`, preserving
+today's first-fire semantics.
+
+Revision ID: 079_notification_backoff
+Revises: 078_project_codegen_command
+Create Date: 2026-07-22
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "079_notification_backoff"
+down_revision = "078_project_codegen_command"
+branch_labels: dict[str, str] | None = None
+depends_on: dict[str, str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "notifications",
+        sa.Column(
+            "reescalation_count",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+        ),
+    )
+    op.add_column(
+        "notifications",
+        sa.Column(
+            "reescalation_delivered_count",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+        ),
+    )
+    op.add_column(
+        "notifications",
+        sa.Column(
+            "last_reescalated_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("notifications", "last_reescalated_at")
+    op.drop_column("notifications", "reescalation_delivered_count")
+    op.drop_column("notifications", "reescalation_count")

roboco/config.py

@@ -302,6 +302,27 @@ class Settings(BaseSettings):
             "(legacy: expires_at stays NULL, notifications never expire)."
         ),
     )
+    notification_reescalation_base_seconds: int = Field(
+        default=3600,
+        ge=1,
+        description=(
+            "Base interval for the per-notification re-escalation backoff: "
+            "the first re-escalation fires at expiry, each one after that "
+            "doubles the wait from this base (1h, 2h, 4h, 8h, ...) capped at "
+            "24h between attempts. Without this a static pile of expired, "
+            "still-unacked notifications re-escalates every sweep tick "
+            "(~1min) forever."
+        ),
+    )
+    notification_max_reescalations: int = Field(
+        default=5,
+        ge=1,
+        description=(
+            "Hard cap on re-escalations per notification. Past this many "
+            "attempts a still-unacked notification is logged once as "
+            "permanently-unacked and never re-escalated again."
+        ),
+    )
     audit_interval_seconds: int = Field(
         default=21600,
         ge=0,

roboco/db/tables.py

@@ -1041,6 +1041,24 @@ class NotificationTable(Base):
         DateTime(timezone=True), nullable=True
     )
 
+    # Re-escalation backoff (sweep_expired_notifications): how many times this
+    # row has been re-escalated past expiry, and when the last one fired —
+    # drives the exponential schedule so a static pile of stale rows doesn't
+    # re-fire every sweep tick forever. `reescalation_count` is the attempt
+    # counter (bumped by a compare-and-set claim even when delivery then
+    # fails); `reescalation_delivered_count` is how many of those attempts
+    # actually reached a recipient — the two can diverge (a broken escalation
+    # chain burns attempts with zero deliveries).
+    reescalation_count: Mapped[int] = mapped_column(
+        Integer, default=0, server_default="0", nullable=False
+    )
+    reescalation_delivered_count: Mapped[int] = mapped_column(
+        Integer, default=0, server_default="0", nullable=False
+    )
+    last_reescalated_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+
     # Timestamps
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), default=lambda: datetime.now(UTC), nullable=False

roboco/foundation/policy/communications.py

@@ -6,10 +6,16 @@ Single source of truth for:
   - NotificationType -> requires_ack mapping (replaces 7 hand-set callsites in
     services/notification_delivery.py)
   - Priority enum (re-exports models.base.NotificationPriority for SQLAlchemy compat)
+  - Re-escalation backoff schedule (pure; consumed by
+    services/notification_delivery.py's sweep)
 """
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Literal
+
 from roboco.foundation.identity import Role
 from roboco.models.base import NotificationPriority, NotificationType
 
@@ -94,3 +100,51 @@ ACK_REQUIRED_BY_TYPE: dict[NotificationType, bool] = {
     NotificationType.MENTION: False,  # chat @mention, no ack
     NotificationType.A2A_REQUEST: False,  # request/reply lives at message layer
 }
+
+
+# Doubling-schedule ceiling: no re-escalation ever waits longer than this
+# between attempts, however high `notification_max_reescalations` is set.
+REESCALATION_INTERVAL_CAP_SECONDS = 24 * 3600
+
+ReescalationDecision = Literal["due", "wait", "capped"]
+
+
+@dataclass(frozen=True)
+class ReescalationPolicy:
+    """The two `settings.notification_*` knobs the backoff schedule reads,
+    bundled so `reescalation_decision` stays under the 5-arg lint ceiling —
+    they always travel together (both come straight from `settings`),
+    unlike the per-row facts (`now`/`expires_at`/`count`/`last_reescalated_at`)."""
+
+    base_seconds: int
+    max_reescalations: int
+
+
+def reescalation_decision(
+    *,
+    now: datetime,
+    expires_at: datetime,
+    count: int,
+    last_reescalated_at: datetime | None,
+    policy: ReescalationPolicy,
+) -> ReescalationDecision:
+    """Pure per-notification re-escalation backoff decision.
+
+    Schedule: the first re-escalation (``count == 0``) is due at
+    ``expires_at`` itself. Each one after that doubles the wait from
+    ``policy.base_seconds`` (1h, 2h, 4h, 8h, ...) measured from
+    ``last_reescalated_at``, capped at ``REESCALATION_INTERVAL_CAP_SECONDS``.
+    Past ``policy.max_reescalations``, always "capped" — the caller must
+    never act on it again. A legacy row with no backoff state reads as
+    ``count=0``, preserving the original first-fire-at-expiry behaviour.
+    """
+    if count >= policy.max_reescalations:
+        return "capped"
+    if count == 0:
+        due_at = expires_at
+    else:
+        interval = min(
+            policy.base_seconds * (2 ** (count - 1)), REESCALATION_INTERVAL_CAP_SECONDS
+        )
+        due_at = (last_reescalated_at or expires_at) + timedelta(seconds=interval)
+    return "due" if now >= due_at else "wait"

roboco/services/notification_delivery.py

@@ -19,7 +19,7 @@ from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast
 from uuid import UUID
 
 import structlog
-from sqlalchemy import and_, event, select
+from sqlalchemy import CursorResult, and_, event, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from roboco.agents_config import (
@@ -27,9 +27,14 @@ from roboco.agents_config import (
     get_pm_for_agent,
     get_pm_for_team,
 )
+from roboco.config import settings
 from roboco.db.tables import AgentTable, NotificationTable, TaskTable
 from roboco.events import Event, EventType, get_event_bus
-from roboco.foundation.policy.communications import ACK_REQUIRED_BY_TYPE
+from roboco.foundation.policy.communications import (
+    ACK_REQUIRED_BY_TYPE,
+    ReescalationPolicy,
+    reescalation_decision,
+)
 from roboco.models.base import AgentRole, NotificationPriority, NotificationType
 from roboco.services.base import BaseService, NotFoundError
 from roboco.services.notification_dedup import (
@@ -338,20 +343,45 @@ class NotificationDeliveryService(BaseService):
             recipient_count=len(n.to_agents or []),
             ack_count=len(n.acked_by or []),
             expired_at=n.expires_at.isoformat() if n.expires_at else None,
+            reescalation_count=n.reescalation_count,
+        )
+
+    def _log_permanently_unacked(self, n: NotificationTable) -> None:
+        """One-time terminal log: `n` hit the re-escalation cap and will never
+        be re-escalated again (fires exactly once, on the tick of its last
+        permitted re-escalation). Carries both totals so "seen and ignored"
+        (delivered > 0, recipients just never acked) reads distinctly from
+        "route never worked" (delivered == 0 despite every attempt — a
+        broken escalation chain, e.g. no configured up-role)."""
+        self.log.warning(
+            "Notification permanently unacked — re-escalation cap reached",
+            notification_id=str(n.id),
+            type=n.type.value if n.type else None,
+            priority=n.priority.value if n.priority else None,
+            recipient_count=len(n.to_agents or []),
+            ack_count=len(n.acked_by or []),
+            reescalation_count=n.reescalation_count,
+            reescalation_delivered_count=n.reescalation_delivered_count,
         )
 
     async def sweep_expired_notifications(self) -> int:
-        """Re-escalate then log ack-required notifications past `expires_at`.
+        """Re-escalate (per a backoff schedule) then log ack-required
+        notifications past `expires_at`.
 
         `NotificationTable.expires_at` existed but nothing acted on it. This
         sweep surfaces notifications that have become stale. For an
         ack-required row still unacked past the threshold, the recipient's
-        up-role (the PM's PM, or the CEO) is re-notified BEFORE the row is
-        logged as expired — so an inattentive PM can't both miss a blocker
-        and prevent anyone upstream from seeing it. Non-ack-required rows
-        and already-acked rows are not re-escalated. We log rather than
-        auto-cancel because the notification is the record; rewriting
-        status would be ambiguous. Returns the count of stale unacked items.
+        up-role (the PM's PM, or the CEO) is re-notified — but only when
+        `reescalation_decision` says it's due: the first re-escalation fires
+        immediately at expiry, each one after that backs off exponentially
+        (`notification_reescalation_base_seconds`, doubling, capped at 24h),
+        and past `notification_max_reescalations` the row is left alone for
+        good. Without the schedule, a static pile of stale rows re-escalated
+        on every ~1min sweep tick forever. Non-ack-required rows and
+        already-acked rows are never re-escalated. We log rather than
+        auto-cancel because the notification is the record; rewriting status
+        would be ambiguous. Returns the count of stale unacked items (not just
+        the ones actually re-escalated this tick).
         """
         now = datetime.now(UTC)
         result = await self.session.execute(
@@ -374,36 +404,109 @@ class NotificationDeliveryService(BaseService):
             if n.requires_ack and not self._notification_is_fully_acked(n)
         ]
         for n in unacked:
-            await self._re_escalate_unacked(n)
-            self._log_expired_notification(n)
+            await self._maybe_reescalate(n, now)
         return len(unacked)
 
-    async def _re_escalate_unacked(self, n: NotificationTable) -> None:
+    async def _maybe_reescalate(self, n: NotificationTable, now: datetime) -> None:
+        """Re-escalate `n` only when its backoff schedule says it's due.
+
+        `_persist_and_deliver`'s 60s dedup guard does NOT backstop a
+        concurrent double-sweep here: `BLOCKER_ESCALATION` — the type every
+        re-escalation notification is created as — is not in
+        `_LOOP_PRONE_TYPES` (notification_dedup.py), so that guard returns
+        False unconditionally for it. The real guard is
+        `_claim_reescalation_slot`'s compare-and-set: it must succeed BEFORE
+        any delivery is attempted, so two sweep ticks racing the same row
+        can never both deliver.
+        """
+        decision = reescalation_decision(
+            now=now,
+            expires_at=cast("datetime", n.expires_at),
+            count=n.reescalation_count,
+            last_reescalated_at=n.last_reescalated_at,
+            policy=ReescalationPolicy(
+                base_seconds=settings.notification_reescalation_base_seconds,
+                max_reescalations=settings.notification_max_reescalations,
+            ),
+        )
+        if decision != "due":
+            return  # "wait": not due yet; "capped": already logged + done
+        if not await self._claim_reescalation_slot(n, now):
+            return  # another sweep tick already claimed this attempt
+        delivered = await self._re_escalate_unacked(n)
+        n.reescalation_delivered_count += delivered
+        if n.reescalation_count >= settings.notification_max_reescalations:
+            self._log_permanently_unacked(n)
+        else:
+            self._log_expired_notification(n)
+
+    async def _claim_reescalation_slot(
+        self, n: NotificationTable, now: datetime
+    ) -> bool:
+        """Compare-and-set claim on `n`'s attempt slot, BEFORE any delivery.
+
+        A guarded `UPDATE ... WHERE id = :id AND reescalation_count = :n`:
+        Postgres takes a row lock, so a concurrent claim against the same
+        `reescalation_count` value blocks until this one commits, then loses
+        (0 rows matched — the count already moved). Only the winner proceeds
+        to `_re_escalate_unacked`. The slot is consumed (count bumped) even
+        though delivery hasn't happened yet: a transient delivery failure
+        still burns an attempt, which is what stops a permanently-broken
+        escalation chain from looping forever rather than eventually capping.
+        """
+        result = await self.session.execute(
+            update(NotificationTable)
+            .where(
+                NotificationTable.id == n.id,
+                NotificationTable.reescalation_count == n.reescalation_count,
+            )
+            .values(
+                reescalation_count=n.reescalation_count + 1, last_reescalated_at=now
+            )
+            .execution_options(synchronize_session=False)
+        )
+        # UPDATE always yields a CursorResult (has `.rowcount`); `execute`'s
+        # declared return type is the generic `Result` supertype, so peel it.
+        claimed = cast("CursorResult[Any]", result).rowcount == 1
+        if claimed:
+            # Mirror the winning UPDATE onto the in-memory object so the rest
+            # of this tick (and the eventual `reescalation_delivered_count`
+            # flush) sees consistent state. SQLAlchemy will re-flush these
+            # same two columns with the caller's next commit — a harmless
+            # no-op re-write of the value we just committed, not a bug.
+            n.reescalation_count += 1
+            n.last_reescalated_at = now
+        return claimed
+
+    async def _re_escalate_unacked(self, n: NotificationTable) -> int:
         """Re-send an unacked ack-required notification to each non-acking
-        recipient's up-role before expiry. Best-effort: a missing chain,
-        target, or a dedup-suppressed re-fire is logged-and-skipped, never
-        raises — the expiry log still fires. The loop-prone dedup guard in
-        `_persist_and_deliver` caps repeat re-escalations within the 60s
-        window so a tight sweep loop can't flood the upstream role."""
+        recipient's up-role. Best-effort: a missing chain or target is
+        logged-and-skipped, never raises. Returns how many recipients were
+        actually re-notified — used to distinguish a broken escalation chain
+        (0 delivered despite an attempt) from one that works but is ignored."""
         acked = {str(a) for a in (n.acked_by or [])}
+        delivered = 0
         for recipient_id in n.to_agents or []:
             if str(recipient_id) in acked:
                 continue
-            await self._re_escalate_recipient(n, cast("UUID", recipient_id))
+            if await self._re_escalate_recipient(n, cast("UUID", recipient_id)):
+                delivered += 1
+        return delivered
 
     async def _re_escalate_recipient(
         self, n: NotificationTable, recipient_id: UUID
-    ) -> None:
-        """Resolve one recipient's up-role and re-fire the escalation."""
+    ) -> bool:
+        """Resolve one recipient's up-role and re-fire the escalation.
+        Returns True iff it was actually persisted+delivered."""
         recipient = await self._get_agent_by_id(recipient_id)
         if not recipient or not recipient.slug:
-            return
+            return False
         target_slug = get_escalation_target(recipient.slug)
         if not target_slug:
-            return
+            return False
         target = await self._get_agent_by_slug(target_slug)
         if not target:
-            return
+            return False
         notification = NotificationTable(
             type=NotificationType.BLOCKER_ESCALATION,
             priority=NotificationPriority.HIGH,
@@ -422,7 +525,7 @@ class NotificationDeliveryService(BaseService):
             acked_by=[],
         )
         try:
-            await self._persist_and_deliver(notification)
+            return await self._persist_and_deliver(notification)
         except Exception as e:
             self.log.warning(
                 "Re-escalation deliver failed",
@@ -430,6 +533,7 @@ class NotificationDeliveryService(BaseService):
                 target_slug=target_slug,
                 error=str(e),
             )
+            return False
 
     async def get_pending_for_agent(
         self,
@@ -1219,8 +1323,18 @@ class NotificationDeliveryService(BaseService):
         """Find the auditor agent (org-wide; earliest-created if many)."""
         return await get_agent_by_role(self.session, AgentRole.AUDITOR)
 
-    async def _persist_and_deliver(self, notification: NotificationTable) -> None:
-        """Add to session, flush (to get an id), deliver. Caller commits."""
+    async def _persist_and_deliver(self, notification: NotificationTable) -> bool:
+        """Add to session, flush (to get an id), deliver. Caller commits.
+
+        Returns True iff actually persisted+delivered, False if suppressed by
+        the 60s dedup guard below. That guard only ever applies to
+        `_LOOP_PRONE_TYPES` (notification_dedup.py) — BLOCKER_ESCALATION,
+        the type re-escalations use, is NOT one of them, so for that path
+        this always returns True or raises; the real double-delivery guard
+        for re-escalations is the CAS claim in
+        `NotificationDeliveryService._claim_reescalation_slot`, upstream of
+        this call.
+        """
         # Re-fire guard (loop-prone types): this path skips the DB dedup, so
         # apply the same 60s Redis SET-NX window. Fail-open on Redis down.
         # Casts peel the SA UUID column type-leak for the type checker.
@@ -1241,10 +1355,11 @@ class NotificationDeliveryService(BaseService):
                 if notification.related_task_id is not None
                 else None,
             )
-            return
+            return False
         self.session.add(notification)
         await self.session.flush()
         await self.deliver(require_uuid(notification.id))
+        return True
 
     # =========================================================================
     # API-FACING LIST + CRUD (consumed by api/routes/notifications.py)

tests/integration/test_migration_079_notification_backoff.py

@@ -0,0 +1,83 @@
+"""Re-escalation backoff columns on notifications (migration 079).
+
+Migration 079 adds ``notifications.reescalation_count`` /
+``.reescalation_delivered_count`` (integer, not null, default 0) and
+``.last_reescalated_at`` (timestamptz, null). The real upgrade/downgrade
+chain is verified separately against a throwaway Postgres; these assertions
+guard the resulting schema shape and a value round-trip.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import AgentTable, NotificationTable
+from roboco.models import AgentRole, AgentStatus, NotificationPriority, NotificationType
+from roboco.models.base import Team
+from sqlalchemy import select
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_notification(db_session: AsyncSession) -> NotificationTable:
+    sender = AgentTable(
+        id=uuid4(),
+        name="Dev",
+        slug=f"be-dev-{uuid4().hex[:8]}",
+        role=AgentRole.DEVELOPER,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="dev",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(sender)
+    await db_session.flush()
+    notification = NotificationTable(
+        type=NotificationType.BLOCKER_ESCALATION,
+        priority=NotificationPriority.HIGH,
+        from_agent=sender.id,
+        to_agents=[sender.id],
+        subject="stale",
+        body="body",
+        requires_ack=True,
+    )
+    db_session.add(notification)
+    await db_session.flush()
+    return notification
+
+
+@pytest.mark.asyncio
+async def test_reescalation_backoff_columns_default(db_session: AsyncSession) -> None:
+    notification = await _seed_notification(db_session)
+    assert notification.reescalation_count == 0
+    assert notification.reescalation_delivered_count == 0
+    assert notification.last_reescalated_at is None
+
+
+@pytest.mark.asyncio
+async def test_reescalation_backoff_columns_round_trip(
+    db_session: AsyncSession,
+) -> None:
+    notification = await _seed_notification(db_session)
+    stamped_at = datetime.now(UTC)
+    attempts, delivered = 3, 2
+    notification.reescalation_count = attempts
+    notification.reescalation_delivered_count = delivered
+    notification.last_reescalated_at = stamped_at
+    await db_session.flush()
+
+    row = (
+        await db_session.execute(
+            select(NotificationTable).where(NotificationTable.id == notification.id)
+        )
+    ).scalar_one()
+    assert row.reescalation_count == attempts
+    assert row.reescalation_delivered_count == delivered
+    assert row.last_reescalated_at == stamped_at

tests/unit/services/test_notification_delivery.py

@@ -5,6 +5,17 @@ unacked must be re-escalated to the recipient's up-role (the PM's PM or
 the CEO) BEFORE the sweep logs/expiring it — not just logged-and-dropped.
 Combined with H12 (Task 6), an inattentive PM can't both miss a blocker
 and prevent main-pm from seeing it.
+
+Re-escalation backoff: a static pile of stale notifications used to
+re-escalate on *every* sweep tick (~1min) forever. `reescalation_decision`
+(pure, in `foundation/policy/communications.py`) gates each tick behind a
+per-notification exponential schedule + a hard retry cap.
+
+Double-delivery race: `_persist_and_deliver`'s 60s dedup guard is a no-op for
+`BLOCKER_ESCALATION` (not in `_LOOP_PRONE_TYPES`), so it can't backstop two
+concurrent sweep ticks racing the same stale row — a compare-and-set claim
+(`_claim_reescalation_slot`) is the real guard, exercised below by racing two
+service instances against the same row.
 """
 
 from __future__ import annotations
@@ -15,8 +26,14 @@ from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import UUID, uuid4
 
 import pytest
+from roboco.config import settings
+from roboco.foundation.policy.communications import (
+    ReescalationPolicy,
+    reescalation_decision,
+)
 from roboco.models import NotificationPriority, NotificationType
 from roboco.services.notification_delivery import NotificationDeliveryService
+from sqlalchemy import Update
 
 
 def _stale_notification(
@@ -24,7 +41,8 @@ def _stale_notification(
     requires_ack: bool = True,
     acked: bool = False,
     recipient_id: UUID | None = None,
-    from_agent_id: UUID | None = None,
+    reescalation_count: int = 0,
+    last_reescalated_at: datetime | None = None,
 ) -> MagicMock:
     n = MagicMock()
     n.id = uuid4()
@@ -39,8 +57,11 @@ def _stale_notification(
     n.acked_by = [rid] if acked else []
     n.read_by = []
     n.requires_ack = requires_ack
-    n.from_agent = from_agent_id or uuid4()
+    n.from_agent = uuid4()
     n.related_task_id = uuid4()
+    n.reescalation_count = reescalation_count
+    n.reescalation_delivered_count = 0  # no test needs a nonzero starting value
+    n.last_reescalated_at = last_reescalated_at
     return n
 
 
@@ -74,14 +95,37 @@ def _svc_with_agents(
     return svc
 
 
-def _session_returning(notifications: list[MagicMock]) -> MagicMock:
-    """A session whose `execute(...).scalars().all()` returns `notifications`."""
+def _assign_id_on_add(obj: Any) -> None:
+    """`session.add` side effect: a real flush assigns the SQLAlchemy-default
+    id; this mock has no engine to do that, so stand in for it here — without
+    it `require_uuid(notification.id)` in `_persist_and_deliver` always raises
+    on the freshly-built re-escalation row, making every "delivered" outcome
+    in this suite look like a failure."""
+    if getattr(obj, "id", None) is None:
+        obj.id = uuid4()
+
+
+def _session_returning(
+    notifications: list[MagicMock], *, claim_succeeds: bool = True
+) -> MagicMock:
+    """A session whose SELECT (the sweep's stale-notifications query) returns
+    `notifications`; every re-escalation CAS UPDATE (`_claim_reescalation_slot`)
+    reports 1 row affected — the claim wins — unless `claim_succeeds` is False,
+    simulating a concurrent sweep tick that already claimed this row's slot."""
     session = MagicMock()
-    session.add = MagicMock()
+    session.add = MagicMock(side_effect=_assign_id_on_add)
     session.flush = AsyncMock()
-    result = MagicMock()
-    result.scalars.return_value.all.return_value = notifications
-    session.execute = AsyncMock(return_value=result)
+
+    select_result = MagicMock()
+    select_result.scalars.return_value.all.return_value = notifications
+
+    update_result = MagicMock()
+    update_result.rowcount = 1 if claim_succeeds else 0
+
+    async def _execute(statement: Any, *_args: Any, **_kwargs: Any) -> MagicMock:
+        return update_result if isinstance(statement, Update) else select_result
+
+    session.execute = AsyncMock(side_effect=_execute)
     return session
 
 
@@ -119,6 +163,7 @@ async def test_sweep_re_escalates_stale_unacked_ack_required() -> None:
     assert re_escalated.type == NotificationType.BLOCKER_ESCALATION
     assert re_escalated.requires_ack is True
     assert "Re-escalation" in re_escalated.subject
+    assert notif.reescalation_delivered_count == 1  # the attempt was delivered
 
 
 @pytest.mark.asyncio
@@ -182,7 +227,11 @@ async def test_sweep_does_not_re_escalate_non_ack_required() -> None:
 async def test_sweep_skips_re_escalation_when_no_chain_target() -> None:
     """Recipient with no configured escalation target → no re-escalation, but
     the stale unacked count still surfaces (best-effort: missing chain is
-    logged-and-skipped, never raises)."""
+    logged-and-skipped, never raises). The attempt slot is still consumed
+    (reescalation_count bumps) even though nothing was delivered — a broken
+    chain burns attempts rather than looping forever; delivered stays 0,
+    which is exactly the "route never worked" signal `_log_permanently_unacked`
+    now carries."""
     recipient = _agent("ghost-role")
     notif = _stale_notification(
         requires_ack=True, acked=False, recipient_id=recipient.id
@@ -205,12 +254,164 @@ async def test_sweep_skips_re_escalation_when_no_chain_target() -> None:
 
     assert count == 1  # still stale + unacked
     session.add.assert_not_called()
+    assert notif.reescalation_count == 1  # attempt slot consumed regardless
+    assert notif.reescalation_delivered_count == 0  # ...but nothing delivered
 
 
 @pytest.mark.asyncio
-async def test_sweep_dedup_suppresses_repeat_re_escalation() -> None:
-    """A repeat sweep within the dedup window does not re-fire the same
-    re-escalation (loop-prone guard in `_persist_and_deliver`)."""
+async def test_sweep_cas_claim_prevents_double_delivery_race() -> None:
+    """Two service instances (simulating two concurrent sweep ticks) race the
+    same stale row. `_persist_and_deliver`'s 60s dedup guard cannot arbitrate
+    this — BLOCKER_ESCALATION isn't a `_LOOP_PRONE_TYPES` member, so it's a
+    no-op for this path. The CAS claim in `_claim_reescalation_slot` is what
+    actually decides it: exactly one instance wins the guarded UPDATE and
+    delivers; the loser (0 rows updated) skips delivery entirely, without
+    raising."""
+    recipient = _agent("be-pm")
+    target = _agent("main-pm")
+    notif = _stale_notification(
+        requires_ack=True, acked=False, recipient_id=recipient.id
+    )
+
+    winner_session = _session_returning([notif], claim_succeeds=True)
+    loser_session = _session_returning([notif], claim_succeeds=False)
+    winner = _svc_with_agents(
+        winner_session, recipient=recipient, escalation_target=target
+    )
+    loser = _svc_with_agents(
+        loser_session, recipient=recipient, escalation_target=target
+    )
+
+    with (
+        patch(
+            "roboco.services.notification_delivery.all_recipients_recently_notified",
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "roboco.services.notification_delivery.get_escalation_target",
+            return_value="main-pm",
+        ),
+    ):
+        winner_count = await winner.sweep_expired_notifications()
+        loser_count = await loser.sweep_expired_notifications()
+
+    assert winner_count == 1
+    assert loser_count == 1  # still stale + unacked from the loser's own view
+    assert winner_session.add.call_count == 1  # won the claim, delivered
+    loser_session.add.assert_not_called()  # lost the claim, never touched delivery
+
+
+# =============================================================================
+# reescalation_decision — pure schedule math
+# =============================================================================
+
+
+_DEFAULT_POLICY = ReescalationPolicy(base_seconds=3600, max_reescalations=5)
+
+
+def test_reescalation_decision_first_fire_due_at_expiry() -> None:
+    """count=0 (including a legacy row with no backoff state) is due the
+    instant `now` reaches `expires_at` — preserves the original semantics."""
+    expires_at = datetime(2026, 1, 1, tzinfo=UTC)
+    assert (
+        reescalation_decision(
+            now=expires_at,
+            expires_at=expires_at,
+            count=0,
+            last_reescalated_at=None,
+            policy=_DEFAULT_POLICY,
+        )
+        == "due"
+    )
+
+
+def test_reescalation_decision_first_fire_not_due_before_expiry() -> None:
+    expires_at = datetime(2026, 1, 1, tzinfo=UTC)
+    assert (
+        reescalation_decision(
+            now=expires_at - timedelta(seconds=1),
+            expires_at=expires_at,
+            count=0,
+            last_reescalated_at=None,
+            policy=_DEFAULT_POLICY,
+        )
+        == "wait"
+    )
+
+
+def test_reescalation_decision_backoff_doubles() -> None:
+    """count=2 waits 2*base (2h at the default base) from the last fire."""
+    last = datetime(2026, 1, 1, tzinfo=UTC)
+    expires_at = last - timedelta(hours=3)
+    not_yet = reescalation_decision(
+        now=last + timedelta(hours=2) - timedelta(seconds=1),
+        expires_at=expires_at,
+        count=2,
+        last_reescalated_at=last,
+        policy=_DEFAULT_POLICY,
+    )
+    due = reescalation_decision(
+        now=last + timedelta(hours=2),
+        expires_at=expires_at,
+        count=2,
+        last_reescalated_at=last,
+        policy=_DEFAULT_POLICY,
+    )
+    assert not_yet == "wait"
+    assert due == "due"
+
+
+def test_reescalation_decision_interval_capped_at_24h() -> None:
+    """However high `count` climbs (a raised max_reescalations), the wait
+    between attempts never exceeds 24h."""
+    last = datetime(2026, 1, 1, tzinfo=UTC)
+    expires_at = last - timedelta(days=1)
+    policy = ReescalationPolicy(base_seconds=3600, max_reescalations=20)
+    # Uncapped this would be base*2**8 = 256h; capped it's 24h.
+    not_yet = reescalation_decision(
+        now=last + timedelta(hours=24) - timedelta(seconds=1),
+        expires_at=expires_at,
+        count=9,
+        last_reescalated_at=last,
+        policy=policy,
+    )
+    due = reescalation_decision(
+        now=last + timedelta(hours=24),
+        expires_at=expires_at,
+        count=9,
+        last_reescalated_at=last,
+        policy=policy,
+    )
+    assert not_yet == "wait"
+    assert due == "due"
+
+
+def test_reescalation_decision_capped_past_max_regardless_of_timing() -> None:
+    """count >= max_reescalations is always "capped", even if the schedule
+    math would otherwise say a re-escalation is overdue."""
+    last = datetime(2026, 1, 1, tzinfo=UTC)
+    assert (
+        reescalation_decision(
+            now=last + timedelta(days=365),
+            expires_at=last - timedelta(hours=1),
+            count=5,
+            last_reescalated_at=last,
+            policy=_DEFAULT_POLICY,
+        )
+        == "capped"
+    )
+
+
+# =============================================================================
+# sweep_expired_notifications — backoff integration
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_sweep_backoff_does_not_refire_within_the_interval() -> None:
+    """A row re-escalates once, then a same-tick-ish second sweep (interval
+    not elapsed) does not re-escalate again — and its schedule state (count,
+    last_reescalated_at) is stamped on the notification after the first."""
     recipient = _agent("be-pm")
     target = _agent("main-pm")
     notif = _stale_notification(
@@ -223,7 +424,47 @@ async def test_sweep_dedup_suppresses_repeat_re_escalation() -> None:
     with (
         patch(
             "roboco.services.notification_delivery.all_recipients_recently_notified",
-            AsyncMock(return_value=True),
+            AsyncMock(return_value=False),
+        ),
+        patch(
+            "roboco.services.notification_delivery.get_escalation_target",
+            return_value="main-pm",
+        ),
+    ):
+        first = await svc.sweep_expired_notifications()
+        assert notif.reescalation_count == 1
+        assert notif.last_reescalated_at is not None
+        assert session.add.call_count == 1
+
+        second = await svc.sweep_expired_notifications()
+
+    assert first == 1
+    assert second == 1  # still stale + unacked
+    assert session.add.call_count == 1  # no second re-escalation this soon
+
+
+@pytest.mark.asyncio
+async def test_sweep_capped_row_never_re_escalates_again() -> None:
+    """A row already at the retry cap is skipped forever — no re-escalation,
+    no repeat 'permanently unacked' log — but still counts as stale+unacked."""
+    recipient = _agent("be-pm")
+    target = _agent("main-pm")
+    capped_count = settings.notification_max_reescalations
+    notif = _stale_notification(
+        requires_ack=True,
+        acked=False,
+        recipient_id=recipient.id,
+        reescalation_count=capped_count,
+        last_reescalated_at=datetime.now(UTC) - timedelta(days=1),
+    )
+
+    session = _session_returning([notif])
+    svc = _svc_with_agents(session, recipient=recipient, escalation_target=target)
+
+    with (
+        patch(
+            "roboco.services.notification_delivery.all_recipients_recently_notified",
+            AsyncMock(return_value=False),
         ),
         patch(
             "roboco.services.notification_delivery.get_escalation_target",
@@ -232,5 +473,6 @@ async def test_sweep_dedup_suppresses_repeat_re_escalation() -> None:
     ):
         count = await svc.sweep_expired_notifications()
 
-    assert count == 1  # stale + unacked still reported
-    session.add.assert_not_called()  # dedup suppressed the re-escalation row
+    assert count == 1  # still stale + unacked
+    session.add.assert_not_called()
+    assert notif.reescalation_count == capped_count  # untouched — no further attempts