src/gitea_proxy.py

@@ -709,41 +709,6 @@ 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(),
-                gitea_proxy.unread_notification_snapshot,
+                notifications,
                 session_active=dashboard_auth.managed_session_active,
                 lease_seconds=lease_seconds,
                 send_timeout_seconds=send_timeout,
@@ -1735,9 +1735,7 @@ async def subscribe_push(payload: PushSubscriptionPayload, request: Request):
         payload.model_dump(),
     )
     try:
-        current = await gitea_proxy.unread_notification_snapshot(
+        current = await asyncio.wait_for(notifications(), NOTIFICATION_PAGE_TIMEOUT_SECONDS)
-            deadline_seconds=NOTIFICATION_PAGE_TIMEOUT_SECONDS
-        )
     except Exception as error:
         await asyncio.to_thread(_push_subscription_store.delete_session, device_id)
         raise HTTPException(
@@ -1745,13 +1743,13 @@ async def subscribe_push(payload: PushSubscriptionPayload, request: Request):
             detail="Unread updates are temporarily unavailable",
             headers={"Retry-After": "1"},
         ) from error
-    existing_revisions = {
+    existing_ids = {
-        int(item["id"]): str(item.get("updated_at") or "")
+        int(item["id"])
         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_revisions
+        _push_subscription_store.mark_delivered, device_id, existing_ids
     )
     return {"subscribed": True}
 

src/push_notifications.py

@@ -59,14 +59,11 @@ 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,33 +223,6 @@ 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,123 +40,6 @@ 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,67 +84,6 @@ 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 = {
@@ -201,8 +140,7 @@ 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
 
@@ -214,7 +152,6 @@ 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
@@ -701,10 +638,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(**_kwargs):
+    async def no_unread():
-        return {"items": [], "complete": True}
+        return {"items": []}
 
-    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", no_unread)
+    monkeypatch.setattr(main, "notifications", no_unread)
     request = SimpleNamespace(state=SimpleNamespace(
         dashboard_session=SimpleNamespace(session_id="session-a")
     ))
@@ -722,50 +659,6 @@ 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")
@@ -779,11 +672,11 @@ async def test_subscription_baselines_production_notification_ids(tmp_path, monk
     async def management_id(_session):
         return "device-a"
 
-    async def current_unread(**_kwargs):
+    async def current_unread():
-        return {"items": [{"id": 42}], "complete": True}
+        return {"items": [{"id": 42}]}
 
     monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
-    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", current_unread)
+    monkeypatch.setattr(main, "notifications", current_unread)
     request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
     payload = main.PushSubscriptionPayload(
         endpoint="https://push.example/device-a",
@@ -807,11 +700,11 @@ async def test_subscription_fails_closed_when_existing_unread_baseline_is_unavai
     async def management_id(_session):
         return "device-a"
 
-    async def unavailable(**_kwargs):
+    async def unavailable():
         raise RuntimeError("Gitea unavailable")
 
     monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
-    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", unavailable)
+    monkeypatch.setattr(main, "notifications", unavailable)
     request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
     payload = main.PushSubscriptionPayload(
         endpoint="https://push.example/device-a",