fix(alerts): a flapping phone still buzzed every cycle — cap pushes per worker

app/main.py

@@ -408,6 +408,12 @@ async def _push_alert(kind: str, subject: str, title: str, message: str, alert_i
         return
     metrics.record_notify_delivery(bool(delivered))
     if not delivered:
+        # The worker push floor is armed at spawn time, before the outcome is
+        # known — a TOTAL delivery failure must release it, or the un-dedupe
+        # retry below would be suppressed for the whole floor window and the
+        # "will retry on the next cycle" promise becomes false.
+        if kind == "worker":
+            _worker_last_offline_push.pop(subject, None)
         try:
             undeduped = await database.delete_alert(alert_id)
         except Exception as exc:  # noqa: BLE001
@@ -977,6 +983,20 @@ async def _run_vacuum() -> None:
 _STREAK_CLEARED = -1
 _worker_online_streak: dict[str, int] = {}
 
+#: Floor between offline PUSHES for one worker, independent of episodes.
+#: Sustained-recovery damping alone was defeated by a phone whose NORMAL duty
+#: cycle is 7-10 minutes awake, 2-3 minutes dozing: every wake window passed
+#: the sustained bar, re-armed the alert, and the next nap pushed — eight
+#: pushes in two hours, live, AFTER the episode damping shipped. A worker
+#: whose ordinary operation is rhythmic flapping can never be rate-limited by
+#: episode detection, so the cap sits on the notification itself: the first
+#: offline push is immediate, repeats for the same worker wait out the floor.
+#: The bell and dashboard stay real-time; only the phone-buzz is capped.
+#: In-memory on purpose — a restart forgets the floor, which at worst costs
+#: one early push.
+_WORKER_OFFLINE_PUSH_COOLDOWN_S = 6 * 3600
+_worker_last_offline_push: dict[str, float] = {}
+
 
 async def _check_stale_workers() -> None:
     """Mark workers as offline if stale, and purge never-enrolled workers offline > 1 hour.
@@ -1078,15 +1098,30 @@ async def _raise_worker_offline_alert(w: dict[str, Any]) -> None:
         "containers keep running and earning, but CashPilot cannot see or manage them until it reconnects."
     )
     if alert_id := await database.record_alert("worker", cid, msg):
-        _spawn(
-            _push_alert(
-                "worker",
-                cid,
-                f"CashPilot: worker '{w['name']}' went offline",
-                msg,
-                alert_id,
+        # The push floor, checked only on a FRESH row (a deduped record never
+        # gets here): a new episode inside the floor keeps its row and bell
+        # entry — the dashboard tells the truth — but does not buzz the phone
+        # again. Deliberately never reset on recovery; resetting there is
+        # exactly the re-arm loop this exists to end.
+        now = time.monotonic()
+        last_push = _worker_last_offline_push.get(cid)
+        if last_push is None or now - last_push >= _WORKER_OFFLINE_PUSH_COOLDOWN_S:
+            _worker_last_offline_push[cid] = now
+            _spawn(
+                _push_alert(
+                    "worker",
+                    cid,
+                    f"CashPilot: worker '{w['name']}' went offline",
+                    msg,
+                    alert_id,
+                )
+            )
+        else:
+            logger.info(
+                "Offline push for worker '%s' suppressed by the per-worker floor (%.0f min since last)",
+                w["name"],
+                (now - last_push) / 60,
             )
-        )
     # Into the bell NOW (the sweep runs every 2 min); the hourly collection
     # rebuild re-derives it from the workers table.
     if not any(a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts):

tests/conftest.py

@@ -134,6 +134,7 @@ def _reset_process_wide_caches():
     for module_name, attr in (
         ("app.main", "_net_baselines"),
         ("app.main", "_worker_online_streak"),
+        ("app.main", "_worker_last_offline_push"),
         ("app.credential_test", "_last_attempt"),
     ):
         with contextlib.suppress(Exception):

tests/test_worker_offline_alerts.py

@@ -75,9 +75,11 @@ def _isolate_bell():
     before = main._collector_alerts
     main._collector_alerts = []
     main._worker_online_streak.clear()
+    main._worker_last_offline_push.clear()
     yield
     main._collector_alerts = before
     main._worker_online_streak.clear()
+    main._worker_last_offline_push.clear()
 
 
 class TestOfflineTransition:
@@ -337,8 +339,12 @@ def test_a_sustained_recovery_rearms_the_next_episode(self):
         self._sweep(rows=[self._fresh_row()])  # online sweep 1 — no clear yet
         second = self._sweep(rows=[self._fresh_row()])  # online sweep 2 — clears
         second.clear.assert_awaited_once_with("worker", "cid-watchtower")
-        # Cleared row -> a NEW offline episode records fresh and pushes again.
+        # Cleared row -> a NEW offline episode records fresh again. The PUSH
+        # additionally obeys the per-worker floor (TestOfflinePushFloor), so
+        # it fires here only once the floor has passed.
+        main._worker_last_offline_push["cid-watchtower"] -= main._WORKER_OFFLINE_PUSH_COOLDOWN_S + 1
         episode2 = self._sweep(rows=[_worker_row()])
+        episode2.record.assert_awaited_once()
         assert len(episode2.sends) == 1
 
     def test_the_streak_resets_on_every_flap(self):
@@ -369,3 +375,59 @@ def test_a_successful_clear_is_not_repeated_while_online(self):
         cleared.clear.assert_awaited_once()
         later = self._sweep(rows=[self._fresh_row()])
         later.clear.assert_not_awaited()
+
+
+class TestOfflinePushFloor:
+    """Episode damping cannot rate-limit a worker whose NORMAL operation is
+    rhythmic flapping — the live OPPO phone stays awake 7-10 minutes between
+    2-3 minute dozes, so every wake window passes the sustained-recovery bar,
+    re-arms the alert, and the next nap pushed again (eight pushes in two
+    hours AFTER episode damping shipped). The floor caps the buzz itself:
+    the first push is immediate, repeats per worker wait out the floor, and
+    the dashboard/bell stay real-time truth."""
+
+    _sweep = TestFlapDamping._sweep
+    _fresh_row = TestFlapDamping._fresh_row
+
+    def _cycle_to_rearmed(self):
+        first = self._sweep(rows=[_worker_row()])  # episode 1: pushes, arms the floor
+        assert len(first.sends) == 1
+        self._sweep(rows=[self._fresh_row()])
+        self._sweep(rows=[self._fresh_row()])  # sustained recovery: cleared + re-armed
+
+    def test_a_new_episode_inside_the_floor_records_but_does_not_push(self):
+        self._cycle_to_rearmed()
+        second = self._sweep(rows=[_worker_row()])
+        second.record.assert_awaited_once()  # the row and bell stay truthful
+        assert second.sends == []  # the phone does not buzz again
+        assert any(a.get("kind") == "worker" for a in main._collector_alerts)
+
+    def test_the_floor_expires_and_the_next_episode_pushes(self):
+        self._cycle_to_rearmed()
+        main._worker_last_offline_push["cid-watchtower"] -= main._WORKER_OFFLINE_PUSH_COOLDOWN_S + 1
+        second = self._sweep(rows=[_worker_row()])
+        assert len(second.sends) == 1
+
+    def test_recovery_never_resets_the_floor(self):
+        # The whole point: recovering is what re-armed the spam loop, so a
+        # recovery must not shorten the floor.
+        self._cycle_to_rearmed()
+        armed_at = main._worker_last_offline_push["cid-watchtower"]
+        self._sweep(rows=[self._fresh_row()])
+        self._sweep(rows=[self._fresh_row()])
+        assert main._worker_last_offline_push["cid-watchtower"] == armed_at
+
+    def test_a_failed_delivery_releases_the_floor(self):
+        """The floor is armed at spawn time, before the outcome is known — a
+        total delivery failure must release it or the un-dedupe retry would be
+        suppressed for the whole floor window."""
+        import time as _time
+
+        main._worker_last_offline_push["cid-watchtower"] = _time.monotonic()
+        with (
+            patch("app.main.notify.send", new_callable=AsyncMock, return_value=0),
+            patch("app.main.notify.is_enabled", return_value=True),
+            patch("app.main.database.delete_alert", new_callable=AsyncMock, return_value=True),
+        ):
+            _run(main._push_alert("worker", "cid-watchtower", "t", "m", 123))
+        assert "cid-watchtower" not in main._worker_last_offline_push