Add sync and async export page iteration

README.md

@@ -51,6 +51,29 @@ with ProoflineClient() as client:
     # Continue with cursor=page.next_cursor when it is not None.
 ```
 
+`iter_export_pages` runs that loop for one window and yields every page while it
+follows the signed cursors, keeping the dataset, window, format and grouping fixed.
+It stops at the last page, refuses a cursor it has already received and raises
+`ProoflineError` when `max_pages` (100 by default, at most 1000) is reached.
+`AsyncProoflineClient` exposes the same helper for `async for`.
+
+```python
+import os
+from uuid import UUID
+
+with ProoflineClient() as client:
+    rows = [
+        row
+        for export in client.iter_export_pages(
+            UUID(os.environ["PROOFLINE_ORGANIZATION_ID"]),
+            "findings",
+            from_time="2026-10-01T00:00:00Z",
+            to_time="2026-10-25T00:00:00Z",
+        )
+        for row in export.rows
+    ]
+```
+
 ## Coverage uploads
 
 Use a repository-restricted token with `coverage:write` and current repository write

proofline/client.py

@@ -180,12 +180,34 @@ def _next_cursor(page: ReviewPage, seen: set[str]) -> str | None:
     return cursor
 
 
-def _pagination_limit() -> ProoflineError:
+def _pagination_limit(subject: str = "Review") -> ProoflineError:
     return ProoflineError(
-        0, "urn:proofline:sdk:pagination-limit", "Review pagination reached the configured bound."
+        0,
+        "urn:proofline:sdk:pagination-limit",
+        f"{subject} pagination reached the configured bound.",
     )
 
 
+def _export_limits(max_pages: int) -> None:
+    if not 1 <= max_pages <= 1_000:
+        raise ValueError("Export iteration requires max_pages 1..1000.")
+
+
+def _export_cursor(page: ExportPage, seen: set[str]) -> str | None:
+    """A missing continuation ends the window; a repeated one is refused."""
+    cursor = page.next_cursor
+    if not cursor:
+        return None
+    if cursor in seen:
+        raise ProoflineError(
+            0,
+            "urn:proofline:sdk:invalid-pagination",
+            "The export page has no new continuation cursor.",
+        )
+    seen.add(cursor)
+    return cursor
+
+
 class ProoflineClient:
     """Synchronous typed HTTP operations.
 
@@ -491,6 +513,44 @@ def iter_reviews(
             params["cursor"] = cursor
         raise _pagination_limit()
 
+    def iter_export_pages(
+        self,
+        organization_id: UUID,
+        dataset: ExportDataset,
+        *,
+        from_time: str,
+        to_time: str,
+        format: ExportFormat = "ndjson",
+        group_by: str | None = None,
+        cursor: str | None = None,
+        max_pages: int = 100,
+    ) -> Iterator[ExportPage]:
+        """Read every export page of one window by following signed cursors.
+
+        The window, format and grouping stay fixed across the requests, so a
+        continuation only changes the cursor. Iteration stops at the last page,
+        refuses a repeated cursor and stops at the configured page bound.
+        """
+        _export_limits(max_pages)
+        seen: set[str] = set()
+        if cursor:
+            seen.add(cursor)
+        for _ in range(max_pages):
+            page = self.export_reviews(
+                organization_id,
+                dataset,
+                from_time=from_time,
+                to_time=to_time,
+                format=format,
+                cursor=cursor,
+                group_by=group_by,
+            )
+            yield page
+            cursor = _export_cursor(page, seen)
+            if cursor is None:
+                return
+        raise _pagination_limit("Export")
+
     def wait_for_review(
         self, organization_id: UUID, review_id: UUID, *, timeout: float = 300
     ) -> ReviewOutcome:
@@ -850,6 +910,44 @@ async def iter_reviews(
             params["cursor"] = cursor
         raise _pagination_limit()
 
+    async def iter_export_pages(
+        self,
+        organization_id: UUID,
+        dataset: ExportDataset,
+        *,
+        from_time: str,
+        to_time: str,
+        format: ExportFormat = "ndjson",
+        group_by: str | None = None,
+        cursor: str | None = None,
+        max_pages: int = 100,
+    ) -> AsyncIterator[ExportPage]:
+        """Read every export page of one window by following signed cursors.
+
+        The window, format and grouping stay fixed across the requests, so a
+        continuation only changes the cursor. Iteration stops at the last page,
+        refuses a repeated cursor and stops at the configured page bound.
+        """
+        _export_limits(max_pages)
+        seen: set[str] = set()
+        if cursor:
+            seen.add(cursor)
+        for _ in range(max_pages):
+            page = await self.export_reviews(
+                organization_id,
+                dataset,
+                from_time=from_time,
+                to_time=to_time,
+                format=format,
+                cursor=cursor,
+                group_by=group_by,
+            )
+            yield page
+            cursor = _export_cursor(page, seen)
+            if cursor is None:
+                return
+        raise _pagination_limit("Export")
+
     async def wait_for_review(
         self, organization_id: UUID, review_id: UUID, *, timeout: float = 300
     ) -> ReviewOutcome:

scripts/contracts.py

@@ -29,7 +29,7 @@ def main() -> None:
             "The generated JSON operations must match the twenty-one public JSON token routes."
         )
     tree = ast.parse((ROOT / "proofline/client.py").read_text())
-    helpers = {"close", "aclose", "iter_reviews", "wait_for_review"}
+    helpers = {"close", "aclose", "iter_reviews", "iter_export_pages", "wait_for_review"}
     clients = {
         node.name: {
             member.name

tests/test_http.py

@@ -28,6 +28,7 @@
     CustomerAssessmentDetail,
     CustomerFinding,
     CustomerReviewDetail,
+    ExportPage,
     FileReviewRequest,
     FileReviewResponse,
     PatchReviewRequest,
@@ -95,6 +96,13 @@ async def collect(client: Client, **kwargs: Any) -> list[Any]:
     return list(iterator)
 
 
+async def collect_pages(client: Client, dataset: str, **kwargs: Any) -> list[ExportPage]:
+    iterator = client.iter_export_pages(ORG, dataset, **kwargs)
+    if hasattr(iterator, "__aiter__"):
+        return [export async for export in iterator]
+    return list(iterator)
+
+
 class Clock:
     def __init__(self, monkeypatch: pytest.MonkeyPatch) -> None:
         self.now = 100.0
@@ -1392,6 +1400,86 @@ def respond(request: Request, index: int) -> Reply:
             assert caught.value.status == 404
 
 
+@MODES
+async def test_export_iteration_keeps_the_window_and_refuses_cycles_or_bounds(mode: str) -> None:
+    kwargs = {
+        "from_time": "2026-10-01T00:00:00Z",
+        "to_time": "2026-10-25T00:00:00Z",
+        "format": "csv",
+        "group_by": "severity",
+    }
+
+    def csv_reply(cursor: str | None) -> Reply:
+        headers = {
+            "Content-Type": "text/csv",
+            "X-Proofline-Export-Version": "1",
+            "X-Proofline-Export-Settled-Until": "2026-10-25T00:00:00Z",
+        }
+        if cursor is not None:
+            headers["X-Proofline-Export-Next-Cursor"] = cursor
+        return Reply(body=b"dataset_version,severity\r\n1,high\r\n", headers=headers)
+
+    def respond(request: Request, index: int) -> Reply:
+        query = parse_qs(urlsplit(request.path).query)
+        assert request.path.startswith(f"/api/v1/organizations/{ORG}/exports/finding_outcomes?")
+        assert query["from"] == ["2026-10-01T00:00:00Z"]
+        assert query["to"] == ["2026-10-25T00:00:00Z"]
+        assert query["format"] == ["csv"]
+        assert query["group_by"] == ["severity"]
+        assert query.get("cursor") == ([f"cursor-{index - 1}"] if index else None)
+        headers = {
+            "Content-Type": "text/csv",
+            "X-Proofline-Export-Version": "1",
+            "X-Proofline-Export-Settled-Until": "2026-10-25T00:00:00Z",
+        }
+        if index < 2:
+            headers["X-Proofline-Export-Next-Cursor"] = f"cursor-{index}"
+        return Reply(
+            body=f"dataset_version,severity\r\n1,page-{index}\r\n".encode(), headers=headers
+        )
+
+    with Server(respond) as server:
+        async with session(mode, server.url) as client:
+            exports = await collect_pages(client, "finding_outcomes", **kwargs)
+            assert [export.rows[0]["severity"] for export in exports] == [
+                "page-0",
+                "page-1",
+                "page-2",
+            ]
+            assert exports[0].text.startswith("dataset_version,severity\r\n")
+            assert [export.next_cursor for export in exports] == ["cursor-0", "cursor-1", None]
+            assert exports[-1].settled_until == "2026-10-25T00:00:00Z"
+        assert len(server.requests) == 3
+
+    with Server(lambda request, index: csv_reply("repeated")) as server:
+        async with session(mode, server.url) as client:
+            with pytest.raises(ProoflineError) as raised:
+                await collect_pages(client, "finding_outcomes", **kwargs)
+            assert raised.value.type == "urn:proofline:sdk:invalid-pagination"
+        assert len(server.requests) == 2
+
+    with Server(lambda request, index: csv_reply("seeded")) as server:
+        async with session(mode, server.url) as client:
+            with pytest.raises(ProoflineError) as raised:
+                await collect_pages(client, "finding_outcomes", cursor="seeded", **kwargs)
+            assert raised.value.type == "urn:proofline:sdk:invalid-pagination"
+        assert len(server.requests) == 1
+
+    with Server(lambda request, index: csv_reply(f"cursor-{index}")) as server:
+        async with session(mode, server.url) as client:
+            with pytest.raises(ProoflineError) as raised:
+                await collect_pages(client, "finding_outcomes", max_pages=2, **kwargs)
+            assert raised.value.detail == "Export pagination reached the configured bound."
+        assert len(server.requests) == 2
+
+    with Server(lambda request, index: csv_reply(None)) as server:
+        async with session(mode, server.url) as client:
+            for bound in (0, 1_001):
+                with pytest.raises(ValueError):
+                    await collect_pages(client, "finding_outcomes", max_pages=bound, **kwargs)
+        assert server.requests == []
+
+
 @MODES
 async def test_knowledge_negotiation_scope_and_no_retry(mode: str) -> None:
     refusal = False