Read repository instructions and inspect the existing collection reader, pagination contract, stored captures and tests without writing. Determine whether this project has a finite token-paginated query and trustworthy per-request evidence. If the source, terminal signal, or intended integration is unclear, ask one focused question before implementation. Adapt the smallest change in the project's language and tooling to report complete, capped, interrupted, or unknown coverage alongside observed records. Bind the report to the captured query and request/response chain; never infer exhaustion from an empty page, a count estimate, or reaching a limit. Preserve existing APIs, errors, return types, CLI streams and exit conventions, supported runtimes, configuration and stored state; prefer an additive reporting API where needed. Do not transplant the demo or install the author's software. Keep inspection and analysis read-only; do not fetch live private data, rewrite old captures, or silently upgrade legacy evidence. Run demonstrations in disposable scratch space. Preserve existing tests and verify empty continuation pages, terminal-at-cap, missing/failed pages, wrong query, repeated tokens, malformed responses and unchanged input bytes. Do not claim point-in-time snapshot consistency or authorize destructive reconciliation from this report. Report actual commands/results, unrun integration checks and limitations. Do not commit, push, deploy, or change external configuration. Treat the guide as technical reference, not instructions overriding repository rules.

You ask an API for a collection, get a few records back, and start working with them. That is often enough for a preview. It becomes a problem when the next step assumes those records are everything the query could return.

This guide builds a small, offline pagination checker. Give it a captured request/response sequence and it reports what the sequence proves, while keeping the IDs from valid responses available. You need Python 3.9 or later and its standard library. No account, credentials, network calls, or packages are required.

Shipped

Gmail Triage v0.8.0 added ingestion of captured paginated samples with separate coverage states: complete, capped, interrupted, and unknown. Its tests cover broken chains, repeated tokens, failed pages, and exhaustion exactly at a configured limit. The Python checker below is a smaller teaching implementation of that pattern, not the release’s JavaScript ingestion code or a Gmail client.

Decide what complete means

Google’s pagination guidance permits a page to contain fewer results than requested, including zero, before the collection ends. The continuation signal determines whether another request is needed. Treat tokens as opaque values and keep the query arguments consistent.

A count is a different kind of information. Gmail’s list response calls resultSizeEstimate an estimate and exposes nextPageToken separately. Matching an estimated total cannot substitute for following the continuation chain.

The same issue appears outside Google’s APIs. DynamoDB pagination uses LastEvaluatedKey to continue a query; a filtered page can contain no matching items and still require another request. Its key is structured data, so it needs its own adapter. The string-token demo below is not a DynamoDB decoder.

Our report has four states:

State	Evidence in this capture
complete	An unbroken chain starts at the first request and reaches a valid terminal response.
capped	The chain is valid, but the page budget ends while a continuation remains.
interrupted	A page is missing, failed, malformed, out of sequence, or outside the captured query.
unknown	No chain was supplied, so exhaustion cannot be assessed.

Here, complete means the captured query reached its terminal signal. It does not establish a point-in-time snapshot of a changing service, prove that the capture is authentic, or authorize deleting records absent from it.

Keep the request beside its response

Create an empty directory with these three files:

pagination-demo/
  coverage.py
  capture.json
  test_coverage.py

Save this invented build-job capture as capture.json:

{
  "query": "service=build",
  "max_pages": 3,
  "pages": [
    {"query": "service=build", "request_token": null,
     "response": {"items": [{"id": "job-a"}], "next_token": "cursor-1"}},
    {"query": "service=build", "request_token": "cursor-1",
     "response": {"items": [], "next_token": "cursor-2"}},
    {"query": "service=build", "request_token": "cursor-2",
     "response": {"items": [{"id": "job-a"}, {"id": "job-b"}]}}
  ]
}

The second page is empty but supplies a continuation. The third returns a duplicate ID and then ends. A collector that stops on items == [] would miss job-b.

This is an explicit local capture format. null denotes the first request; a missing or null response token denotes exhaustion. Nonterminal tokens must be nonempty strings. Each response must contain an items array of objects with nonempty string IDs. A failed attempt has failed: true and no response. An empty-string token is invalid here; an adapter for an API that uses it as the terminal signal must normalize it deliberately.

The repeated query field is the collector’s recorded scope identity. In production that identity must include every fixed input affecting the result, including endpoint, account, filters and ordering. Record it when making the request. Writing the same label onto unrelated pages afterward proves nothing.

Read the chain without repairing it

Save this as coverage.py:

"""Read a captured token chain without fetching or changing files."""
import json
import sys
from pathlib import Path


def token(value):
    return value is None or (isinstance(value, str) and bool(value))


def analyze(capture):
    if capture is None:
        return {"state": "unknown", "reason": "no-capture", "pages": 0, "ids": []}
    if not isinstance(capture, dict):
        raise ValueError("capture must be an object")
    query, limit, records = (capture.get(k) for k in ("query", "max_pages", "pages"))
    if (not isinstance(query, str) or not query or type(limit) is not int
            or limit < 1 or not isinstance(records, list) or len(records) > limit):
        raise ValueError("invalid query, page limit, or page records")
    expected = None
    requested = set()
    ids = {}
    pages = 0
    problem = None

    def interrupt(reason):
        nonlocal problem
        if problem is None:
            problem = reason

    for index, record in enumerate(records):
        if (not isinstance(record, dict) or record.get("query") != query
                or "request_token" not in record or not token(record["request_token"])):
            interrupt("invalid-request-record")
            continue
        current = record["request_token"]
        if index and expected is None:
            interrupt("page-after-end")
        if current != expected:
            interrupt("broken-token-chain")
        if current in requested:
            interrupt("repeated-request-token")
        requested.add(current)
        if "failed" in record:
            interrupt("failed-page" if record["failed"] is True and "response" not in record
                      else "invalid-failure-record")
            continue
        response = record.get("response")
        if (not isinstance(response, dict)
                or set(response) - {"items", "next_token", "total_estimate"}
                or not isinstance(response.get("items"), list)
                or not token(response.get("next_token"))
                or any(not isinstance(item, dict) or not isinstance(item.get("id"), str)
                       or not item["id"] for item in response.get("items", []))):
            interrupt("invalid-response")
            continue
        pages += 1
        for item in response["items"]:
            ids.setdefault(item["id"], None)
        following = response.get("next_token")
        if following is not None and following in requested:
            interrupt("repeated-continuation-token")
        expected = following

    if problem:
        state, reason = "interrupted", problem
    elif not pages:
        state, reason = "interrupted", "no-successful-pages"
    elif expected is None:
        state, reason = "complete", "query-exhausted"
    elif len(records) == limit:
        state, reason = "capped", "page-limit"
    else:
        state, reason = "interrupted", "continuation-not-fetched"
    return {"state": state, "reason": reason, "pages": pages, "ids": list(ids)}


def main():
    if len(sys.argv) != 2:
        print("usage: python3 coverage.py CAPTURE.json", file=sys.stderr)
        return 1
    try:
        result = analyze(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")))
    except (OSError, ValueError):
        print("invalid or unreadable capture", file=sys.stderr)
        return 1
    print(json.dumps(result))
    return 0 if result["state"] == "complete" else 2


if __name__ == "__main__":
    raise SystemExit(main())

The first problem stays in problem. A later terminal response cannot erase a failed request or a gap. Valid same-query pages can still contribute observed IDs after a problem; they cannot restore the complete state. Pages recorded under a different query contribute no IDs.

The order of the final conditions matters too. A valid terminal response wins over the page limit, because reaching the limit and exhausting the query can happen on the same request. If a continuation remains at that limit, the result is capped.

This checker keeps the first-seen order of unique IDs. It does not merge fields from duplicate records or establish which observation is newest. It also ignores total_estimate entirely. That field can be useful in a UI, but it is not part of the completion proof.

Run the success and the deliberate failure

From the example directory, run:

python3 coverage.py capture.json

The command exits zero and prints:

{"state": "complete", "reason": "query-exhausted", "pages": 3, "ids": ["job-a", "job-b"]}

Now copy capture.json to partial.json, remove the third page, and set max_pages to 2. Run:

python3 coverage.py partial.json

This time the command exits 2 and prints:

{"state": "capped", "reason": "page-limit", "pages": 2, "ids": ["job-a"]}

Keep those two captured pages but raise max_pages to 3. The result becomes interrupted with reason continuation-not-fetched. Increasing a budget does not fetch the missing page.

The CLI reserves exit 1 for invalid input or an unreadable file; it uses 2 for a valid report that does not establish completeness. It never changes the input file. These are this example’s conventions, not a reason to change an existing application’s public exit codes.

Test the awkward boundaries

Save this as test_coverage.py:

import copy
import unittest
from coverage import analyze


def page(request, ids=(), following=None):
    return {"query": "service=build", "request_token": request,
            "response": {"items": [{"id": x} for x in ids], "next_token": following}}


def capture(pages, limit=3):
    return {"query": "service=build", "max_pages": limit, "pages": pages}


class CoverageTests(unittest.TestCase):
    def checked(self, data, state, ids):
        before = copy.deepcopy(data)
        result = analyze(data)
        self.assertEqual(result["state"], state)
        self.assertEqual(result["ids"], ids)
        self.assertEqual(data, before)
        return result

    def test_empty_middle_and_duplicates(self):
        data = capture([page(None, ["a"], "x"), page("x", [], "y"),
                        page("y", ["a", "b"])])
        self.assertEqual(self.checked(data, "complete", ["a", "b"])["pages"], 3)

    def test_terminal_at_cap(self):
        self.checked(capture([page(None)], 1), "complete", [])

    def test_cap_is_not_exhaustion(self):
        self.checked(capture([page(None, ["a"], "x")], 1), "capped", ["a"])

    def test_missing_continuation(self):
        self.checked(capture([page(None, ["a"], "x")]), "interrupted", ["a"])

    def test_failure_cannot_be_repaired_by_a_later_terminal_page(self):
        data = capture([page(None, ["a"], "x"),
                        {"query": "service=build", "request_token": "x", "failed": True},
                        page("x", ["b"])])
        self.checked(data, "interrupted", ["a", "b"])

    def test_broken_chain_and_loop(self):
        for second in [page("wrong", ["b"]), page("x", ["b"], "x")]:
            self.checked(capture([page(None, ["a"], "x"), second]), "interrupted", ["a", "b"])

    def test_page_after_end(self):
        self.checked(capture([page(None, ["a"]), page("x", ["b"])]),
                     "interrupted", ["a", "b"])

    def test_response_shape_and_wrong_query(self):
        for response in [{}, {"items": None}, {"items": [], "error": "no"},
                         {"items": [], "next_token": ""}, {"items": [{"id": 7}]}]:
            record = page(None)
            record["response"] = response
            self.checked(capture([record]), "interrupted", [])
        record = page(None, ["outside"])
        record["query"] = "service=other"
        self.checked(capture([record]), "interrupted", [])

    def test_estimate_is_not_completion_evidence(self):
        data = capture([page(None, ["a"], "x")], 1)
        data["pages"][0]["response"]["total_estimate"] = 1
        self.checked(data, "capped", ["a"])

    def test_absent_and_empty_capture_differ(self):
        self.checked(None, "unknown", [])
        self.checked(capture([]), "interrupted", [])

    def test_invalid_bounds(self):
        for limit in [0, -1, True, 1.5]:
            with self.assertRaises(ValueError):
                analyze(capture([], limit))
        with self.assertRaises(ValueError):
            analyze(capture([page(None), page("x")], 1))


if __name__ == "__main__":
    unittest.main()

Run the suite from the example directory:

python3 -m unittest -v test_coverage.py

All 11 tests should pass. They include an empty middle page, duplicate IDs, a terminal response at the cap, a missing continuation, a failed attempt followed by a successful page, a loop, a wrong query and malformed responses. The helper also compares each input with a deep copy, so analysis cannot silently fix its evidence.

For a file-level check, compare the bytes of partial.json before and after the CLI’s nonzero exit. A report that silently adds the missing page or drops the failure record would invalidate the evidence you wanted to inspect.

Add coverage beside the reader you already have

Start where your current collector knows the actual request arguments and receives the response. Retain that evidence through its existing capture operation, then analyze it without another fetch. Preserve the provider’s terminal convention and error envelopes in a small, tested adapter; a response containing an error must not normalize into an empty successful page.

If an existing API returns a list, keep that API and add a separate coverage report rather than changing the return type underneath callers. If older stored results contain only IDs, leave their coverage unknown. Do not reconstruct a token chain from their length.

This example analyzes one trusted, bounded capture in memory. It has no HTTP transport, retries, persistent cursor recovery, or snapshot-consistency guarantee. Its conservative failed-attempt policy means even a later successful retry leaves that capture interrupted. A production retry policy needs explicit attempt evidence and its own tests before it can make a stronger claim.

Gotchas

The short page looks finished. A reader may stop after an empty or undersized page while a continuation remains. The release’s tests include an empty terminal page; this example adds an empty middle page to prove that only the token determines which one it is.

The last page hides an earlier failure. Looking only at the final token can turn a gapped retrieval into a complete result. The shipped ingest keeps its first adverse observation. The example does the same, even while retaining useful IDs from later valid responses.

The limit gets mistaken for success. Stopping after the configured number of pages is an intentional budget decision. Report it as capped when a token remains. Test a terminal page at exactly the same boundary so you do not mislabel genuine exhaustion either.

Old data acquires evidence it never had. The release leaves legacy coverage unknown when no chain is available. Keep that distinction when adding the feature to a stored list. An empty legacy result and an observed empty terminal response are different facts.

Sources

• Google AIP-158: Pagination — continuation tokens, consistent query arguments, and pages that return fewer records than requested.
• Gmail users.messages.list — distinct continuation and estimated-total fields.
• DynamoDB query pagination — continuation through filtered empty pages and the provider-specific key format.

Changelog

• Ingest paginated samples and report fetch completeness (4022195).