src/gitea_proxy.py

@@ -709,6 +709,41 @@ async def notifications() -> dict:
     return await notification_page(1)
 
 
+async def unread_notification_snapshot(
+    *,
+    limit: int = 50,
+    max_pages: int = 20,
+    max_concurrency: int = 4,
+    deadline_seconds: float = 5.0,
+) -> dict:
+    """Load one complete, bounded snapshot of unread notification threads."""
+    async with asyncio.timeout(deadline_seconds):
+        first = await notification_page(1, limit)
+        page_count = max(1, (first["total"] + limit - 1) // limit)
+        if page_count > max_pages:
+            raise ValueError("Unread notification snapshot exceeds the scan limit")
+        semaphore = asyncio.Semaphore(max(1, max_concurrency))
+
+        async def load(page: int) -> dict:
+            async with semaphore:
+                return await notification_page(page, limit)
+
+        remaining = await asyncio.gather(
+            *(load(page) for page in range(2, page_count + 1))
+        )
+    pages = [first, *remaining]
+    if any(page["total"] != first["total"] for page in pages[1:]):
+        raise ValueError("Unread notification pagination changed during the scan")
+    items = [item for page in pages for item in page["items"]]
+    thread_ids = [
+        item.get("id") for item in items
+        if isinstance(item, dict) and isinstance(item.get("id"), int) and item["id"] > 0
+    ]
+    if len(items) != first["total"] or len(set(thread_ids)) != first["total"]:
+        raise ValueError("Unread notification snapshot has an incomplete thread set")
+    return {"items": items, "total": first["total"], "complete": True}
+
+
 async def notification_page(page: int, limit: int = 50) -> dict:
     response = await _get_client().get(
         f"/api/v1/notifications?status-types=unread&limit={limit}&page={page}",

src/main.py

@@ -100,7 +100,7 @@ async def _push_poll_loop() -> None:
             await dispatch_unread_updates(
                 _push_subscription_store,
                 _push_configuration(),
-                notifications,
+                gitea_proxy.unread_notification_snapshot,
                 session_active=dashboard_auth.managed_session_active,
                 lease_seconds=lease_seconds,
                 send_timeout_seconds=send_timeout,
@@ -1735,7 +1735,9 @@ async def subscribe_push(payload: PushSubscriptionPayload, request: Request):
         payload.model_dump(),
     )
     try:
-        current = await asyncio.wait_for(notifications(), NOTIFICATION_PAGE_TIMEOUT_SECONDS)
+        current = await gitea_proxy.unread_notification_snapshot(
+            deadline_seconds=NOTIFICATION_PAGE_TIMEOUT_SECONDS
+        )
     except Exception as error:
         await asyncio.to_thread(_push_subscription_store.delete_session, device_id)
         raise HTTPException(
@@ -1743,13 +1745,13 @@ async def subscribe_push(payload: PushSubscriptionPayload, request: Request):
             detail="Unread updates are temporarily unavailable",
             headers={"Retry-After": "1"},
         ) from error
-    existing_ids = {
-        int(item["id"])
+    existing_revisions = {
+        int(item["id"]): str(item.get("updated_at") or "")
         for item in current.get("items", [])
         if isinstance(item, dict) and str(item.get("id", "")).isdigit()
     }
     await asyncio.to_thread(
-        _push_subscription_store.mark_delivered, device_id, existing_ids
+        _push_subscription_store.mark_delivered, device_id, existing_revisions
     )
     return {"subscribed": True}
 

src/push_notifications.py

@@ -59,11 +59,14 @@ async def dispatch_unread_updates(
         return 0
     try:
         page = await unread()
+        if page.get("complete") is False:
+            return 0
         thread_revisions = {
             int(item["id"]): str(item.get("updated_at") or "")
             for item in page.get("items", [])
             if isinstance(item, dict) and str(item.get("id", "")).isdigit()
         }
+        await asyncio.to_thread(store.reconcile_unread, thread_revisions)
         deliveries = await asyncio.to_thread(store.claim_unseen, thread_revisions)
         if session_active is not None:
             try:

src/push_subscription_store.py

@@ -223,6 +223,33 @@ class PushSubscriptionStore:
                     )
             return deliveries
 
+    def reconcile_unread(self, thread_ids: Iterable[int]) -> None:
+        """Prune per-device checkpoints that are absent from a complete snapshot."""
+        unread_ids = tuple(sorted({int(value) for value in thread_ids if int(value) > 0}))
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            connection.execute(
+                "CREATE TEMP TABLE current_unread(thread_id INTEGER PRIMARY KEY)"
+            )
+            connection.executemany(
+                "INSERT INTO current_unread(thread_id) VALUES (?)",
+                ((thread_id,) for thread_id in unread_ids),
+            )
+            connection.execute(
+                """DELETE FROM push_deliveries
+                   WHERE NOT EXISTS (
+                       SELECT 1 FROM current_unread
+                       WHERE current_unread.thread_id = push_deliveries.thread_id
+                   )"""
+            )
+            connection.execute(
+                """DELETE FROM push_digest_pending
+                   WHERE NOT EXISTS (
+                       SELECT 1 FROM current_unread
+                       WHERE current_unread.thread_id = push_digest_pending.thread_id
+                   )"""
+            )
+
     def mark_digest_pending(
         self,
         session_id: str,

tests/test_gitea_notifications.py

@@ -40,6 +40,123 @@ async def test_notification_page_preserves_upstream_total_without_loading_other_
     assert [item["id"] for item in result["items"]] == [42]
 
 
+@pytest.mark.anyio
+async def test_unread_notification_snapshot_loads_every_page_with_bounded_concurrency():
+    active = 0
+    peak = 0
+    requested_pages = []
+
+    async def upstream(request):
+        nonlocal active, peak
+        page = int(request.url.params["page"])
+        requested_pages.append(page)
+        active += 1
+        peak = max(peak, active)
+        await __import__("asyncio").sleep(0)
+        active -= 1
+        start = (page - 1) * 50 + 1
+        end = min(start + 50, 126)
+        return httpx.Response(
+            200,
+            headers={"X-Total-Count": "125"},
+            json=[{
+                "id": thread_id,
+                "updated_at": f"2026-08-11T12:{thread_id % 60:02d}:00Z",
+                "subject": {},
+                "repository": {},
+            } for thread_id in range(start, end)],
+        )
+
+    gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
+    try:
+        result = await gitea_proxy.unread_notification_snapshot(
+            max_pages=3, max_concurrency=2, deadline_seconds=1
+        )
+    finally:
+        await gitea_proxy.stop_client()
+
+    assert requested_pages[0] == 1
+    assert sorted(requested_pages) == [1, 2, 3]
+    assert peak == 2
+    assert result["complete"] is True
+    assert result["total"] == 125
+    assert [item["id"] for item in result["items"]] == list(range(1, 126))
+
+
+@pytest.mark.anyio
+async def test_unread_notification_snapshot_rejects_pagination_that_changes_mid_scan():
+    def upstream(request):
+        page = int(request.url.params["page"])
+        total = 51 if page == 1 else 50
+        start = (page - 1) * 50 + 1
+        return httpx.Response(
+            200,
+            headers={"X-Total-Count": str(total)},
+            json=[{
+                "id": thread_id,
+                "subject": {},
+                "repository": {},
+            } for thread_id in range(start, min(start + 50, total + 1))],
+        )
+
+    gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
+    try:
+        with pytest.raises(ValueError, match="changed during the scan"):
+            await gitea_proxy.unread_notification_snapshot(deadline_seconds=1)
+    finally:
+        await gitea_proxy.stop_client()
+
+
+@pytest.mark.anyio
+async def test_unread_notification_snapshot_rejects_missing_or_duplicate_threads():
+    def upstream(_request):
+        return httpx.Response(
+            200,
+            headers={"X-Total-Count": "3"},
+            json=[
+                {"id": 7, "subject": {}, "repository": {}},
+                {"id": 7, "subject": {}, "repository": {}},
+                "malformed",
+            ],
+        )
+
+    gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
+    try:
+        with pytest.raises(ValueError, match="incomplete thread set"):
+            await gitea_proxy.unread_notification_snapshot(deadline_seconds=1)
+    finally:
+        await gitea_proxy.stop_client()
+
+
+@pytest.mark.anyio
+async def test_unread_notification_snapshot_fails_before_exceeding_the_scan_limit():
+    requested_pages = []
+
+    def upstream(request):
+        page = int(request.url.params["page"])
+        requested_pages.append(page)
+        start = (page - 1) * 50 + 1
+        return httpx.Response(
+            200,
+            headers={"X-Total-Count": "101"},
+            json=[
+                {"id": thread_id, "subject": {}, "repository": {}}
+                for thread_id in range(start, min(start + 50, 102))
+            ],
+        )
+
+    gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
+    try:
+        with pytest.raises(ValueError, match="scan limit"):
+            await gitea_proxy.unread_notification_snapshot(
+                max_pages=2, deadline_seconds=1
+            )
+    finally:
+        await gitea_proxy.stop_client()
+
+    assert requested_pages == [1]
+
+
 @pytest.mark.anyio
 async def test_unread_notifications_are_normalized_for_mobile_handoff():
     result = gitea_proxy._normalize_notifications(

tests/test_push_notifications.py

@@ -84,6 +84,67 @@ def test_subscription_store_does_not_realert_for_an_older_snapshot_revision(tmp_
     assert store.claim_unseen({42: "2026-08-11T12:00:00Z"}) == []
 
 
+def test_complete_unread_snapshot_reconciles_delivery_and_digest_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"},
+    })
+    revisions = {1: "r1", 2: "r2", 3: "r3"}
+    store.mark_delivered("session-a", revisions)
+    store.mark_digest_pending("session-a", {3: "r3"})
+
+    store.reconcile_unread({2})
+    delivery = store.claim_unseen(revisions)[0]
+
+    assert delivery.thread_ids == (1, 3)
+    assert delivery.digest_ids == ()
+
+
+@pytest.mark.anyio
+async def test_dispatch_reconciles_stale_checkpoints_after_a_complete_snapshot(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"},
+    })
+    store.mark_delivered("session-a", {1: "r1"})
+
+    async def unread():
+        return {"items": [{"id": 2, "updated_at": "r2"}], "complete": True}
+
+    async def send(_subscription, _payload):
+        return None
+
+    config = PushConfiguration("public", "private", "mailto:ops@example.com")
+    assert await dispatch_unread_updates(store, config, unread, send) == 1
+
+    remaining = store.claim_unseen({1: "r1", 2: "r2"})
+    assert remaining[0].thread_ids == (1,)
+
+
+@pytest.mark.anyio
+async def test_dispatch_sends_nothing_and_preserves_checkpoints_for_an_incomplete_snapshot(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"},
+    })
+    store.mark_delivered("session-a", {1: "r1"})
+    sent = []
+
+    async def unread():
+        return {"items": [{"id": 2, "updated_at": "r2"}], "complete": False}
+
+    async def send(_subscription, payload):
+        sent.append(payload)
+
+    config = PushConfiguration("public", "private", "mailto:ops@example.com")
+    assert await dispatch_unread_updates(store, config, unread, send) == 0
+    assert sent == []
+    assert store.claim_unseen({1: "r1"}) == []
+
+
 def test_subscription_store_migrates_legacy_delivery_without_replaying_it(tmp_path):
     path = tmp_path / "push.sqlite3"
     subscription = {
@@ -140,7 +201,8 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
     async def no_wait(_seconds):
         return None
 
-    async def stop_after_capture(*_args, **kwargs):
+    async def stop_after_capture(*args, **kwargs):
+        captured["unread"] = args[2]
         captured.update(kwargs)
         raise asyncio.CancelledError
 
@@ -152,6 +214,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
     with pytest.raises(asyncio.CancelledError):
         await main._push_poll_loop()
 
+    assert captured["unread"] is main.gitea_proxy.unread_notification_snapshot
     assert captured["session_active"] is dashboard_auth.managed_session_active
     assert captured["max_concurrency"] == 3
     assert captured["max_individual_notifications"] == 4
@@ -638,10 +701,10 @@ async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(
 
     monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
 
-    async def no_unread():
-        return {"items": []}
+    async def no_unread(**_kwargs):
+        return {"items": [], "complete": True}
 
-    monkeypatch.setattr(main, "notifications", no_unread)
+    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", no_unread)
     request = SimpleNamespace(state=SimpleNamespace(
         dashboard_session=SimpleNamespace(session_id="session-a")
     ))
@@ -659,6 +722,50 @@ async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(
     assert (await main.push_status(request))["subscribed"] is False
 
 
+@pytest.mark.anyio
+async def test_subscription_baselines_every_revision_from_the_complete_snapshot(tmp_path, monkeypatch):
+    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
+    monkeypatch.setattr(main, "_push_subscription_store", store)
+    monkeypatch.setattr(
+        main,
+        "_push_configuration",
+        lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
+    )
+
+    async def management_id(_session):
+        return "device-a"
+
+    revisions = {
+        thread_id: f"2026-08-11T12:{thread_id % 60:02d}:00Z"
+        for thread_id in range(1, 126)
+    }
+
+    async def complete_snapshot(**_kwargs):
+        return {
+            "items": [
+                {"id": thread_id, "updated_at": revision}
+                for thread_id, revision in revisions.items()
+            ],
+            "total": 125,
+            "complete": True,
+        }
+
+    async def first_page_must_not_be_used():
+        raise AssertionError("subscription used the first-page inbox endpoint")
+
+    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
+    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", complete_snapshot)
+    monkeypatch.setattr(main, "notifications", first_page_must_not_be_used)
+    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
+    payload = main.PushSubscriptionPayload(
+        endpoint="https://push.example/device-a",
+        keys={"p256dh": "public-key", "auth": "auth-secret"},
+    )
+
+    assert await main.subscribe_push(payload, request) == {"subscribed": True}
+    assert store.claim_unseen(revisions) == []
+
+
 @pytest.mark.anyio
 async def test_subscription_baselines_production_notification_ids(tmp_path, monkeypatch):
     store = PushSubscriptionStore(tmp_path / "push.sqlite3")
@@ -672,11 +779,11 @@ async def test_subscription_baselines_production_notification_ids(tmp_path, monk
     async def management_id(_session):
         return "device-a"
 
-    async def current_unread():
-        return {"items": [{"id": 42}]}
+    async def current_unread(**_kwargs):
+        return {"items": [{"id": 42}], "complete": True}
 
     monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
-    monkeypatch.setattr(main, "notifications", current_unread)
+    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", current_unread)
     request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
     payload = main.PushSubscriptionPayload(
         endpoint="https://push.example/device-a",
@@ -700,11 +807,11 @@ async def test_subscription_fails_closed_when_existing_unread_baseline_is_unavai
     async def management_id(_session):
         return "device-a"
 
-    async def unavailable():
+    async def unavailable(**_kwargs):
         raise RuntimeError("Gitea unavailable")
 
     monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
-    monkeypatch.setattr(main, "notifications", unavailable)
+    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", unavailable)
     request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
     payload = main.PushSubscriptionPayload(
         endpoint="https://push.example/device-a",