feat: complete unread push snapshots (Closes #573)

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",