feat: make Updates discovery atomic (Closes #753)

tests/test_my_work.py

@@ -113,6 +113,33 @@ launcher.open().then(result => process.stdout.write(JSON.stringify({{result, res
     }
 
 
+def test_update_triage_launch_applies_atomic_snapshot_before_starting():
+    script = f"""
+const createLauncher = require({json.dumps(str(UPDATE_TRIAGE_LAUNCHER))});
+const events = [];
+const snapshot = {{items:[{{id:1}},{{id:51}}],total:2,complete:true}};
+const launcher = createLauncher({{
+  selectUpdates: () => events.push('selected'),
+  discover: async () => snapshot,
+  applySnapshot: value => events.push(['applied', value.items.map(item => item.id)]),
+  isUpdatesSelected: () => true,
+  hasMore: () => false,
+  hasCheckpoint: () => false,
+  resume: () => {{ throw new Error('must start'); }},
+  start: () => {{ events.push('started'); return true; }},
+  announce: () => {{}},
+}});
+launcher.open().then(result => process.stdout.write(JSON.stringify({{result, events}})));
+"""
+    result = subprocess.run(["node", "-e", script], capture_output=True, text=True)
+
+    assert result.returncode == 0, result.stderr
+    assert json.loads(result.stdout) == {
+        "result": "opened",
+        "events": ["selected", ["applied", [1, 51]], "started"],
+    }
+
+
 def test_notification_pager_load_all_retries_from_failed_page_without_duplicates():
     script = f"""
 const buildMyWork = require({json.dumps(str(MY_WORK))});

tests/test_notification_pages.py

@@ -31,3 +31,55 @@ async def test_notification_page_endpoint_loads_only_requested_page_and_is_not_c
         "has_more": True,
     }
     assert requested == [2]
+
+
+@pytest.mark.anyio
+async def test_notification_snapshot_endpoint_returns_one_complete_atomic_no_store_result(monkeypatch):
+    requested = []
+
+    async def snapshot_loader(*, deadline_seconds):
+        requested.append(deadline_seconds)
+        return {
+            "items": [
+                {"id": 1, "title": "Newest update"},
+                {"id": 51, "title": "Older update"},
+            ],
+            "total": 2,
+            "complete": True,
+        }
+
+    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", snapshot_loader)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get("/api/v1/notifications/snapshot")
+
+    assert response.status_code == 200
+    assert response.headers["cache-control"] == "no-store"
+    assert response.json() == {
+        "items": [
+            {"id": 1, "title": "Newest update"},
+            {"id": 51, "title": "Older update"},
+        ],
+        "total": 2,
+        "complete": True,
+    }
+    assert requested == [main.NOTIFICATION_PAGE_TIMEOUT_SECONDS]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("failure", [TimeoutError(), ValueError("pagination changed")])
+async def test_notification_snapshot_endpoint_is_bounded_and_retryable(monkeypatch, failure):
+    async def snapshot_loader(*, deadline_seconds):
+        raise failure
+
+    monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", snapshot_loader)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get("/api/v1/notifications/snapshot")
+
+    assert response.status_code == 503
+    assert response.headers["cache-control"] == "no-store"
+    assert response.headers["retry-after"] == "1"
+    assert response.json() == {
+        "error": "Complete unread updates are temporarily unavailable. Please retry."
+    }