"""Bounded public GitHub issue collection. Teaching example; tests NOT RUN.

No LLM, write API, credentials, redirects, retries or production claim.
Python 3.12+ / standard library. Live invocation makes real public GETs.
"""
from __future__ import annotations

import argparse
from dataclasses import dataclass
from datetime import datetime, timezone
import json
import re
import time
from typing import Callable
from urllib.error import HTTPError, URLError
from urllib.parse import parse_qsl, urlencode, urlsplit
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

ALLOWED_REPOS = {"psf/requests", "python/cpython"}
API = "https://api.github.com"
API_VERSION = "2026-03-10"
PER_PAGE = 5
MAX_BODY_BYTES = 262_144
MAX_LINK_CHARS = 16_384
FILTERS = {"state": "open", "sort": "created", "direction": "asc"}


class Stop(Exception):
    def __init__(self, reason: str):
        self.reason = reason
        super().__init__(reason)


@dataclass(frozen=True)
class Reply:
    status: int
    headers: dict[str, str]
    body: bytes


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def live_get(url: str, timeout: float) -> Reply:
    # Disabling proxy discovery is an explicit teaching boundary, not SSRF proof.
    opener = build_opener(ProxyHandler({}), NoRedirect())
    req = Request(url, method="GET", headers={
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": API_VERSION,
        "User-Agent": "hybridmind-pagination-teaching/1.0",
    })
    try:
        with opener.open(req, timeout=timeout) as response:
            headers = {k.lower(): v for k, v in response.headers.items()}
            links = response.headers.get_all("Link", [])
            if links:
                headers["link"] = ", ".join(links)
            return Reply(response.status, headers, response.read(MAX_BODY_BYTES + 1))
    except HTTPError as exc:
        with exc:
            return Reply(exc.code, {k.lower(): v for k, v in exc.headers.items()}, b"")
    except (URLError, TimeoutError, OSError) as exc:
        raise Stop("NETWORK_ERROR") from exc


def no_constant(value):
    raise ValueError("Non-finite JSON constant")


def no_duplicate_keys(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("Duplicate JSON key")
        result[key] = value
    return result


def decode(reply: Reply):
    if reply.status != 200:
        raise Stop(f"HTTP_{reply.status}")
    if len(reply.body) > MAX_BODY_BYTES:
        raise Stop("BODY_TOO_LARGE")
    media = reply.headers.get("content-type", "").split(";", 1)[0].strip().lower()
    if media not in {"application/json", "application/vnd.github+json"}:
        raise Stop("UNEXPECTED_CONTENT_TYPE")
    try:
        return json.loads(reply.body.decode("utf-8"), parse_constant=no_constant,
                          object_pairs_hook=no_duplicate_keys)
    except (UnicodeError, ValueError, RecursionError) as exc:
        raise Stop("INVALID_JSON") from exc


def positive_int(value) -> bool:
    return type(value) is int and value > 0


def repository_id(data, repo: str) -> int:
    if (not isinstance(data, dict) or not positive_int(data.get("id"))
            or not isinstance(data.get("full_name"), str)
            or data["full_name"].lower() != repo
            or data.get("private") is not False
            or data.get("url") != f"{API}/repos/{repo}"):
        raise Stop("REPOSITORY_IDENTITY_INVALID")
    return data["id"]


def checked_rows(data, repo: str, rid: int):
    if not isinstance(data, list) or len(data) > PER_PAGE:
        raise Stop("PAGE_SCHEMA_INVALID")
    result = []
    repo_urls = {f"{API}/repos/{repo}", f"{API}/repositories/{rid}"}
    for row in data:
        if (not isinstance(row, dict) or not positive_int(row.get("id"))
                or not positive_int(row.get("number"))
                or row.get("repository_url") not in repo_urls
                or row.get("state") != "open"
                or not isinstance(row.get("title"), str)
                or len(row["title"]) > 4096):
            raise Stop("ITEM_IDENTITY_INVALID")
        is_pr = "pull_request" in row
        if is_pr and not isinstance(row["pull_request"], dict):
            raise Stop("ITEM_IDENTITY_INVALID")
        expected = f"https://github.com/{repo}/{'pull' if is_pr else 'issues'}/{row['number']}"
        if row.get("html_url") != expected:
            raise Stop("ITEM_URL_INVALID")
        result.append({"id": row["id"], "number": row["number"],
                       "title": row["title"], "url": expected, "is_pr": is_pr})
    return result


def continuation(header: str | None, repo: str, rid: int, page: int) -> str | None:
    """Narrow documented GitHub Link layout, NOT a general RFC 8288 parser.

    Unsupported syntax is an honest stop, never absence-of-next evidence.
    All entries must be common absolute-URL ; rel="one relation" form.
    """
    if header is None:
        return None
    if not header.strip() or len(header) > MAX_LINK_CHARS:
        raise Stop("LINK_UNSUPPORTED")
    links = {}
    seen_rel = set()
    for entry in header.split(","):
        match = re.fullmatch(r'\s*<([^<>\s,]+)>\s*;\s*rel\s*=\s*"(next|prev|first|last)"\s*', entry)
        if not match or match[2] in seen_rel:
            raise Stop("LINK_UNSUPPORTED")
        seen_rel.add(match[2])
        candidate, relation = match[1], match[2]
        try:
            parsed = urlsplit(candidate)
            pairs = parse_qsl(parsed.query, keep_blank_values=True, strict_parsing=True)
            query = dict(pairs)
            page_text = query.get("page", "")
            if not re.fullmatch(r"[1-9][0-9]{0,9}", page_text):
                raise Stop("LINK_UNSUPPORTED")
            linked_page = int(page_text)
            paths = {f"/repos/{repo}/issues", f"/repositories/{rid}/issues"}
            expected = dict(FILTERS, per_page=str(PER_PAGE), page=page_text)
            if (parsed.scheme != "https" or parsed.netloc != "api.github.com"
                    or parsed.path not in paths or parsed.fragment
                    or len(pairs) != len(query) or query != expected):
                raise Stop("LINK_OUT_OF_SCOPE")
            if ((relation == "next" and linked_page != page + 1)
                    or (relation == "prev" and linked_page != page - 1)
                    or (relation == "first" and linked_page != 1)
                    or (relation == "last" and linked_page < page)):
                raise Stop("LINK_INCONSISTENT")
            links[relation] = (candidate, linked_page)
        except ValueError as exc:
            raise Stop("LINK_UNSUPPORTED") from exc
    if "next" not in links:
        if "last" in links and links["last"][1] > page:
            raise Stop("LINK_INCONSISTENT")
        return None
    if "last" in links and links["last"][1] < links["next"][1]:
        raise Stop("LINK_INCONSISTENT")
    return links["next"][0]  # Actual validated Link; never a guessed next query.


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def collect(repo: str, *, max_pages: int = 3, budget_seconds: float = 20.0,
            fetch: Callable[[str, float], Reply] = live_get,
            clock: Callable[[], float] = time.monotonic) -> dict:
    if repo not in ALLOWED_REPOS:
        raise ValueError("Repository is not allowlisted")
    if type(max_pages) is not int or not 1 <= max_pages <= 3:
        raise ValueError("max_pages must be 1..3")
    if type(budget_seconds) not in (int, float) or not 0 < budget_seconds <= 60:
        raise ValueError("budget_seconds must be >0 and <=60")
    started = utc_now()
    deadline = clock() + budget_seconds
    page = 1
    pages = []
    issues = {}
    seen_ids = {}
    seen_numbers = {}
    raw_count = pr_count = duplicate_count = request_count = 0
    status, reason = "FAILED", "NOT_STARTED"
    next_page = None
    rid = None
    endpoint = f"{API}/repos/{repo}/issues"
    evidence = []
    warnings = []

    def get(url):
        nonlocal request_count
        remaining = deadline - clock()
        if remaining <= 0:
            raise Stop("TIME_BUDGET_EXHAUSTED")
        request_count += 1
        reply = fetch(url, min(5.0, remaining))
        if clock() >= deadline:
            raise Stop("TIME_BUDGET_EXHAUSTED")
        return Reply(reply.status, {k.lower(): v for k, v in reply.headers.items()}, reply.body)

    try:
        meta_reply = get(f"{API}/repos/{repo}")
        rid = repository_id(decode(meta_reply), repo)
        if (meta_reply.headers.get("retry-after") is not None
                or meta_reply.headers.get("x-ratelimit-remaining") == "0"):
            raise Stop("SERVER_WAIT_REQUIRED")
        next_page = 1
        url = endpoint + "?" + urlencode(dict(FILTERS, per_page=PER_PAGE, page=1))
        while True:
            reply = get(url)
            rows = checked_rows(decode(reply), repo, rid)
            # Reject an inconsistent page atomically, retaining only earlier pages.
            page_ids, page_numbers = dict(seen_ids), dict(seen_numbers)
            for row in rows:
                identity = (row["number"], row["is_pr"])
                if ((row["id"] in page_ids and page_ids[row["id"]] != identity)
                        or (row["number"] in page_numbers and page_numbers[row["number"]] != row["id"])):
                    raise Stop("CONFLICTING_IDENTITY")
                page_ids[row["id"]] = identity
                page_numbers[row["number"]] = row["id"]
            if clock() >= deadline:
                raise Stop("TIME_BUDGET_EXHAUSTED")
            for row in rows:
                if row["id"] in seen_ids:
                    duplicate_count += 1
                if row["is_pr"]:
                    pr_count += 1
                else:
                    issues[row["id"]] = {k: row[k] for k in ("id", "number", "title", "url")}
                seen_ids[row["id"]] = (row["number"], row["is_pr"])
                seen_numbers[row["number"]] = row["id"]
            raw_count += len(rows)
            pages.append({"page": page, "raw_records": len(rows), "url": url})
            next_page = None  # Until a trustworthy continuation is validated.
            following = continuation(reply.headers.get("link"), repo, rid, page)
            evidence.append({"page": page, "http_status": 200,
                             "continuation": "NEXT" if following else "NO_NEXT_ACCEPTED"})
            if following is None:
                status, reason = "COMPLETE", "NO_NEXT_LINK"
                break
            next_page = page + 1
            if (reply.headers.get("retry-after") is not None
                    or reply.headers.get("x-ratelimit-remaining") == "0"):
                raise Stop("SERVER_WAIT_REQUIRED")
            if len(pages) >= max_pages:
                raise Stop("MAX_PAGES_REACHED")
            page += 1
            url = following
    except Stop as exc:
        status = "PARTIAL" if pages else "FAILED"
        reason = exc.reason
    if duplicate_count:
        warnings.append("DUPLICATES_OBSERVED_NO_SNAPSHOT_GUARANTEE")
    return {
        "status": status, "stop_reason": reason, "repository": repo,
        "repository_id": rid, "source_endpoint": endpoint,
        "filters": dict(FILTERS), "per_page": PER_PAGE, "max_pages": max_pages,
        "request_count": request_count, "pages_accepted": pages,
        "raw_records_seen": raw_count, "pull_request_records_excluded": pr_count,
        "duplicate_records_seen": duplicate_count, "unique_issues": len(issues),
        "issues": list(issues.values()), "next_page": next_page,
        "exhaustion_evidence": evidence, "warnings": warnings,
        "started_at_utc": started, "finished_at_utc": utc_now(),
        "snapshot_guaranteed": False,
        "scope": "Public open issues in this repository; page traversal, not an immutable inventory",
    }


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("repository", choices=sorted(ALLOWED_REPOS))
    parser.add_argument("--max-pages", type=int, default=3, choices=(1, 2, 3))
    args = parser.parse_args()
    print(json.dumps(collect(args.repository, max_pages=args.max_pages),
                     ensure_ascii=False, indent=2))

"""Authored synthetic tests. NOT RUN. Never requests live network resources."""
import json
import unittest

from pagination_tool import API, FILTERS, PER_PAGE, Reply, collect, continuation
from urllib.parse import urlencode

REPO = "psf/requests"
RID = 123


def response(data, link=None, **extra_headers):
    headers = {"content-type": "application/json", **extra_headers}
    if link is not None:
        headers["link"] = link
    return Reply(200, headers, json.dumps(data).encode())


def meta(**overrides):
    return response({"id": RID, "full_name": REPO, "private": False,
                     "url": f"{API}/repos/{REPO}", **overrides})


def row(n, *, pr=False, **overrides):
    record = {"id": n + 1000, "number": n, "repository_url": f"{API}/repos/{REPO}",
              "state": "open", "title": f"Synthetic item {n}",
              "html_url": f"https://github.com/{REPO}/{'pull' if pr else 'issues'}/{n}"}
    if pr:
        record["pull_request"] = {"url": f"{API}/repos/{REPO}/pulls/{n}"}
    record.update(overrides)
    return record


def next_link(page, *, numeric=False, **overrides):
    query = dict(FILTERS, per_page=PER_PAGE, page=page)
    query.update(overrides)
    path = f"/repositories/{RID}/issues" if numeric else f"/repos/{REPO}/issues"
    return f'<{API}{path}?{urlencode(query)}>; rel="next"'


class SyntheticTransport:
    def __init__(self, replies):
        self.replies = list(replies)
        self.requests = []

    def __call__(self, url, timeout):
        self.requests.append((url, timeout))
        if not self.replies:
            raise AssertionError("Unplanned request: synthetic transport only")
        return self.replies.pop(0)


class PaginationContractTests(unittest.TestCase):
    def run_case(self, replies, **options):
        transport = SyntheticTransport(replies)
        result = collect(REPO, fetch=transport, clock=lambda: 0.0, **options)
        return result, transport

    def test_one_page_empty_complete(self):
        result, transport = self.run_case([meta(), response([])])
        self.assertEqual(result["status"], "COMPLETE")
        self.assertEqual(result["unique_issues"], 0)
        self.assertEqual(len(result["pages_accepted"]), 1)
        self.assertEqual(len(transport.requests), 2)
        self.assertFalse(result["snapshot_guaranteed"])

    def test_full_page_without_next_is_complete(self):
        result, _ = self.run_case([meta(), response([row(n) for n in range(1, 6)])])
        self.assertEqual(result["status"], "COMPLETE")
        self.assertEqual(result["raw_records_seen"], 5)

    def test_follows_actual_validated_numeric_link(self):
        link = next_link(2, numeric=True)
        result, transport = self.run_case([meta(), response([row(1)], link), response([row(2)])])
        self.assertEqual(result["status"], "COMPLETE")
        self.assertEqual(transport.requests[2][0], link[1:link.index(">")])
        self.assertEqual(result["unique_issues"], 2)

    def test_three_page_cap_is_partial_and_no_fourth_page_fetch(self):
        pages = [response([row(n) for n in range(p * 5 - 4, p * 5 + 1)], next_link(p + 1))
                 for p in range(1, 4)]
        result, transport = self.run_case([meta(), *pages])
        self.assertEqual((result["status"], result["stop_reason"]), ("PARTIAL", "MAX_PAGES_REACHED"))
        self.assertEqual(result["raw_records_seen"], 15)
        self.assertEqual(result["next_page"], 4)
        self.assertEqual(len(transport.requests), 4)

    def test_excludes_pull_requests_and_records_duplicates(self):
        result, _ = self.run_case([meta(), response([row(1), row(2, pr=True)], next_link(2)),
                                   response([row(1), row(3)])])
        self.assertEqual(result["status"], "COMPLETE")
        self.assertEqual(result["raw_records_seen"], 4)
        self.assertEqual(result["unique_issues"], 2)
        self.assertEqual(result["pull_request_records_excluded"], 1)
        self.assertEqual(result["duplicate_records_seen"], 1)
        self.assertFalse(result["snapshot_guaranteed"])

    def test_invalid_next_host_stops_partial_without_external_request(self):
        link = next_link(2).replace("api.github.com", "example.invalid")
        result, transport = self.run_case([meta(), response([row(1)], link)])
        self.assertEqual(result["status"], "PARTIAL")
        self.assertEqual(result["stop_reason"], "LINK_OUT_OF_SCOPE")
        self.assertEqual(len(transport.requests), 2)
        self.assertIsNone(result["next_page"])

    def test_wrong_numeric_repository_id_stops_partial(self):
        link = next_link(2, numeric=True).replace("repositories/123", "repositories/999")
        result, _ = self.run_case([meta(), response([row(1)], link)])
        self.assertEqual(result["stop_reason"], "LINK_OUT_OF_SCOPE")

    def test_scope_change_stops_partial(self):
        result, _ = self.run_case([meta(), response([row(1)], next_link(2, state="all"))])
        self.assertEqual(result["status"], "PARTIAL")
        self.assertEqual(result["stop_reason"], "LINK_OUT_OF_SCOPE")

    def test_malformed_or_relative_last_only_is_not_complete(self):
        for link in ('<not-an-absolute-url>; rel="last"',
                     '</repos/psf/requests/issues?state=open&sort=created&direction=asc&per_page=5&page=1>; rel="last"'):
            with self.subTest(link=link):
                result, _ = self.run_case([meta(), response([row(1)], link)])
                self.assertEqual(result["status"], "PARTIAL")
                self.assertIn(result["stop_reason"], ("LINK_UNSUPPORTED", "LINK_OUT_OF_SCOPE"))

    def test_last_beyond_current_without_next_is_inconsistent(self):
        link = next_link(3).replace('rel="next"', 'rel="last"')
        result, _ = self.run_case([meta(), response([row(1)], link)])
        self.assertEqual(result["status"], "PARTIAL")
        self.assertEqual(result["stop_reason"], "LINK_INCONSISTENT")

    def test_duplicate_next_and_unsupported_valid_rfc_shape_are_not_complete(self):
        for link in (next_link(2) + ", " + next_link(2), next_link(2).replace('rel="next"', 'rel=next')):
            with self.subTest(link=link):
                result, _ = self.run_case([meta(), response([row(1)], link)])
                self.assertEqual(result["status"], "PARTIAL")
                self.assertEqual(result["stop_reason"], "LINK_UNSUPPORTED")

    def test_first_bad_json_is_failed(self):
        bad = Reply(200, {"content-type": "application/json"}, b"not-json")
        result, _ = self.run_case([meta(), bad])
        self.assertEqual(result["status"], "FAILED")
        self.assertEqual(result["stop_reason"], "INVALID_JSON")

    def test_second_page_error_retains_only_first_valid_page(self):
        result, _ = self.run_case([meta(), response([row(1)], next_link(2)), Reply(503, {}, b"")])
        self.assertEqual(result["status"], "PARTIAL")
        self.assertEqual(result["unique_issues"], 1)
        self.assertEqual(result["next_page"], 2)
        self.assertEqual(result["stop_reason"], "HTTP_503")

    def test_wrong_repository_row_rejects_whole_page(self):
        result, _ = self.run_case([meta(), response([row(1), row(2, repository_url=f"{API}/repos/x/y")])])
        self.assertEqual(result["status"], "FAILED")
        self.assertEqual(result["raw_records_seen"], 0)
        self.assertEqual(len(result["pages_accepted"]), 0)

    def test_non_string_metadata_name_is_honest_failure(self):
        result, transport = self.run_case([meta(full_name=None)])
        self.assertEqual(result["stop_reason"], "REPOSITORY_IDENTITY_INVALID")
        self.assertEqual(len(transport.requests), 1)

    def test_server_wait_stops_before_next_page(self):
        result, transport = self.run_case([meta(), response([row(1)], next_link(2), **{"retry-after": "10"})])
        self.assertEqual(result["status"], "PARTIAL")
        self.assertEqual(result["stop_reason"], "SERVER_WAIT_REQUIRED")
        self.assertEqual(len(transport.requests), 2)

    def test_redirect_is_failed_and_not_followed(self):
        result, _ = self.run_case([Reply(301, {"location": "https://example.invalid"}, b"")])
        self.assertEqual(result["status"], "FAILED")
        self.assertEqual(result["stop_reason"], "HTTP_301")

    def test_deadline_before_request_is_failed(self):
        ticks = iter([0.0, 21.0])
        transport = SyntheticTransport([])
        result = collect(REPO, fetch=transport, clock=lambda: next(ticks))
        self.assertEqual(result["stop_reason"], "TIME_BUDGET_EXHAUSTED")
        self.assertEqual(len(transport.requests), 0)

    def test_nonallowlisted_input_rejected_without_fetch(self):
        transport = SyntheticTransport([])
        with self.assertRaises(ValueError):
            collect("private/example", fetch=transport)
        self.assertEqual(len(transport.requests), 0)


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

# Honest bounded pagination — Python teaching example

Runtime: Python 3.12+; docs reviewed at Python3.12.15 / GitHub API2026-03-10. Dependencies: standard library only. Environment variables: none; no API token is used. Proxy environment discovery disabled intentionally. No dependency installation required. Place both `.py` files in one directory.

## Execution and permissions

The author has NOT RUN/imported/compiled this code or executed these tests. The following are reader instructions, not recorded test results. Synthetic tests inject a fake transport and do not call live APIs. The CLI `pagination_tool.py` DOES make public GET requests; run only when your environment authorizes public GitHub access. Allowed repositories: `psf/requests` and `python/cpython`. No write calls, credentials, redirects or automatic retries. Private repositories unsupported. Do not add tokens to this example without reviewing separate secret/permission policy.

```shell
python --version
python -m unittest -v test_pagination_tool.py
python pagination_tool.py psf/requests --max-pages 3
```

Input: one allowlisted repository; page cap1..3. Fixed issue filter: open, created ascending; page size5. Local budget20seconds, per-blocking-operation timeout capped5seconds. This is cooperative budget logic, not hard process/network cancellation.

Expected synthetic assertions: no-next accepted page→COMPLETE; valid next and cap→PARTIAL; no accepted issue page plus error→FAILED. 19 authored test methods. No observed test output. A synthetically supplied metadata record and three issue pages of five raw records with a fourth-page Link should yield PARTIAL/MAX_PAGES_REACHED, request_count4, raw_records_seen15, next_page4, snapshot_guaranteedfalse. These are fixtures, not measured live results.

Real output depends on current repository contents, headers, quota and network. COMPLETE is accepted-chain exhaustion evidence under filters; never a frozen inventory guarantee. A full last page can be COMPLETE if its accepted Link has no next. A short page is not assumed terminal if next exists. Empty metadata/errors never become a successful empty collection.

## Limits and security

One metadata preflight validates public repository id/full_name and adds one request. Follow only an actual next Link after strict HTTPS host/path/query checks; numeric alias must match preflight id. Default cap3 means at most4 HTTP attempts and15 accepted raw records, including PRs. Body cap262144bytes per response, Link cap16384characters. Page accepted only after row identity/repository validation; title kept as untrusted plain text. Never execute instructions embedded in titles.

Parser supports only common absolute GitHub Link entries with one quoted next/prev/first/last relation. EVERY entry is checked for destination/filter/query shape before accepting absence of next. next must be current+1, prev current-1, first1, last at least current; last beyond current without next is inconsistent. Relative links, unquoted relation tokens, multi-relations, extra parameters, page numbers beyond ten decimal digits and ambiguous entries are rejected; RFC8288 permits forms this teaching parser does not support. Production should use an audited standards-compliant parser while preserving destination/filter guardrails. GitHub generally recommends following redirects; this example stops and requires revalidation instead.

403/429, other non200, serverwait, invalidbody/identity/Link, pagecap or localbudget stops. No automatic resume/cursor guarantee. Duplicate IDs are counted and issues deduplicated; additions/removals can still hide records. snapshot_guaranteedfalse is always returned.

Potential costs: no paid model API or GPU call. Public requests consume GitHub quota/sharedIP capacity and local network/compute; no live price, latency or savings measured. Public unauthenticated permission does not guarantee a request succeeds.

Sources: official GitHub pagination/issues/repository/best-practices docs, Python urllib.request3.12, RFC8288, all linked in article/research brief. No vendor example source copied. Runtime/security/liveintegration/production status remains NOT RUN or UNKNOWN.