feat: send bounded push digests for update bursts (Closes #569)

tests/test_push_notifications.py

@@ -85,6 +85,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
         raise asyncio.CancelledError
 
     monkeypatch.setenv("STACKCHAIN_PUSH_MAX_CONCURRENCY", "3")
+    monkeypatch.setenv("STACKCHAIN_PUSH_MAX_INDIVIDUAL_NOTIFICATIONS", "4")
     monkeypatch.setattr(main.asyncio, "sleep", no_wait)
     monkeypatch.setattr(main, "dispatch_unread_updates", stop_after_capture)
 
@@ -93,6 +94,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
 
     assert captured["session_active"] is dashboard_auth.managed_session_active
     assert captured["max_concurrency"] == 3
+    assert captured["max_individual_notifications"] == 4
 
 
 def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path):
@@ -148,6 +150,108 @@ async def test_dispatch_sends_one_privacy_safe_deep_link_per_new_thread(tmp_path
     assert "Secret title" not in json.dumps(sent)
 
 
+@pytest.mark.anyio
+async def test_update_burst_sends_bounded_individual_pushes_and_one_private_digest(tmp_path):
+    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
+    store.upsert("session-a", {
+        "endpoint": "https://push.example/device-a",
+        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
+    })
+    sent = []
+
+    async def unread():
+        return {"items": [
+            {
+                "id": notification_id,
+                "repository": "private/repo",
+                "title": f"Secret update {notification_id}",
+            }
+            for notification_id in range(41, 46)
+        ]}
+
+    async def send(_subscription, payload):
+        sent.append(json.loads(payload))
+
+    config = PushConfiguration("public", "private", "mailto:ops@example.com")
+    delivered = await dispatch_unread_updates(
+        store,
+        config,
+        unread,
+        send,
+        max_individual_notifications=2,
+    )
+
+    assert delivered == 3
+    assert [payload["tag"] for payload in sent] == [
+        "stackchain-update-41",
+        "stackchain-update-42",
+        "stackchain-update-digest",
+    ]
+    assert sent[-1] == {
+        "title": "3 new work updates",
+        "body": "Tap to review them in Stackchain.",
+        "route": "#/my-work/updates",
+        "tag": "stackchain-update-digest",
+        "update_count": 3,
+    }
+    assert "private/repo" not in json.dumps(sent)
+    assert "Secret update" not in json.dumps(sent)
+    assert await dispatch_unread_updates(
+        store,
+        config,
+        unread,
+        send,
+        max_individual_notifications=2,
+    ) == 0
+
+
+@pytest.mark.anyio
+async def test_failed_digest_retries_only_overflow_after_individual_checkpoints(tmp_path):
+    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
+    store.upsert("session-a", {
+        "endpoint": "https://push.example/device-a",
+        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
+    })
+    attempts = []
+
+    async def unread():
+        return {"items": [{"id": notification_id} for notification_id in range(51, 56)]}
+
+    async def fail_digest(_subscription, payload):
+        decoded = json.loads(payload)
+        attempts.append(decoded["tag"])
+        if decoded["tag"] == "stackchain-update-digest":
+            raise RuntimeError("push provider unavailable")
+
+    config = PushConfiguration("public", "private", "mailto:ops@example.com")
+    assert await dispatch_unread_updates(
+        store,
+        config,
+        unread,
+        fail_digest,
+        max_individual_notifications=2,
+    ) == 2
+    assert store.claim_unseen(range(51, 56))[0].thread_ids == (53, 54, 55)
+
+    async def succeed(_subscription, payload):
+        attempts.append(json.loads(payload)["tag"])
+
+    assert await dispatch_unread_updates(
+        store,
+        config,
+        unread,
+        succeed,
+        max_individual_notifications=2,
+    ) == 1
+    assert attempts == [
+        "stackchain-update-51",
+        "stackchain-update-52",
+        "stackchain-update-digest",
+        "stackchain-update-digest",
+    ]
+    assert store.claim_unseen(range(51, 56)) == []
+
+
 @pytest.mark.anyio
 async def test_production_notification_page_dispatches_one_unread_push(tmp_path):
     def upstream(_request):

tests/test_service_worker.py

@@ -417,6 +417,32 @@ def test_inbound_push_renders_generic_update_and_tap_opens_existing_workflow():
     assert "must-not-render" not in json.dumps(result["notifications"])
 
 
+def test_update_digest_push_opens_unread_inbox_without_item_actions_or_private_copy():
+    result = run_worker_scenario(
+        """
+  await dispatchPush({
+    title:'must-not-render', body:'private details must-not-render',
+    tag:'stackchain-update-digest', route:'#/my-work/updates', update_count:7,
+  });
+  await dispatchNotificationClick('#/my-work/updates');
+  process.stdout.write(JSON.stringify(state));
+"""
+    )
+
+    assert result["notifications"] == [{
+        "title": "7 new work updates",
+        "options": {
+            "body": "Tap to review them in Stackchain.",
+            "tag": "stackchain-update-digest",
+            "data": {"route": "#/my-work/updates"},
+        },
+    }]
+    assert result["opened"] == [
+        "https://forge.example/dashboard/#/my-work/updates"
+    ]
+    assert "must-not-render" not in json.dumps(result["notifications"])
+
+
 def test_push_mark_read_action_confirms_authenticated_mutation_without_opening_app():
     result = run_worker_scenario(
         """