Delete _page_feedback

_page_feedback/__init__.py

@@ -1,47 +0,0 @@
-# scikitplot/_externals/_sphinx_ext/_sphinx_feedback/__init__.py
-#
-# Authors: The scikit-plots developers
-# SPDX-License-Identifier: BSD-3-Clause
-
-"""
-Privacy-minimal page feedback for Sphinx and static HTML.
-
-The dependency-free helpers in this package can be imported without Sphinx.
-Sphinx-specific integration is imported lazily by :func:`setup`.
-"""
-
-from __future__ import annotations
-
-from ._contracts import (
-    FeedbackConflictError,
-    FeedbackValidationError,
-    build_feedback_event,
-    canonical_feedback_bytes,
-    decode_feedback_request,
-    feedback_event_request_hash,
-    feedback_request_from_event,
-    feedback_request_hash,
-    parse_feedback_request,
-)
-
-__version__ = "0.6.1"
-
-__all__ = [
-    "FeedbackConflictError",
-    "FeedbackValidationError",
-    "build_feedback_event",
-    "canonical_feedback_bytes",
-    "decode_feedback_request",
-    "feedback_event_request_hash",
-    "feedback_request_from_event",
-    "feedback_request_hash",
-    "parse_feedback_request",
-    "setup",
-]
-
-
-def setup(app):
-    """Register the Sphinx adapter lazily so helpers remain dependency-free."""
-    from ._sphinx import setup_extension  # noqa: PLC0415 -- Sphinx is optional
-
-    return setup_extension(app)

_page_feedback/_contracts.py

@@ -1,560 +0,0 @@
-"""Strict, privacy-minimal contracts for generic page feedback."""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import re
-import unicodedata
-from typing import Any
-
-REQUEST_CONTRACT = "page.feedback-request.v1"
-EVENT_CONTRACT = "page.feedback-event.v1"
-RECEIPT_CONTRACT = "page.feedback-receipt.v1"
-AGGREGATE_CONTRACT = "page.feedback-aggregate.v3"
-
-# A 2,000-code-point Unicode comment can exceed 8 KiB once encoded. Keep the
-# canonical envelope and transport limits aligned so the documented comment
-# limit remains usable for non-ASCII text as well as ASCII prose.
-MAX_REQUEST_BYTES = 16 * 1024
-MAX_AGGREGATE_BYTES = 2 * 1024 * 1024
-MAX_SITE_ID = 96
-MAX_PAGE_ID = 512
-MAX_PAGE_REVISION = 128
-MAX_COMMENT = 2000
-MAX_CONTRIBUTOR = 80
-
-_SITE_ID_RE = re.compile(r"[A-Za-z0-9](?:[A-Za-z0-9._-]{0,94}[A-Za-z0-9])?\Z")
-_FEEDBACK_ID_RE = re.compile(r"feedback-[0-9a-f]{48}\Z")
-_REQUEST_HASH_RE = re.compile(r"[0-9a-f]{64}\Z")
-_ALLOWED_TOP = frozenset(
-    {
-        "contract",
-        "action",
-        "site_id",
-        "page_id",
-        "page_revision",
-        "feedback_id",
-        "rating",
-        "mode",
-        "comment",
-        "contributor",
-    }
-)
-_ALLOWED_CONTRIBUTOR = frozenset({"display_name"})
-_ALLOWED_EVENT_TOP = frozenset(
-    {
-        "contract",
-        "site_id",
-        "page_id",
-        "page_revision",
-        "feedback",
-    },
-)
-_ALLOWED_EVENT_FEEDBACK = frozenset(
-    {
-        "id",
-        "rating",
-        "mode",
-        "comment",
-        "contributor",
-    },
-)
-
-
-class FeedbackValidationError(ValueError):
-    """Raised when a public feedback envelope violates the contract."""
-
-    def __init__(self, message: str, *, code: str = "invalid_feedback") -> None:
-        super().__init__(message)
-        self.code = code
-
-
-class FeedbackConflictError(RuntimeError):
-    """Raised when an idempotency key is replayed with different content."""
-
-
-def _reject_unknown_keys(
-    value: dict[str, Any],
-    allowed: frozenset[str],
-    path: str,
-) -> None:
-    extras = sorted(str(key) for key in value if key not in allowed)
-    if extras:
-        raise FeedbackValidationError(
-            f"{path}: unsupported field(s): {', '.join(extras)}",
-            code="unexpected_field",
-        )
-
-
-def _nfc_text(value: Any, *, path: str, limit: int, allow_empty: bool) -> str:
-    if not isinstance(value, str):
-        raise FeedbackValidationError(f"{path}: must be a string")
-    try:
-        text = unicodedata.normalize("NFC", value)
-        text.encode("utf-8", "strict")
-    except (UnicodeError, ValueError) as exc:
-        raise FeedbackValidationError(f"{path}: invalid Unicode text") from exc
-    if len(text) > limit:
-        raise FeedbackValidationError(f"{path}: exceeds {limit} characters")
-    if not allow_empty and not text:
-        raise FeedbackValidationError(f"{path}: must not be empty")
-    return text
-
-
-def _public_credit(value: Any) -> str:
-    text = _nfc_text(
-        value,
-        path="contributor.display_name",
-        limit=MAX_CONTRIBUTOR,
-        allow_empty=True,
-    )
-    if any(
-        ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-        for ch in text
-    ):
-        raise FeedbackValidationError(
-            "contributor.display_name: control characters are not allowed"
-        )
-    text = re.sub(r" +", " ", text.strip(" "))
-    if len(text) > MAX_CONTRIBUTOR:
-        raise FeedbackValidationError(
-            f"contributor.display_name: exceeds {MAX_CONTRIBUTOR} characters"
-        )
-    return text
-
-
-def _comment(value: Any) -> str:
-    text = _nfc_text(
-        value,
-        path="comment",
-        limit=MAX_COMMENT,
-        allow_empty=True,
-    ).strip(" \t\n\r")
-    # Preserve line breaks and tabs in deliberate prose, but reject the rest of
-    # C0/DEL. This mirrors the AI Learn reviewed-comment boundary.
-    for ch in text:
-        code = ord(ch)
-        if code == 127 or (  # ruff: ignore[magic-value-comparison]
-            code < 32 and ch not in "\t\n\r"  # ruff: ignore[magic-value-comparison]
-        ):
-            raise FeedbackValidationError("comment: unsupported control characters")
-    return text
-
-
-def normalize_page_id(value: Any) -> str:
-    """Validate a canonical page identity without interpreting it as a URL/path."""
-    text = _nfc_text(
-        value, path="page_id", limit=MAX_PAGE_ID, allow_empty=False
-    ).strip()
-    if (
-        text.startswith("/")
-        or text.endswith("/")
-        or "\\" in text
-        or "?" in text
-        or "#" in text
-        or any(
-            ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-            for ch in text
-        )
-    ):
-        raise FeedbackValidationError("page_id: invalid canonical page name")
-    parts = text.split("/")
-    if any(part in {"", ".", ".."} for part in parts):
-        raise FeedbackValidationError("page_id: invalid path segment")
-    return text
-
-
-def normalize_page_revision(value: Any) -> str:
-    if value is None or value == "":
-        return ""
-    text = _nfc_text(
-        value,
-        path="page_revision",
-        limit=MAX_PAGE_REVISION,
-        allow_empty=True,
-    ).strip()
-    if any(
-        ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-        for ch in text
-    ):
-        raise FeedbackValidationError(
-            "page_revision: control characters are not allowed",
-        )
-    return text
-
-
-def _site_id(value: Any) -> str:
-    site_id = _nfc_text(
-        value, path="site_id", limit=MAX_SITE_ID, allow_empty=False
-    ).strip()
-    if not _SITE_ID_RE.fullmatch(site_id):
-        raise FeedbackValidationError("site_id: use a stable ASCII identifier")
-    return site_id
-
-
-def normalize_site_id(value: Any) -> str:
-    """Validate and canonicalize a stable public site identifier."""
-    return _site_id(value)
-
-
-def _feedback_id(value: Any) -> str:
-    if not isinstance(value, str) or not _FEEDBACK_ID_RE.fullmatch(value):
-        raise FeedbackValidationError(
-            "feedback_id: expected feedback- followed by 48 lowercase hexadecimal characters"
-        )
-    return value
-
-
-def _rating(value: Any) -> int:
-    _value = -5 <= value <= 5  # ruff: ignore[magic-value-comparison]
-    if isinstance(value, bool) or not isinstance(value, int) or not _value:
-        raise FeedbackValidationError("rating: expected an integer from -5 through +5")
-    return value
-
-
-def _mode(value: Any) -> str:
-    if value not in {"quick", "detailed"}:
-        raise FeedbackValidationError(
-            "mode: expected 'quick' or 'detailed'",
-        )
-    return value
-
-
-def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
-    result: dict[str, Any] = {}
-    for key, value in pairs:
-        if key in result:
-            raise FeedbackValidationError(f"duplicate JSON field: {key}")
-        result[key] = value
-    return result
-
-
-def _reject_nonfinite_json(value: str) -> None:
-    raise FeedbackValidationError(f"non-finite JSON number is not allowed: {value}")
-
-
-def decode_feedback_request(raw: bytes | str) -> dict[str, Any]:
-    """
-    Decode one wire request with strict JSON semantics, then validate it.
-
-    Duplicate object keys and non-finite numeric spellings are rejected before
-    contract normalization so different JSON parsers cannot disagree about the
-    event that an idempotency key represents.
-    """
-    if not isinstance(raw, (bytes, str)):
-        raise FeedbackValidationError("Feedback body must be JSON text")
-    try:
-        payload = json.loads(
-            raw,
-            object_pairs_hook=_reject_duplicate_json_keys,
-            parse_constant=_reject_nonfinite_json,
-        )
-    except FeedbackValidationError:
-        raise
-    except (
-        UnicodeDecodeError,
-        json.JSONDecodeError,
-        RecursionError,
-        ValueError,
-    ) as exc:
-        raise FeedbackValidationError(
-            "Invalid JSON",
-            code="invalid_json",
-        ) from exc
-    return parse_feedback_request(payload)
-
-
-def decode_feedback_event(raw: bytes | str) -> dict[str, Any]:
-    """Decode one durable event with the same strict JSON semantics as requests."""
-    if not isinstance(raw, (bytes, str)):
-        raise FeedbackValidationError("Feedback event must be JSON text")
-    try:
-        payload = json.loads(
-            raw,
-            object_pairs_hook=_reject_duplicate_json_keys,
-            parse_constant=_reject_nonfinite_json,
-        )
-    except FeedbackValidationError:
-        raise
-    except (
-        UnicodeDecodeError,
-        json.JSONDecodeError,
-        RecursionError,
-        ValueError,
-    ) as exc:
-        raise FeedbackValidationError(
-            "Invalid feedback event JSON",
-            code="invalid_event_json",
-        ) from exc
-    return parse_feedback_event(payload)
-
-
-def parse_feedback_request(payload: Any) -> dict[str, Any]:
-    """
-    Validate and normalize one browser feedback request.
-
-    Unknown fields are rejected deliberately. This prevents future browser,
-    device, account, referrer, IP, or analytics fields from silently becoming
-    durable feedback data.
-    """
-    if not isinstance(payload, dict):
-        raise FeedbackValidationError("Feedback body must be a JSON object")
-    _reject_unknown_keys(payload, _ALLOWED_TOP, "feedback")
-
-    if payload.get("contract") != REQUEST_CONTRACT:
-        raise FeedbackValidationError(
-            "Unsupported feedback contract", code="unsupported_contract"
-        )
-    if payload.get("action") != "submit":
-        raise FeedbackValidationError("Unsupported feedback action")
-
-    site_id = normalize_site_id(payload.get("site_id"))
-    page_id = normalize_page_id(payload.get("page_id"))
-    page_revision = normalize_page_revision(payload.get("page_revision", ""))
-    feedback_id = _feedback_id(payload.get("feedback_id"))
-    rating = _rating(payload.get("rating"))
-    mode = _mode(payload.get("mode"))
-
-    raw_contributor = payload.get("contributor", {})
-    if raw_contributor is None:
-        raw_contributor = {}
-    if not isinstance(raw_contributor, dict):
-        raise FeedbackValidationError("contributor: must be an object")
-    _reject_unknown_keys(raw_contributor, _ALLOWED_CONTRIBUTOR, "contributor")
-    contributor = _public_credit(raw_contributor.get("display_name", ""))
-    if contributor.lower() == "anonymous":
-        raise FeedbackValidationError(
-            "contributor.display_name: 'Anonymous' is reserved; leave public credit blank instead"
-        )
-    comment = _comment(payload.get("comment", ""))
-
-    if mode == "quick":
-        if rating not in {-1, 1}:
-            raise FeedbackValidationError("quick feedback rating must be -1 or +1")
-        if comment or contributor:
-            raise FeedbackValidationError(
-                "quick feedback must not include comment or contributor content"
-            )
-
-    normalized: dict[str, Any] = {
-        "contract": REQUEST_CONTRACT,
-        "action": "submit",
-        "site_id": site_id,
-        "page_id": page_id,
-        "feedback_id": feedback_id,
-        "rating": rating,
-        "mode": mode,
-        "contributor": {"display_name": contributor},
-    }
-    if page_revision:
-        normalized["page_revision"] = page_revision
-    if comment:
-        normalized["comment"] = comment
-    return normalized
-
-
-def canonical_feedback_bytes(request: dict[str, Any]) -> bytes:
-    """Return canonical UTF-8 JSON used for request commitments/idempotency."""
-    normalized = parse_feedback_request(request)
-    encoded = json.dumps(
-        normalized,
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
-    ).encode("utf-8")
-    if len(encoded) > MAX_REQUEST_BYTES:
-        raise FeedbackValidationError(
-            f"feedback request exceeds {MAX_REQUEST_BYTES} canonical bytes"
-        )
-    return encoded
-
-
-def feedback_request_hash(request: dict[str, Any]) -> str:
-    """Return a deterministic commitment to the normalized feedback request."""
-    return hashlib.sha256(canonical_feedback_bytes(request)).hexdigest()
-
-
-def build_feedback_event(request: dict[str, Any]) -> dict[str, Any]:
-    """Project a request into the minimal durable feedback event."""
-    normalized = parse_feedback_request(request)
-    feedback: dict[str, Any] = {
-        "id": normalized["feedback_id"],
-        "rating": normalized["rating"],
-        "mode": normalized["mode"],
-        "contributor": normalized["contributor"]["display_name"] or "Anonymous",
-    }
-    if normalized.get("comment"):
-        feedback["comment"] = normalized["comment"]
-    event: dict[str, Any] = {
-        "contract": EVENT_CONTRACT,
-        "site_id": normalized["site_id"],
-        "page_id": normalized["page_id"],
-        "feedback": feedback,
-    }
-    if normalized.get("page_revision"):
-        event["page_revision"] = normalized["page_revision"]
-    return event
-
-
-def feedback_request_from_event(event: dict[str, Any]) -> dict[str, Any]:
-    """
-    Reconstruct the unique normalized request represented by a durable event.
-
-    This inverse projection is intentionally strict. It gives storage/provider
-    adapters an independent way to verify that a supplied request commitment
-    actually commits to the exact durable event they are about to accept.
-    """
-    normalized = parse_feedback_event(event)
-    feedback = normalized["feedback"]
-    contributor = feedback["contributor"]
-    request: dict[str, Any] = {
-        "contract": REQUEST_CONTRACT,
-        "action": "submit",
-        "site_id": normalized["site_id"],
-        "page_id": normalized["page_id"],
-        "feedback_id": feedback["id"],
-        "rating": feedback["rating"],
-        "mode": feedback["mode"],
-        "contributor": {
-            "display_name": "" if contributor == "Anonymous" else contributor
-        },
-    }
-    if normalized.get("page_revision"):
-        request["page_revision"] = normalized["page_revision"]
-    if feedback.get("comment"):
-        request["comment"] = feedback["comment"]
-    return parse_feedback_request(request)
-
-
-def feedback_event_request_hash(event: dict[str, Any]) -> str:
-    """Return the request commitment implied by one canonical durable event."""
-    return feedback_request_hash(feedback_request_from_event(event))
-
-
-def parse_feedback_event(  # ruff: ignore[too-many-branches]
-    payload: Any,
-) -> dict[str, Any]:
-    """Validate one durable event and require canonical public text forms."""
-    if not isinstance(payload, dict):
-        raise FeedbackValidationError("feedback event must be an object")
-    _reject_unknown_keys(payload, _ALLOWED_EVENT_TOP, "feedback event")
-    if payload.get("contract") != EVENT_CONTRACT:
-        raise FeedbackValidationError("Unsupported feedback event contract")
-
-    site_id = normalize_site_id(payload.get("site_id"))
-    page_id = normalize_page_id(payload.get("page_id"))
-    page_revision = normalize_page_revision(payload.get("page_revision", ""))
-    raw_feedback = payload.get("feedback")
-    if not isinstance(raw_feedback, dict):
-        raise FeedbackValidationError("feedback event feedback must be an object")
-    _reject_unknown_keys(
-        raw_feedback,
-        _ALLOWED_EVENT_FEEDBACK,
-        "feedback event.feedback",
-    )
-    feedback_id = _feedback_id(raw_feedback.get("id"))
-    rating = _rating(raw_feedback.get("rating"))
-    mode = _mode(raw_feedback.get("mode"))
-
-    raw_contributor = raw_feedback.get("contributor")
-    if raw_contributor == "Anonymous":
-        contributor = "Anonymous"
-    else:
-        contributor = _public_credit(raw_contributor)
-        if contributor.lower() == "anonymous":
-            raise FeedbackValidationError(
-                "feedback event contributor reserves 'Anonymous' for the anonymity sentinel"
-            )
-        if not contributor:
-            raise FeedbackValidationError(
-                "feedback event contributor must be 'Anonymous' or a non-empty public credit"
-            )
-        if contributor != raw_contributor:
-            raise FeedbackValidationError(
-                "feedback event contributor must already be canonicalized"
-            )
-
-    comment = _comment(raw_feedback.get("comment", ""))
-    if "comment" in raw_feedback and comment != raw_feedback.get("comment"):
-        raise FeedbackValidationError(
-            "feedback event comment must already be canonicalized",
-        )
-    if mode == "quick":
-        if rating not in {-1, 1}:
-            raise FeedbackValidationError(
-                "quick feedback rating must be -1 or +1",
-            )
-        if comment or contributor != "Anonymous":
-            raise FeedbackValidationError(
-                "quick feedback event must not contain comment or public credit",
-            )
-
-    normalized_feedback: dict[str, Any] = {
-        "id": feedback_id,
-        "rating": rating,
-        "mode": mode,
-        "contributor": contributor,
-    }
-    if comment:
-        normalized_feedback["comment"] = comment
-    normalized: dict[str, Any] = {
-        "contract": EVENT_CONTRACT,
-        "site_id": site_id,
-        "page_id": page_id,
-        "feedback": normalized_feedback,
-    }
-    if page_revision:
-        normalized["page_revision"] = page_revision
-    return normalized
-
-
-def canonical_event_bytes(event: dict[str, Any]) -> bytes:
-    """Return deterministic bytes for durable event equality checks."""
-    normalized = parse_feedback_event(event)
-    return (
-        json.dumps(
-            normalized,
-            ensure_ascii=False,
-            sort_keys=True,
-            separators=(",", ":"),
-        )
-        + "\n"
-    ).encode("utf-8")
-
-
-def repository_event_bytes(event: dict[str, Any]) -> bytes:
-    """
-    Return deterministic human-readable JSON for repository event files.
-
-    Repository presentation is intentionally separate from
-    :func:`canonical_event_bytes`: whitespace must never become feedback-event
-    identity or alter request/idempotency commitments.
-    """
-    normalized = parse_feedback_event(event)
-    return (
-        json.dumps(
-            normalized,
-            ensure_ascii=False,
-            indent=2,
-            sort_keys=True,
-        )
-        + "\n"
-    ).encode("utf-8")
-
-
-def validate_request_hash(value: Any) -> str:
-    """Validate a lowercase SHA-256 request commitment."""
-    if not isinstance(value, str) or not _REQUEST_HASH_RE.fullmatch(value):
-        raise FeedbackValidationError(
-            "request_hash must be 64 lowercase hexadecimal characters",
-        )
-    return value
-
-
-def page_digest(site_id: str, page_id: str) -> str:
-    """Return a filesystem-safe opaque page bucket without hiding page_id in JSON."""
-    site = normalize_site_id(site_id)
-    page = normalize_page_id(page_id)
-    raw = f"{site}\0{page}".encode()
-    return hashlib.sha256(raw).hexdigest()[:24]

_page_feedback/_service/__init__.py

@@ -1,20 +0,0 @@
-"""Server-only generic page-feedback service API."""
-
-from ._config import (
-    FeedbackServiceConfig,
-    FeedbackServiceConfigError,
-    StorageTarget,
-    load_service_config,
-    parse_storage_targets,
-)
-from ._core import FeedbackServiceUnavailable, PageFeedbackService
-
-__all__ = [
-    "FeedbackServiceConfig",
-    "FeedbackServiceConfigError",
-    "FeedbackServiceUnavailable",
-    "PageFeedbackService",
-    "StorageTarget",
-    "load_service_config",
-    "parse_storage_targets",
-]

_page_feedback/_service/_config.py

@@ -1,775 +0,0 @@
-"""Server-only feedback storage/review policy parsing."""
-
-from __future__ import annotations
-
-import json
-import os
-import re
-from dataclasses import dataclass
-from pathlib import Path, PurePosixPath
-from typing import Any, Mapping
-
-from .._contracts import normalize_page_id, normalize_page_revision, normalize_site_id
-
-PROVIDERS = frozenset({"sqlite", "github", "huggingface", "gitlab", "bitbucket"})
-IMPLEMENTED_PROVIDERS = frozenset({"sqlite", "github"})
-DEFAULT_FEEDBACK_STORAGE_TARGETS: tuple = ()
-ROLES = frozenset({"primary", "mirror"})
-REVIEW_MODES = frozenset({"disabled", "sqlite", "provider-pr"})
-PAGE_AUTHORITY_CONTRACT = "page.feedback-authority.v1"
-_MAX_PAGE_AUTHORITY_BYTES = 4 * 1024 * 1024
-_MAX_PAGE_AUTHORITY_SITES = 32
-_MAX_PAGE_AUTHORITY_PAGES_PER_SITE = 250_000
-_ALLOWED_TARGET_KEYS = frozenset(
-    {
-        "id",
-        "label",
-        "authority",
-        "provider",
-        "role",
-        "repo",
-        "branch",
-        "paths",
-        "token_env",
-        "database",
-        "expose_links",
-    }
-)
-_ALLOWED_PATH_KEYS = frozenset({"feedback"})
-_REPO_RE = re.compile(r"[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}\Z")
-_ENV_RE = re.compile(r"[A-Z][A-Z0-9_]{0,127}\Z")
-_ALLOWED_EXACT_TOKEN_ENVS = frozenset(
-    {
-        "GITHUB_TOKEN",
-        "HF_TOKEN",
-        "GITLAB_TOKEN",
-        "BITBUCKET_TOKEN",
-    },
-)
-_FEEDBACK_TOKEN_ENV_RE = re.compile(r"FEEDBACK_[A-Z0-9_]*TOKEN[A-Z0-9_]*\Z")
-_ALLOWED_TOKEN_PREFIXES = ("AI_RECORD_STORAGE_TOKEN_",)
-
-
-class FeedbackServiceConfigError(ValueError):
-    pass
-
-
-def _reject_duplicate_json_keys(pairs):
-    result = {}
-    for key, value in pairs:
-        if key in result:
-            raise FeedbackServiceConfigError(
-                f"duplicate JSON field in feedback configuration: {key}",
-            )
-        result[key] = value
-    return result
-
-
-def _reject_nonfinite_json(value: str) -> None:
-    raise FeedbackServiceConfigError(
-        f"non-finite JSON number is not allowed in feedback configuration: {value}"
-    )
-
-
-@dataclass(frozen=True)
-class StorageTarget:
-    id: str
-    label: str
-    provider: str
-    role: str
-    authority: str = "feedback"
-    repo: str = ""
-    branch: str = "main"
-    feedback_path: str = "feedback"
-    token_env: tuple[str, ...] = ()
-    database: str = ""
-    expose_links: bool = False
-
-    def resolve_token(self, env: Mapping[str, str] = os.environ) -> tuple[str, str]:
-        for name in self.token_env:
-            raw_value = env.get(name, "")
-            if raw_value in (None, ""):
-                continue
-            if not isinstance(raw_value, str):
-                raise FeedbackServiceConfigError(
-                    f"feedback credential value in {name!r} must be a string"
-                )
-            value = raw_value
-            _len = len(value) > 4096  # ruff: ignore[magic-value-comparison]
-            if _len or any(
-                not 33 <= ord(ch) <= 126  # ruff: ignore[magic-value-comparison]
-                for ch in value
-            ):
-                raise FeedbackServiceConfigError(
-                    f"feedback credential value in {name!r} is invalid"
-                )
-            return value, name
-        return "", ""
-
-
-@dataclass(frozen=True)
-class FeedbackServiceConfig:
-    review_mode: str
-    targets: tuple[StorageTarget, ...]
-    allowed_site_ids: tuple[str, ...]
-    page_authority: tuple[tuple[str, str, str], ...] | None
-    max_body_bytes: int
-    rate_limit_per_hour: int
-    mirror_timeout_seconds: float
-
-    @property
-    def primary(self) -> StorageTarget | None:
-        return next(
-            (target for target in self.targets if target.role == "primary"),
-            None,
-        )
-
-    @property
-    def mirrors(self) -> tuple[StorageTarget, ...]:
-        return tuple(target for target in self.targets if target.role == "mirror")
-
-
-def _token_envs(value: Any, provider: str) -> tuple[str, ...]:
-    if value in (None, ""):
-        defaults = {
-            "github": (
-                "FEEDBACK_GITHUB_TOKEN",
-                "GITHUB_TOKEN",
-                "AI_RECORD_STORAGE_TOKEN_GITHUB_MIRROR",
-            ),
-            "huggingface": (
-                "FEEDBACK_HF_TOKEN",
-                "HF_TOKEN",
-                "AI_RECORD_STORAGE_TOKEN_HF_PRIMARY",
-            ),
-            "gitlab": (
-                "FEEDBACK_GITLAB_TOKEN",
-                "GITLAB_TOKEN",
-            ),
-            "bitbucket": (
-                "FEEDBACK_BITBUCKET_TOKEN",
-                "BITBUCKET_TOKEN",
-            ),
-            "sqlite": (),
-        }
-        return defaults[provider]
-    if isinstance(value, str):
-        values = (value,)
-    elif isinstance(value, (list, tuple)):
-        values = tuple(value)
-    else:
-        values = ()
-    if not values and value not in ([], ()):
-        raise FeedbackServiceConfigError(
-            "token_env must be a string or list of environment-variable names",
-        )
-    if provider != "sqlite" and not values:
-        raise FeedbackServiceConfigError(
-            f"{provider} storage target requires at least one token_env credential alias",
-        )
-    if len(values) > 8:  # ruff: ignore[magic-value-comparison]
-        raise FeedbackServiceConfigError(
-            "token_env contains too many credential aliases",
-        )
-    result = []
-    for item in values:
-        if not isinstance(item, str):
-            raise FeedbackServiceConfigError(
-                "token_env entries must be environment-variable name strings",
-            )
-        name = item.strip()
-        if not _ENV_RE.fullmatch(name):
-            raise FeedbackServiceConfigError(
-                "token_env contains an invalid environment-variable name",
-            )
-        if (
-            name not in _ALLOWED_EXACT_TOKEN_ENVS
-            and not _FEEDBACK_TOKEN_ENV_RE.fullmatch(name)
-            and not name.startswith(_ALLOWED_TOKEN_PREFIXES)
-        ):
-            raise FeedbackServiceConfigError(
-                f"token_env {name!r} is outside the feedback credential allowlist",
-            )
-        if name not in result:
-            result.append(name)
-    return tuple(result)
-
-
-def _target_string(
-    raw: dict[str, Any], key: str, *, default: str = "", allow_empty: bool = True
-) -> str:
-    value = raw.get(key)
-    if value is None:
-        value = default
-    if not isinstance(value, str):
-        raise FeedbackServiceConfigError(f"storage target {key} must be a string")
-    try:
-        value.encode("utf-8", "strict")
-    except UnicodeError as exc:
-        raise FeedbackServiceConfigError(
-            f"storage target {key} contains invalid Unicode"
-        ) from exc
-    text = value.strip()
-    if not allow_empty and not text:
-        raise FeedbackServiceConfigError(f"storage target {key} must not be empty")
-    return text
-
-
-def _safe_feedback_path(value: Any) -> str:
-    if not isinstance(value, str):
-        raise FeedbackServiceConfigError("feedback path must be a string")
-    try:
-        value.encode("utf-8", "strict")
-    except UnicodeError as exc:
-        raise FeedbackServiceConfigError(
-            "feedback path contains invalid Unicode",
-        ) from exc
-    text = value.strip()
-    if (
-        not text
-        or len(text) > 512  # ruff: ignore[magic-value-comparison]
-        or text.startswith("/")
-        or text.endswith("/")
-        or "\\" in text
-        or any(
-            ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-            for ch in text
-        )
-    ):
-        raise FeedbackServiceConfigError("feedback path is invalid")
-    path = PurePosixPath(text)
-    if (
-        path.is_absolute()
-        or any(part in {"", ".", ".."} for part in path.parts)
-        or str(path) != text
-    ):
-        raise FeedbackServiceConfigError(
-            "feedback path must be a canonical safe relative POSIX path"
-        )
-    return text
-
-
-def _parse_target(  # ruff: ignore[too-many-branches]
-    raw: Any,
-) -> StorageTarget:
-    if not isinstance(raw, dict):
-        raise FeedbackServiceConfigError(
-            "each feedback storage target must be an object",
-        )
-    forbidden = {
-        "token",
-        "secret",
-        "password",
-        "authorization",
-    } & set(raw)
-    if forbidden:
-        raise FeedbackServiceConfigError(
-            "storage targets may name token_env values but may not contain literal secrets",
-        )
-    unknown = sorted(set(raw) - _ALLOWED_TARGET_KEYS)
-    if unknown:
-        raise FeedbackServiceConfigError(
-            (
-                "feedback storage target contains unsupported field(s): "
-                + ", ".join(unknown)
-            ),
-        )
-    target_id = _target_string(raw, "id", allow_empty=False)
-    if not re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,63}", target_id):
-        raise FeedbackServiceConfigError(
-            "storage target id must be a lowercase stable identifier",
-        )
-    authority = _target_string(
-        raw,
-        "authority",
-        default="feedback",
-        allow_empty=False,
-    ).lower()
-    if authority != "feedback":
-        raise FeedbackServiceConfigError(
-            "storage target authority must be 'feedback'",
-        )
-    provider = _target_string(raw, "provider", allow_empty=False).lower()
-    role = _target_string(raw, "role", allow_empty=False).lower()
-    if provider not in PROVIDERS:
-        raise FeedbackServiceConfigError(
-            f"unsupported feedback storage provider {provider!r}",
-        )
-    if role not in ROLES:
-        raise FeedbackServiceConfigError(
-            "storage target role must be primary or mirror",
-        )
-    label = _target_string(raw, "label", default=target_id, allow_empty=False)
-    _len = len(label) > 120  # ruff: ignore[magic-value-comparison]
-    if _len or any(
-        ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-        for ch in label
-    ):
-        raise FeedbackServiceConfigError(
-            "storage target label is invalid",
-        )
-    repo = _target_string(raw, "repo")
-    if provider != "sqlite":
-        if not _REPO_RE.fullmatch(repo):
-            raise FeedbackServiceConfigError(
-                f"{provider} storage target requires owner/repo",
-            )
-        owner, repository = repo.split("/", 1)
-        if owner in {".", ".."} or repository in {".", ".."}:
-            raise FeedbackServiceConfigError(
-                f"{provider} storage target requires a canonical owner/repo",
-            )
-    branch = _target_string(
-        raw,
-        "branch",
-        default="main",
-        allow_empty=False,
-    )
-    if (
-        len(branch) > 128  # ruff: ignore[magic-value-comparison]
-        or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,127}", branch)
-        or ".." in branch
-        or "//" in branch
-        or branch.endswith(("/", ".", ".lock"))
-        or any(
-            part in {"", ".", ".."} or part.endswith(".lock")
-            for part in branch.split("/")
-        )
-    ):
-        raise FeedbackServiceConfigError(
-            "storage target branch is invalid",
-        )
-    if provider == "github" and branch == "feedback":
-        raise FeedbackServiceConfigError(
-            "GitHub base branch 'feedback' conflicts with the deterministic feedback/<nonce> review namespace",
-        )
-    paths = raw.get("paths", None)
-    if paths is None:
-        paths = {}
-    if not isinstance(paths, dict):
-        raise FeedbackServiceConfigError(
-            "storage target paths must be an object",
-        )
-    unknown_paths = sorted(set(paths) - _ALLOWED_PATH_KEYS)
-    if unknown_paths:
-        raise FeedbackServiceConfigError(
-            (
-                "storage target paths contains unsupported field(s): "
-                + ", ".join(unknown_paths)
-            ),
-        )
-    feedback_path = _safe_feedback_path(paths.get("feedback", "feedback"))
-    database = _target_string(raw, "database")
-    if len(database) > 1024 or any(  # ruff: ignore[magic-value-comparison]
-        ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-        for ch in database
-    ):
-        raise FeedbackServiceConfigError(
-            "sqlite database path is invalid",
-        )
-    if provider == "sqlite" and not database:
-        database = "feedback.sqlite3"
-    if provider == "sqlite" and database == ":memory:":
-        raise FeedbackServiceConfigError(
-            "sqlite :memory: is unsupported because feedback opens bounded independent connections",
-        )
-    if provider != "sqlite" and database:
-        raise FeedbackServiceConfigError(
-            "database is only valid for sqlite targets",
-        )
-    if provider == "sqlite":
-        if repo:
-            raise FeedbackServiceConfigError(
-                "repo is not valid for sqlite targets",
-            )
-        if "branch" in raw:
-            raise FeedbackServiceConfigError(
-                "branch is not valid for sqlite targets",
-            )
-        if "paths" in raw:
-            raise FeedbackServiceConfigError(
-                "paths is not valid for sqlite targets",
-            )
-        if raw.get("token_env") not in (None, "", [], ()):
-            raise FeedbackServiceConfigError(
-                "token_env is not valid for sqlite targets",
-            )
-        if raw.get("expose_links", False) not in (False, None):
-            raise FeedbackServiceConfigError(
-                "expose_links is not valid for sqlite targets",
-            )
-    token_env = _token_envs(raw.get("token_env"), provider)
-    expose_links = raw.get("expose_links", False)
-    if not isinstance(expose_links, bool):
-        raise FeedbackServiceConfigError(
-            "storage target expose_links must be a boolean",
-        )
-    return StorageTarget(
-        id=target_id,
-        label=label,
-        provider=provider,
-        role=role,
-        authority=authority,
-        repo=repo,
-        branch=branch,
-        feedback_path=feedback_path,
-        token_env=token_env,
-        database=database,
-        expose_links=expose_links,
-    )
-
-
-def parse_storage_targets(raw: Any) -> tuple[StorageTarget, ...]:
-    if raw in (None, "", []):
-        return ()
-    if isinstance(raw, str):
-        try:
-            raw_size = len(raw.encode("utf-8", "strict"))
-        except UnicodeError as exc:
-            raise FeedbackServiceConfigError(
-                "FEEDBACK_STORAGE_TARGETS contains invalid Unicode"
-            ) from exc
-        if raw_size > 64 * 1024:
-            raise FeedbackServiceConfigError(
-                "FEEDBACK_STORAGE_TARGETS exceeds the 64 KiB safety limit"
-            )
-        try:
-            raw = json.loads(
-                raw,
-                object_pairs_hook=_reject_duplicate_json_keys,
-                parse_constant=_reject_nonfinite_json,
-            )
-        except FeedbackServiceConfigError:
-            raise
-        except (json.JSONDecodeError, ValueError, RecursionError) as exc:
-            raise FeedbackServiceConfigError(
-                "FEEDBACK_STORAGE_TARGETS is not valid JSON",
-            ) from exc
-    if raw == []:
-        return ()
-    _len = len(raw) > 5  # ruff: ignore[magic-value-comparison]
-    if not isinstance(raw, list) or _len:
-        raise FeedbackServiceConfigError(
-            "feedback storage targets must be a list with at most five entries",
-        )
-    targets = tuple(_parse_target(item) for item in raw)
-    if len({target.id for target in targets}) != len(targets):
-        raise FeedbackServiceConfigError(
-            "feedback storage target ids must be unique",
-        )
-    primaries = [target for target in targets if target.role == "primary"]
-    if len(primaries) != 1:
-        raise FeedbackServiceConfigError(
-            "feedback storage topology requires exactly one primary target",
-        )
-    return targets
-
-
-def _bounded_int(value: Any, default: int, low: int, high: int, *, name: str) -> int:
-    if value in (None, ""):
-        return default
-    if isinstance(value, bool):
-        raise FeedbackServiceConfigError(f"{name} must be an integer")
-    try:
-        parsed = int(value)
-    except (TypeError, ValueError) as exc:
-        raise FeedbackServiceConfigError(f"{name} must be an integer") from exc
-    if not low <= parsed <= high:
-        raise FeedbackServiceConfigError(f"{name} must be between {low} and {high}")
-    return parsed
-
-
-def _bounded_float(
-    value: Any, default: float, low: float, high: float, *, name: str
-) -> float:
-    if value in (None, ""):
-        return default
-    if isinstance(value, bool):
-        raise FeedbackServiceConfigError(f"{name} must be a number")
-    try:
-        parsed = float(value)
-    except (TypeError, ValueError) as exc:
-        raise FeedbackServiceConfigError(f"{name} must be a number") from exc
-    if not low <= parsed <= high:
-        raise FeedbackServiceConfigError(f"{name} must be between {low} and {high}")
-    return parsed
-
-
-def _load_page_authority(  # ruff: ignore[too-many-branches]
-    value: Any,
-) -> tuple[tuple[str, str, str], ...] | None:
-    """Load an optional bounded server-side page/revision authority manifest."""
-    if value in (None, ""):
-        return None
-    if not isinstance(value, str):
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_PAGE_AUTHORITY_FILE must be a filesystem path string"
-        )
-    try:
-        value.encode("utf-8", "strict")
-    except UnicodeError as exc:
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_PAGE_AUTHORITY_FILE contains invalid Unicode"
-        ) from exc
-    path_text = value.strip()
-    _len = len(path_text) > 2048  # ruff: ignore[magic-value-comparison]
-    if (
-        not path_text
-        or _len
-        or any(
-            ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-            for ch in path_text
-        )
-    ):
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_PAGE_AUTHORITY_FILE path is invalid",
-        )
-    path = Path(path_text)
-    try:
-        data = path.read_bytes()
-    except OSError as exc:
-        raise FeedbackServiceConfigError(
-            "unable to read FEEDBACK_PAGE_AUTHORITY_FILE"
-        ) from exc
-    if len(data) > _MAX_PAGE_AUTHORITY_BYTES:
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_PAGE_AUTHORITY_FILE exceeds the 4 MiB safety limit"
-        )
-    try:
-        payload = json.loads(
-            data,
-            object_pairs_hook=_reject_duplicate_json_keys,
-            parse_constant=_reject_nonfinite_json,
-        )
-    except FeedbackServiceConfigError:
-        raise
-    except (
-        UnicodeDecodeError,
-        json.JSONDecodeError,
-        ValueError,
-        RecursionError,
-    ) as exc:
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_PAGE_AUTHORITY_FILE is not valid UTF-8 JSON",
-        ) from exc
-    if (
-        not isinstance(payload, dict)
-        or payload.get("contract") != PAGE_AUTHORITY_CONTRACT
-    ):
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_PAGE_AUTHORITY_FILE has an unsupported contract",
-        )
-    unknown = sorted(set(payload) - {"contract", "sites"})
-    if unknown:
-        raise FeedbackServiceConfigError(
-            (
-                "feedback page authority contains unsupported field(s): "
-                + ", ".join(unknown)
-            ),
-        )
-    sites = payload.get("sites")
-    if not isinstance(sites, dict) or len(sites) > _MAX_PAGE_AUTHORITY_SITES:
-        raise FeedbackServiceConfigError(
-            "feedback page authority sites must be a bounded object",
-        )
-    rows: list[tuple[str, str, str]] = []
-    for raw_site_id, pages in sites.items():
-        try:
-            site_id = normalize_site_id(raw_site_id)
-        except ValueError as exc:
-            raise FeedbackServiceConfigError(
-                "feedback page authority contains an invalid site_id",
-            ) from exc
-        if site_id != raw_site_id:
-            raise FeedbackServiceConfigError(
-                "feedback page authority site_id must already be canonicalized",
-            )
-        if (
-            not isinstance(pages, dict)
-            or len(pages) > _MAX_PAGE_AUTHORITY_PAGES_PER_SITE
-        ):
-            raise FeedbackServiceConfigError(
-                "feedback page authority pages must be a bounded object",
-            )
-        for raw_page_id, raw_revision in pages.items():
-            try:
-                page_id = normalize_page_id(raw_page_id)
-                revision = normalize_page_revision(raw_revision)
-            except ValueError as exc:
-                raise FeedbackServiceConfigError(
-                    "feedback page authority contains an invalid page or revision"
-                ) from exc
-            if page_id != raw_page_id or revision != raw_revision:
-                raise FeedbackServiceConfigError(
-                    "feedback page authority values must already be canonicalized"
-                )
-            rows.append((site_id, page_id, revision))
-    return tuple(rows)
-
-
-def _allowed_site_ids(value: Any) -> tuple[str, ...]:
-    """
-    Parse an optional exact server-side site allowlist.
-
-    The browser-provided ``site_id`` is descriptive event data, not authority.
-    Deployments can bind one feedback service to one or more known site IDs with
-    ``FEEDBACK_ALLOWED_SITE_IDS``. Blank preserves generic multi-site compatibility.
-    """
-    if value in (None, ""):
-        return ()
-    if not isinstance(value, str):
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_ALLOWED_SITE_IDS must be a comma-separated string"
-        )
-    try:
-        raw_size = len(value.encode("utf-8", "strict"))
-    except UnicodeError as exc:
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_ALLOWED_SITE_IDS contains invalid Unicode"
-        ) from exc
-    if raw_size > 4096:  # ruff: ignore[magic-value-comparison]
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_ALLOWED_SITE_IDS exceeds the 4 KiB safety limit"
-        )
-    parts = [part.strip() for part in value.split(",")]
-    if any(not part for part in parts):
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_ALLOWED_SITE_IDS contains an empty site identifier"
-        )
-    if len(parts) > 32:  # ruff: ignore[magic-value-comparison]
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_ALLOWED_SITE_IDS contains too many site identifiers"
-        )
-    result: list[str] = []
-    for part in parts:
-        try:
-            site_id = normalize_site_id(part)
-        except ValueError as exc:
-            raise FeedbackServiceConfigError(
-                "FEEDBACK_ALLOWED_SITE_IDS contains an invalid site identifier"
-            ) from exc
-        if site_id not in result:
-            result.append(site_id)
-    return tuple(result)
-
-
-def load_service_config(
-    env: Mapping[str, str] = os.environ,
-) -> FeedbackServiceConfig:
-    mode = (
-        str(
-            env.get("FEEDBACK_REVIEW_MODE", "disabled") or "disabled",
-        )
-        .strip()
-        .lower()
-    )
-    if mode not in REVIEW_MODES:
-        raise FeedbackServiceConfigError(
-            f"FEEDBACK_REVIEW_MODE must be one of {sorted(REVIEW_MODES)}",
-        )
-    raw_targets = env.get("FEEDBACK_STORAGE_TARGETS", "")
-    if raw_targets:
-        targets = parse_storage_targets(raw_targets)
-    elif mode == "sqlite":
-        targets = parse_storage_targets(
-            [
-                {
-                    "id": "sqlite-primary",
-                    "label": "SQLite Feedback",
-                    "authority": "feedback",
-                    "provider": "sqlite",
-                    "role": "primary",
-                    "database": (
-                        str(
-                            env.get("FEEDBACK_SQLITE_PATH", "feedback.sqlite3")
-                            or "feedback.sqlite3"
-                        ).strip()
-                    ),
-                }
-            ]
-        )
-    elif (
-        mode == "provider-pr"
-        and str(env.get("FEEDBACK_GITHUB_REPOSITORY", "") or "").strip()
-    ):
-        targets = parse_storage_targets(
-            [
-                {
-                    "id": "github-primary",
-                    "label": "GitHub Feedback Review",
-                    "authority": "feedback",
-                    "provider": "github",
-                    "role": "primary",
-                    "repo": str(env["FEEDBACK_GITHUB_REPOSITORY"]).strip(),
-                    "branch": (
-                        str(
-                            env.get("FEEDBACK_GITHUB_DEFAULT_BRANCH", "main") or "main",
-                        ).strip()
-                    ),
-                    "paths": {
-                        "feedback": env.get("FEEDBACK_GITHUB_PATH", "feedback"),
-                    },
-                    "token_env": None,
-                    "expose_links": False,
-                }
-            ]
-        )
-    else:
-        targets = DEFAULT_FEEDBACK_STORAGE_TARGETS
-    if mode == "disabled" and targets:
-        raise FeedbackServiceConfigError(
-            "feedback storage targets must be empty when FEEDBACK_REVIEW_MODE=disabled"
-        )
-    if mode != "disabled" and not targets:
-        raise FeedbackServiceConfigError(
-            "enabled feedback review mode requires one primary storage target"
-        )
-    primary = next((target for target in targets if target.role == "primary"), None)
-    if mode == "sqlite" and primary and primary.provider != "sqlite":
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_REVIEW_MODE=sqlite requires a sqlite primary target"
-        )
-    if mode == "sqlite" and any(target.provider != "sqlite" for target in targets):
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_REVIEW_MODE=sqlite is local-only and requires every storage target to use sqlite"
-        )
-    if mode == "provider-pr" and primary and primary.provider == "sqlite":
-        raise FeedbackServiceConfigError(
-            "FEEDBACK_REVIEW_MODE=provider-pr requires a provider primary target"
-        )
-    if mode != "disabled":
-        unavailable = sorted(
-            {target.provider for target in targets} - IMPLEMENTED_PROVIDERS
-        )
-        if unavailable:
-            raise FeedbackServiceConfigError(
-                "configured feedback provider adapter(s) are not implemented in this release: "
-                + ", ".join(unavailable)
-            )
-    return FeedbackServiceConfig(
-        review_mode=mode,
-        targets=targets,
-        allowed_site_ids=_allowed_site_ids(env.get("FEEDBACK_ALLOWED_SITE_IDS")),
-        page_authority=_load_page_authority(env.get("FEEDBACK_PAGE_AUTHORITY_FILE")),
-        max_body_bytes=_bounded_int(
-            env.get("FEEDBACK_PAGE_MAX_BODY_BYTES"),
-            16 * 1024,
-            1024,
-            65536,
-            name="FEEDBACK_PAGE_MAX_BODY_BYTES",
-        ),
-        rate_limit_per_hour=_bounded_int(
-            env.get("FEEDBACK_PAGE_RATE_LIMIT_PER_HOUR"),
-            20,
-            1,
-            240,
-            name="FEEDBACK_PAGE_RATE_LIMIT_PER_HOUR",
-        ),
-        mirror_timeout_seconds=_bounded_float(
-            env.get("FEEDBACK_MIRROR_TIMEOUT_SECONDS"),
-            8.0,
-            0.5,
-            15.0,
-            name="FEEDBACK_MIRROR_TIMEOUT_SECONDS",
-        ),
-    )

_page_feedback/_service/_core.py

@@ -1,307 +0,0 @@
-"""Provider-neutral feedback service coordinator."""
-
-from __future__ import annotations
-
-import asyncio
-import os
-import sqlite3
-from pathlib import Path
-from typing import Any, Mapping
-
-from .._contracts import (
-    FeedbackConflictError,
-    FeedbackValidationError,
-    build_feedback_event,
-    feedback_request_hash,
-    parse_feedback_request,
-)
-from ._config import (
-    FeedbackServiceConfig,
-    FeedbackServiceConfigError,
-    StorageTarget,
-    load_service_config,
-)
-from ._github import ProviderWriteError, submit_github_review
-from ._sqlite import SQLiteFeedbackStore
-
-
-class FeedbackServiceUnavailable(  # ruff: ignore[error-suffix-on-exception-name]
-    RuntimeError,
-):
-    def __init__(self, code: str, message: str) -> None:
-        super().__init__(message)
-        self.code = code
-
-
-class PageFeedbackService:
-    """Accept one minimal feedback event and coordinate primary/mirror writes."""
-
-    def __init__(
-        self,
-        config: FeedbackServiceConfig,
-        *,
-        credential_env: Mapping[str, str] | None = None,
-    ) -> None:
-        self.config = config
-        self._credential_env = os.environ if credential_env is None else credential_env
-        self._page_authority = (
-            None
-            if config.page_authority is None
-            else {
-                (site_id, page_id): revision
-                for site_id, page_id, revision in config.page_authority
-            }
-        )
-
-    @classmethod
-    def from_env(cls, env=None) -> PageFeedbackService:
-        source = os.environ if env is None else env
-        return cls(load_service_config(source), credential_env=source)
-
-    async def _write_target(
-        self,
-        target: StorageTarget,
-        *,
-        request: dict[str, Any],
-        event: dict[str, Any],
-        request_hash: str,
-        http_client=None,
-    ) -> dict[str, Any]:
-        if target.provider == "sqlite":
-            store = SQLiteFeedbackStore(Path(target.database))
-            try:
-                return await asyncio.to_thread(
-                    store.put,
-                    feedback_id=event["feedback"]["id"],
-                    request_hash=request_hash,
-                    event=event,
-                )
-            except FeedbackConflictError:
-                raise
-            except (OSError, sqlite3.Error) as exc:
-                raise FeedbackServiceUnavailable(
-                    "sqlite_write_failed", "SQLite feedback storage is unavailable"
-                ) from exc
-        if target.provider == "github":
-            try:
-                token, _token_name = target.resolve_token(self._credential_env)
-            except FeedbackServiceConfigError as exc:
-                raise FeedbackServiceUnavailable(
-                    "credential_invalid",
-                    "GitHub feedback credential is invalid",
-                ) from exc
-            if http_client is None:
-                try:
-                    import httpx  # ruff: ignore[import-outside-top-level]
-                except (  # pragma: no cover - runtime dependency boundary
-                    Exception
-                ) as exc:
-                    raise FeedbackServiceUnavailable(
-                        "http_client_unavailable",
-                        "GitHub feedback transport requires httpx",
-                    ) from exc
-                timeout = httpx.Timeout(
-                    20.0, connect=10.0, read=20.0, write=20.0, pool=10.0
-                )
-                async with httpx.AsyncClient(
-                    timeout=timeout, follow_redirects=False
-                ) as client:
-                    return await submit_github_review(
-                        client=client,
-                        target=target,
-                        token=token,
-                        event=event,
-                        request_hash=request_hash,
-                    )
-            return await submit_github_review(
-                client=http_client,
-                target=target,
-                token=token,
-                event=event,
-                request_hash=request_hash,
-            )
-        raise FeedbackServiceUnavailable(
-            "provider_not_implemented",
-            f"Feedback provider {target.provider!r} is modeled but not implemented in this release",
-        )
-
-    def validate_request(
-        self,
-        payload: dict[str, Any],
-    ) -> dict[str, Any]:
-        """Normalize one request and enforce server-side submission authority."""
-        request = parse_feedback_request(payload)
-        if (
-            self.config.allowed_site_ids
-            and request["site_id"] not in self.config.allowed_site_ids
-        ):
-            raise FeedbackValidationError(
-                "site_id is not authorized by this feedback service",
-                code="site_not_allowed",
-            )
-        if self._page_authority is not None:
-            authority_key = (request["site_id"], request["page_id"])
-            if authority_key not in self._page_authority:
-                raise FeedbackValidationError(
-                    "page_id is not authorized by this feedback service",
-                    code="page_not_allowed",
-                )
-            expected_revision = self._page_authority[authority_key]
-            if (
-                expected_revision
-                and request.get("page_revision", "") != expected_revision
-            ):
-                raise FeedbackValidationError(
-                    "page_revision does not match the server-side page authority",
-                    code="page_revision_mismatch",
-                )
-        return request
-
-    @staticmethod
-    def _valid_target_receipt(
-        target: StorageTarget,
-        receipt: Any,
-        *,
-        feedback_id: str,
-        request_hash: str,
-    ) -> bool:
-        return (
-            isinstance(receipt, dict)
-            and receipt.get("status") in {"accepted", "replay"}
-            and receipt.get("provider") == target.provider
-            and receipt.get("feedback_id") == feedback_id
-            and receipt.get("request_hash") == request_hash
-        )
-
-    async def submit(
-        self,
-        payload: dict[str, Any],
-        *,
-        http_client=None,
-    ) -> dict[str, Any]:
-        if self.config.review_mode == "disabled":
-            raise FeedbackServiceUnavailable(
-                "feedback_disabled",
-                "Generic page feedback is disabled on this service",
-            )
-        request = self.validate_request(payload)
-        request_hash = feedback_request_hash(request)
-        event = build_feedback_event(request)
-        primary = self.config.primary
-        if primary is None:
-            raise FeedbackServiceUnavailable(
-                "primary_missing",
-                "No primary feedback storage target is configured",
-            )
-        try:
-            primary_receipt = await self._write_target(
-                primary,
-                request=request,
-                event=event,
-                request_hash=request_hash,
-                http_client=http_client,
-            )
-        except ProviderWriteError as exc:
-            raise FeedbackServiceUnavailable(exc.code, str(exc)) from exc
-
-        if not self._valid_target_receipt(
-            primary,
-            primary_receipt,
-            feedback_id=event["feedback"]["id"],
-            request_hash=request_hash,
-        ):
-            raise FeedbackServiceUnavailable(
-                "provider_receipt_invalid",
-                "Primary feedback provider returned an invalid receipt",
-            )
-        primary_status = primary_receipt["status"]
-
-        mirror_status: dict[str, str] = {}
-
-        async def write_mirror(mirror: StorageTarget) -> tuple[str, str]:
-            try:
-                mirror_receipt = await self._write_target(
-                    mirror,
-                    request=request,
-                    event=event,
-                    request_hash=request_hash,
-                    http_client=http_client,
-                )
-                if not self._valid_target_receipt(
-                    mirror,
-                    mirror_receipt,
-                    feedback_id=event["feedback"]["id"],
-                    request_hash=request_hash,
-                ):
-                    return mirror.id, "degraded"
-                return mirror.id, "ok"
-            except asyncio.CancelledError:
-                raise
-            except (
-                FeedbackConflictError,
-                FeedbackServiceUnavailable,
-                ProviderWriteError,
-                OSError,
-            ):
-                # Mirrors are durability only. Primary authority already accepted.
-                return mirror.id, "degraded"
-
-        tasks = {
-            asyncio.create_task(write_mirror(mirror)): mirror.id
-            for mirror in self.config.mirrors
-        }
-        if tasks:
-            pending: set[asyncio.Task] = set()
-            try:
-                done, pending = await asyncio.wait(
-                    tasks, timeout=self.config.mirror_timeout_seconds
-                )
-                for task in done:
-                    mirror_id = tasks[task]
-                    try:
-                        resolved_id, status = task.result()
-                    except asyncio.CancelledError:
-                        # A child mirror can be cancelled independently by its
-                        # transport/runtime. Primary authority has already
-                        # succeeded, so treat that as durability degradation;
-                        # parent cancellation is handled by the outer finally.
-                        resolved_id, status = mirror_id, "degraded"
-                    except Exception:  # ruff: ignore[blind-except]
-                        resolved_id, status = mirror_id, "degraded"
-                    mirror_status[resolved_id] = status
-                for task in pending:
-                    mirror_status[tasks[task]] = "degraded"
-            finally:
-                # If the request task itself is cancelled after primary authority
-                # has committed, never leave mirror writes detached from their
-                # parent lifecycle.  Cancelling and gathering unfinished children
-                # makes shutdown/disconnect behavior deterministic; a later retry
-                # can safely resume every provider with the same event nonce.
-                unfinished = [task for task in tasks if not task.done()]
-                for task in unfinished:
-                    task.cancel()
-                if unfinished:
-                    await asyncio.gather(*unfinished, return_exceptions=True)
-        return {
-            "ok": True,
-            "contract": "page.feedback-receipt.v1",
-            "feedback_id": event["feedback"]["id"],
-            "status": primary_status,
-            "provider": primary.provider,
-            "request_hash": request_hash,
-            "review_url": (
-                primary_receipt.get("review_url", "") if primary.expose_links else ""
-            ),
-            "mirrors": {
-                mirror_id: mirror_status[mirror_id]
-                for mirror_id in sorted(mirror_status)
-            },
-        }
-
-
-__all__ = [
-    "FeedbackConflictError",
-    "FeedbackServiceConfigError",
-    "FeedbackServiceUnavailable",
-    "PageFeedbackService",
-]

_page_feedback/_service/_github.py

@@ -1,714 +0,0 @@
-"""
-GitHub provider-review adapter for immutable feedback JSON events.
-
-The adapter is deliberately retry-oriented. A network interruption may happen after
-GitHub has created a branch, committed the immutable event, or opened a pull
-request. Replaying the same feedback ID verifies each durable step and resumes at
-the first missing one instead of treating a partially completed submission as a
-permanent conflict.
-"""
-
-from __future__ import annotations
-
-import base64
-import json
-from typing import Any
-from urllib.parse import quote, urlsplit
-
-from .. import __version__
-from .._contracts import (
-    FeedbackConflictError,
-    FeedbackValidationError,
-    canonical_event_bytes,
-    decode_feedback_event,
-    feedback_event_request_hash,
-    page_digest,
-    parse_feedback_event,
-    repository_event_bytes,
-    validate_request_hash,
-)
-from ._config import StorageTarget
-
-_MAX_RESPONSE = 256 * 1024
-_MAX_RESPONSE_CHUNKS = 1024
-_REQUEST_TIMEOUT_SECONDS = 20.0
-
-
-class ProviderWriteError(RuntimeError):
-    """Bounded provider failure safe to surface as a stable service code."""
-
-    def __init__(self, code: str, message: str) -> None:
-        super().__init__(message)
-        self.code = code
-
-
-async def _bounded_json(  # ruff: ignore[too-many-branches]
-    client,
-    method: str,
-    url: str,
-    *,
-    headers: dict[str, str],
-    json_body: Any = None,
-    params: dict[str, str] | None = None,
-) -> tuple[int, Any]:
-    """Perform one bounded GitHub API request even with an unbounded shared client."""
-    response = None
-    try:
-        request = client.build_request(
-            method,
-            url,
-            headers=headers,
-            json=json_body,
-            params=params,
-            timeout=_REQUEST_TIMEOUT_SECONDS,
-        )
-        response = await client.send(request, stream=True, follow_redirects=False)
-        length = response.headers.get("content-length")
-        if length:
-            try:
-                if int(length) > _MAX_RESPONSE:
-                    raise ProviderWriteError(
-                        "provider_response_too_large",
-                        "GitHub response exceeded the safety limit",
-                    )
-            except ValueError:
-                pass
-        chunks: list[bytes] = []
-        total = 0
-        chunk_count = 0
-        async for chunk in response.aiter_bytes():
-            chunk_count += 1
-            if chunk_count > _MAX_RESPONSE_CHUNKS:
-                raise ProviderWriteError(
-                    "provider_response_too_fragmented",
-                    "GitHub response exceeded the response-frame safety limit",
-                )
-            total += len(chunk)
-            if total > _MAX_RESPONSE:
-                raise ProviderWriteError(
-                    "provider_response_too_large",
-                    "GitHub response exceeded the safety limit",
-                )
-            chunks.append(chunk)
-        raw = b"".join(chunks)
-        if not raw:
-            payload: Any = {}
-        else:
-            try:
-                payload = json.loads(raw)
-            except (
-                UnicodeDecodeError,
-                json.JSONDecodeError,
-                ValueError,
-                RecursionError,
-            ) as exc:
-                raise ProviderWriteError(
-                    "provider_response_invalid",
-                    "GitHub returned invalid JSON",
-                ) from exc
-        return int(response.status_code), payload
-    except ProviderWriteError:
-        raise
-    except Exception as exc:
-        # Do not leak low-level network details/tokens to the public route. The
-        # exception remains chained for server-side diagnostics.
-        raise ProviderWriteError(
-            "provider_transport_failed",
-            "GitHub feedback transport failed",
-        ) from exc
-    finally:
-        if response is not None:
-            try:  # ruff: ignore[suppressible-exception]
-                await response.aclose()
-            except Exception:  # ruff: ignore[blind-except]
-                pass
-
-
-def _headers(token: str) -> dict[str, str]:
-    return {
-        "Accept": "application/vnd.github+json",
-        "Authorization": f"Bearer {token}",
-        "X-GitHub-Api-Version": "2022-11-28",
-        "User-Agent": f"sphinx-feedback/{__version__}",
-    }
-
-
-def _event_path(target: StorageTarget, event: dict[str, Any]) -> str:
-    bucket = page_digest(event["site_id"], event["page_id"])
-    return f"{target.feedback_path}/pages/{bucket}/{event['feedback']['id']}.json"
-
-
-def _review_page_text(page_id: str, *, maximum: int | None = None) -> str:
-    """Return a provider-display-safe page label with no Markdown mention syntax."""
-    text = quote(page_id, safe="/._-")
-    if maximum is not None:
-        text = text[:maximum]
-    return text
-
-
-def _branch(event: dict[str, Any]) -> str:
-    # Use the full 192-bit event nonce. Branch identity is event provenance, not
-    # participant identity, and full entropy avoids unnecessary prefix collisions.
-    return "feedback/" + event["feedback"]["id"].removeprefix("feedback-")
-
-
-def _object_sha(payload: Any, *, code: str) -> str:
-    try:
-        sha = str(payload["object"]["sha"])
-    except Exception as exc:
-        raise ProviderWriteError(
-            code,
-            "GitHub branch response was incomplete",
-        ) from exc
-    _len = len(sha) != 40  # ruff: ignore[magic-value-comparison]
-    if _len or any(ch not in "0123456789abcdefABCDEF" for ch in sha):
-        raise ProviderWriteError(
-            code,
-            "GitHub branch response contained an invalid object id",
-        )
-    return sha
-
-
-def _review_url(value: Any, *, repo: str | None = None) -> str:
-    text = str(value or "").strip()
-    try:
-        parsed = urlsplit(text)
-        port = parsed.port
-    except ValueError as exc:
-        raise ProviderWriteError(
-            "review_lookup_invalid", "GitHub feedback review URL was invalid"
-        ) from exc
-    if (
-        parsed.scheme != "https"
-        or (parsed.hostname or "").lower() != "github.com"
-        or parsed.username
-        or parsed.password
-        or parsed.query
-        or parsed.fragment
-        or port not in (None, 443)
-    ):
-        raise ProviderWriteError(
-            "review_lookup_invalid", "GitHub feedback review URL was invalid"
-        )
-    if repo is not None:
-        prefix = f"/{repo}/pull/"
-        number = parsed.path.removeprefix(prefix)
-        if not parsed.path.startswith(prefix) or not number.isdigit():
-            raise ProviderWriteError(
-                "review_lookup_invalid",
-                "GitHub feedback review URL did not belong to the configured repository",
-            )
-    return text
-
-
-async def _existing_review_url(
-    client,
-    *,
-    api: str,
-    headers: dict[str, str],
-    repo: str,
-    branch: str,
-    base_branch: str,
-) -> str | None:
-    owner = repo.split("/", 1)[0]
-    status, pulls = await _bounded_json(
-        client,
-        "GET",
-        f"{api}/pulls",
-        headers=headers,
-        params={"state": "all", "head": f"{owner}:{branch}", "base": base_branch},
-    )
-    if status != 200:  # ruff: ignore[magic-value-comparison]
-        raise ProviderWriteError(
-            "review_lookup_failed",
-            "Existing GitHub feedback review could not be verified",
-        )
-    if not isinstance(pulls, list):
-        raise ProviderWriteError(
-            "review_lookup_invalid",
-            "GitHub feedback review lookup returned an invalid response",
-        )
-    if not pulls:
-        return None
-    first = pulls[0]
-    if not isinstance(first, dict):
-        raise ProviderWriteError(
-            "review_lookup_invalid",
-            "GitHub feedback review lookup returned an invalid entry",
-        )
-    return _review_url(first.get("html_url"), repo=repo)
-
-
-async def _branch_comparison(
-    client,
-    *,
-    api: str,
-    headers: dict[str, str],
-    base_branch: str,
-    branch: str,
-) -> dict[str, Any]:
-    status, comparison = await _bounded_json(
-        client,
-        "GET",
-        f"{api}/compare/{quote(base_branch, safe='')}...{quote(branch, safe='')}",
-        headers=headers,
-    )
-    _status = status != 200  # ruff: ignore[magic-value-comparison]
-    if _status or not isinstance(comparison, dict):
-        raise ProviderWriteError(
-            "existing_branch_unverifiable",
-            "Existing GitHub feedback branch could not be compared safely",
-        )
-    ahead_by = comparison.get("ahead_by")
-    if isinstance(ahead_by, bool) or not isinstance(ahead_by, int) or ahead_by < 0:
-        raise ProviderWriteError(
-            "existing_branch_unverifiable",
-            "GitHub branch comparison was incomplete",
-        )
-    return comparison
-
-
-async def _branch_has_no_unique_commits(
-    client,
-    *,
-    api: str,
-    headers: dict[str, str],
-    base_branch: str,
-    branch: str,
-) -> bool:
-    comparison = await _branch_comparison(
-        client,
-        api=api,
-        headers=headers,
-        base_branch=base_branch,
-        branch=branch,
-    )
-    return comparison["ahead_by"] == 0
-
-
-async def _verify_review_branch_scope(
-    client,
-    *,
-    api: str,
-    headers: dict[str, str],
-    base_branch: str,
-    branch: str,
-    path: str,
-    event: dict[str, Any],
-    review_exists: bool,
-) -> None:
-    """
-    Require the review branch to contain exactly one immutable feedback file.
-
-    A deterministic branch name is resumable, but it is also a security boundary:
-    an existing branch must never be allowed to smuggle unrelated repository changes
-    into the pull request merely because the expected feedback JSON is present.
-    """
-    existing_bytes = await _existing_event_bytes(
-        client, api=api, headers=headers, path=path, branch=branch
-    )
-    if not _event_content_matches(existing_bytes, event):
-        if existing_bytes is None:
-            raise FeedbackConflictError(
-                "feedback review branch is missing the expected immutable event"
-            )
-        raise FeedbackConflictError(
-            "feedback_id was already used for different feedback content"
-        )
-
-    comparison = await _branch_comparison(
-        client,
-        api=api,
-        headers=headers,
-        base_branch=base_branch,
-        branch=branch,
-    )
-    if comparison["ahead_by"] == 0:
-        # After a reviewed pull request is merged, the deterministic branch can
-        # compare equal to the base. Preserve idempotent replay only when a review
-        # already exists and the immutable event is now present on the base branch.
-        if not review_exists:
-            raise FeedbackConflictError(
-                "feedback review branch contains an unexpected commit history"
-            )
-        base_bytes = await _existing_event_bytes(
-            client, api=api, headers=headers, path=path, branch=base_branch
-        )
-        if not _event_content_matches(base_bytes, event):
-            raise FeedbackConflictError(
-                "feedback review branch no longer proves the expected reviewed event"
-            )
-        return
-    if comparison["ahead_by"] != 1:
-        raise FeedbackConflictError(
-            "feedback review branch contains an unexpected commit history"
-        )
-    files = comparison.get("files")
-    if not isinstance(files, list) or len(files) != 1 or not isinstance(files[0], dict):
-        raise FeedbackConflictError(
-            "feedback review branch contains unexpected repository changes"
-        )
-    changed = files[0]
-    if changed.get("filename") != path or changed.get("status") != "added":
-        raise FeedbackConflictError(
-            "feedback review branch contains unexpected repository changes"
-        )
-
-
-async def _branch_exists(
-    client, *, api: str, headers: dict[str, str], branch: str
-) -> bool:
-    status, payload = await _bounded_json(
-        client,
-        "GET",
-        f"{api}/git/ref/heads/{quote(branch, safe='')}",
-        headers=headers,
-    )
-    if status == 404:  # ruff: ignore[magic-value-comparison]
-        return False
-    if status != 200:  # ruff: ignore[magic-value-comparison]
-        raise ProviderWriteError(
-            "existing_branch_unverifiable",
-            "Existing GitHub feedback branch could not be verified",
-        )
-    _object_sha(payload, code="existing_branch_invalid")
-    return True
-
-
-async def _existing_event_bytes(
-    client,
-    *,
-    api: str,
-    headers: dict[str, str],
-    path: str,
-    branch: str,
-) -> bytes | None:
-    content_status, existing = await _bounded_json(
-        client,
-        "GET",
-        f"{api}/contents/{quote(path, safe='/')}",
-        headers=headers,
-        params={"ref": branch},
-    )
-    if content_status == 404:  # ruff: ignore[magic-value-comparison]
-        return None
-    _content_status = content_status != 200  # ruff: ignore[magic-value-comparison]
-    if _content_status or not isinstance(existing, dict):
-        raise ProviderWriteError(
-            "existing_branch_unverifiable",
-            "Existing GitHub feedback event could not be verified",
-        )
-    try:
-        return base64.b64decode(
-            str(existing.get("content", "")).replace("\n", ""), validate=True
-        )
-    except Exception as exc:
-        raise ProviderWriteError(
-            "existing_content_invalid",
-            "Existing GitHub feedback event could not be verified",
-        ) from exc
-
-
-def _event_content_matches(existing_bytes: bytes | None, event: dict[str, Any]) -> bool:
-    """Compare repository JSON by validated event semantics, never whitespace."""
-    if existing_bytes is None:
-        return False
-    try:
-        existing_event = decode_feedback_event(existing_bytes)
-        expected_event = parse_feedback_event(event)
-    except FeedbackValidationError:
-        return False
-    return canonical_event_bytes(existing_event) == canonical_event_bytes(
-        expected_event,
-    )
-
-
-async def submit_github_review(  # ruff: ignore[too-many-branches]
-    *,
-    client,
-    target: StorageTarget,
-    token: str,
-    event: dict[str, Any],
-    request_hash: str,
-) -> dict[str, Any]:
-    """Idempotently ensure branch, immutable event, and provider review exist."""
-    if not token:
-        raise ProviderWriteError(
-            "credential_missing", "GitHub feedback credential is not configured"
-        )
-    event = parse_feedback_event(event)
-    validate_request_hash(request_hash)
-    if request_hash != feedback_event_request_hash(event):
-        raise ProviderWriteError(
-            "request_commitment_mismatch",
-            "GitHub feedback request commitment does not match the durable event",
-        )
-    repo = target.repo
-    api = "https://api.github.com/repos/" + repo
-    headers = _headers(token)
-    branch = _branch(event)
-    if branch == target.branch:
-        raise ProviderWriteError(
-            "branch_namespace_conflict",
-            "GitHub feedback review branch collides with the configured base branch",
-        )
-    path = _event_path(target, event)
-    # Repository files are intentionally pretty-printed for human review.
-    # Idempotency/equality remains semantic so pre-v0.6.1 compact files and
-    # harmless formatting changes do not become false content conflicts.
-    event_bytes = repository_event_bytes(event)
-
-    status, base_ref = await _bounded_json(
-        client,
-        "GET",
-        f"{api}/git/ref/heads/{quote(target.branch, safe='')}",
-        headers=headers,
-    )
-    if status != 200:  # ruff: ignore[magic-value-comparison]
-        raise ProviderWriteError(
-            "base_ref_unavailable", "GitHub feedback base branch could not be resolved"
-        )
-    base_sha = _object_sha(base_ref, code="base_ref_invalid")
-
-    branch_created = False
-    try:
-        status, _created = await _bounded_json(
-            client,
-            "POST",
-            f"{api}/git/refs",
-            headers=headers,
-            json_body={"ref": "refs/heads/" + branch, "sha": base_sha},
-        )
-    except ProviderWriteError as exc:
-        if exc.code != "provider_transport_failed":
-            raise
-        # Broken pipe after GitHub accepted branch creation: reconcile before
-        # asking the browser to retry the whole workflow.
-        try:
-            if await _branch_exists(client, api=api, headers=headers, branch=branch):
-                status = 422
-            else:
-                raise exc  # ruff: ignore[verbose-raise]
-        except ProviderWriteError:
-            raise exc  # ruff: ignore[raise-without-from-inside-except]
-    if status == 201:  # ruff: ignore[magic-value-comparison]
-        branch_created = True
-    elif status == 422:  # ruff: ignore[magic-value-comparison]
-        if not await _branch_exists(client, api=api, headers=headers, branch=branch):
-            raise ProviderWriteError(
-                "branch_create_failed",
-                "GitHub rejected the feedback branch and no resumable branch exists",
-            )
-    else:
-        raise ProviderWriteError(
-            "branch_create_failed", "GitHub feedback review branch could not be created"
-        )
-
-    event_present = False
-    if not branch_created:
-        existing_bytes = await _existing_event_bytes(
-            client, api=api, headers=headers, path=path, branch=branch
-        )
-        if existing_bytes is not None:
-            if not _event_content_matches(existing_bytes, event):
-                raise FeedbackConflictError(
-                    "feedback_id was already used for different feedback content"
-                )
-            event_present = True
-        elif not await _branch_has_no_unique_commits(
-            client,
-            api=api,
-            headers=headers,
-            base_branch=target.branch,
-            branch=branch,
-        ):
-            raise FeedbackConflictError(
-                "feedback review branch exists without the expected immutable event"
-            )
-
-    if not event_present:
-        try:
-            status, _content = await _bounded_json(
-                client,
-                "PUT",
-                f"{api}/contents/{quote(path, safe='/')}",
-                headers=headers,
-                json_body={
-                    "message": f"Add page feedback {event['feedback']['id']}",
-                    "content": base64.b64encode(event_bytes).decode("ascii"),
-                    "branch": branch,
-                },
-            )
-        except ProviderWriteError as exc:
-            if exc.code != "provider_transport_failed":
-                raise
-            # The commit may have succeeded before the connection failed. Verify
-            # the immutable bytes once; otherwise preserve the ambiguous error so
-            # a later retry can safely resume the same deterministic branch.
-            try:
-                existing_bytes = await _existing_event_bytes(
-                    client, api=api, headers=headers, path=path, branch=branch
-                )
-            except ProviderWriteError:
-                raise exc  # ruff: ignore[raise-without-from-inside-except]
-            if _event_content_matches(existing_bytes, event):
-                status = 201
-            else:
-                raise exc  # ruff: ignore[verbose-raise]
-        if status not in {200, 201}:
-            # Concurrent same-event submissions can race between the existence
-            # check and PUT. Re-read immutable content before treating an
-            # explicit provider rejection as failure.
-            existing_bytes = await _existing_event_bytes(
-                client, api=api, headers=headers, path=path, branch=branch
-            )
-            if _event_content_matches(existing_bytes, event):
-                status = 201
-            elif existing_bytes is not None:
-                raise FeedbackConflictError(
-                    "feedback_id was already used for different feedback content"
-                )
-            else:
-                raise ProviderWriteError(
-                    "content_write_failed",
-                    "GitHub feedback event could not be committed",
-                )
-
-    existing_review = await _existing_review_url(
-        client,
-        api=api,
-        headers=headers,
-        repo=repo,
-        branch=branch,
-        base_branch=target.branch,
-    )
-    await _verify_review_branch_scope(
-        client,
-        api=api,
-        headers=headers,
-        base_branch=target.branch,
-        branch=branch,
-        path=path,
-        event=event,
-        review_exists=existing_review is not None,
-    )
-    if existing_review is not None:
-        return {
-            "status": "replay",
-            "provider": "github",
-            "feedback_id": event["feedback"]["id"],
-            "review_url": existing_review if target.expose_links else "",
-            "request_hash": request_hash,
-        }
-
-    rating = int(event["feedback"]["rating"])
-    signed = f"{rating:+d}" if rating else "0"
-    title_page = _review_page_text(event["page_id"], maximum=100)
-    body_page = _review_page_text(event["page_id"])
-    body = (
-        "Automated page-feedback review.\n\n"
-        f"Request commitment: `{request_hash}`\n\n"
-        f"Page: `{body_page}`\n\n"
-        "The reviewed JSON contains no implicit browser, device, account, network, "
-        "referrer, or telemetry fields."
-    )
-    try:
-        status, pull = await _bounded_json(
-            client,
-            "POST",
-            f"{api}/pulls",
-            headers=headers,
-            json_body={
-                "title": f"[Feedback] {signed} · {title_page}",
-                "head": branch,
-                "base": target.branch,
-                "body": body,
-            },
-        )
-    except ProviderWriteError as exc:
-        if exc.code != "provider_transport_failed":
-            raise
-        try:
-            existing_review = await _existing_review_url(
-                client,
-                api=api,
-                headers=headers,
-                repo=repo,
-                branch=branch,
-                base_branch=target.branch,
-            )
-        except ProviderWriteError:
-            raise exc  # ruff: ignore[raise-without-from-inside-except]
-        if existing_review is None:
-            raise exc  # ruff: ignore[verbose-raise]
-        await _verify_review_branch_scope(
-            client,
-            api=api,
-            headers=headers,
-            base_branch=target.branch,
-            branch=branch,
-            path=path,
-            event=event,
-            review_exists=True,
-        )
-        return {
-            "status": "replay",
-            "provider": "github",
-            "feedback_id": event["feedback"]["id"],
-            "review_url": existing_review if target.expose_links else "",
-            "request_hash": request_hash,
-        }
-    if status not in {200, 201}:
-        # A provider race/non-success may still have opened the review. Reconcile
-        # once for every non-success status, not only the common 422 case.
-        existing_review = await _existing_review_url(
-            client,
-            api=api,
-            headers=headers,
-            repo=repo,
-            branch=branch,
-            base_branch=target.branch,
-        )
-        if existing_review is not None:
-            await _verify_review_branch_scope(
-                client,
-                api=api,
-                headers=headers,
-                base_branch=target.branch,
-                branch=branch,
-                path=path,
-                event=event,
-                review_exists=True,
-            )
-            return {
-                "status": "replay",
-                "provider": "github",
-                "feedback_id": event["feedback"]["id"],
-                "review_url": existing_review if target.expose_links else "",
-                "request_hash": request_hash,
-            }
-        raise ProviderWriteError(
-            "review_open_failed", "GitHub feedback pull request could not be opened"
-        )
-    if not isinstance(pull, dict):
-        raise ProviderWriteError(
-            "review_open_invalid", "GitHub feedback pull request response was invalid"
-        )
-    review_url = _review_url(pull.get("html_url"), repo=repo)
-    await _verify_review_branch_scope(
-        client,
-        api=api,
-        headers=headers,
-        base_branch=target.branch,
-        branch=branch,
-        path=path,
-        event=event,
-        review_exists=True,
-    )
-    return {
-        "status": "accepted",
-        "provider": "github",
-        "feedback_id": event["feedback"]["id"],
-        "review_url": review_url if target.expose_links else "",
-        "request_hash": request_hash,
-    }

_page_feedback/_service/_sqlite.py

@@ -1,218 +0,0 @@
-"""Durable SQLite feedback provider with conflict-safe idempotency."""
-
-from __future__ import annotations
-
-import json
-import sqlite3
-import time
-from pathlib import Path
-from typing import Any
-
-from .._contracts import (
-    FeedbackConflictError,
-    canonical_event_bytes,
-    feedback_event_request_hash,
-    normalize_page_revision,
-    normalize_site_id,
-    parse_feedback_event,
-    validate_request_hash,
-)
-
-_SCHEMA = """
-CREATE TABLE IF NOT EXISTS feedback_events (
-    feedback_id TEXT PRIMARY KEY,
-    request_hash TEXT NOT NULL,
-    site_id TEXT NOT NULL,
-    page_id TEXT NOT NULL,
-    event_json TEXT NOT NULL
-) WITHOUT ROWID;
-CREATE INDEX IF NOT EXISTS feedback_events_page_idx
-ON feedback_events(site_id, page_id);
-"""
-
-
-class SQLiteFeedbackStore:
-    """Small transactional event store; no participant/network identity columns."""
-
-    def __init__(self, path: str | Path, *, busy_timeout_ms: int = 5000) -> None:
-        self.path = Path(path)
-        self.busy_timeout_ms = max(100, min(int(busy_timeout_ms), 30_000))
-
-    def _setup_with_retry(self, conn: sqlite3.Connection) -> None:
-        """
-        Establish WAL/schema with bounded retry for first-open races.
-
-        SQLite's busy timeout does not reliably cover every journal-mode transition.
-        Concurrent first connections can therefore observe ``database is locked``
-        before ordinary transactional retry behavior applies. Keep that race inside
-        this initialization boundary rather than surfacing it as a failed feedback
-        submission.
-        """
-        deadline = time.monotonic() + self.busy_timeout_ms / 1000
-        delay = 0.005
-        while True:
-            try:
-                mode = str(
-                    conn.execute("PRAGMA journal_mode").fetchone()[0],
-                ).lower()
-                if mode != "wal":
-                    mode = str(
-                        conn.execute("PRAGMA journal_mode=WAL").fetchone()[0],
-                    ).lower()
-                    if mode != "wal":
-                        raise sqlite3.OperationalError(
-                            "unable to enable WAL journal mode",
-                        )
-                conn.executescript(_SCHEMA)
-                return
-            except sqlite3.OperationalError as exc:  # ruff: ignore[try-except-in-loop]
-                locked = "locked" in str(exc).lower() or "busy" in str(exc).lower()
-                if not locked or time.monotonic() >= deadline:
-                    raise
-                time.sleep(delay)
-                delay = min(delay * 2, 0.1)
-
-    def _connect(self) -> sqlite3.Connection:
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        conn = sqlite3.connect(
-            self.path,
-            timeout=self.busy_timeout_ms / 1000,
-            isolation_level=None,
-        )
-        try:
-            conn.execute(f"PRAGMA busy_timeout={self.busy_timeout_ms}")
-            self._setup_with_retry(conn)
-            conn.execute("PRAGMA synchronous=FULL")
-            return conn
-        except Exception:
-            conn.close()
-            raise
-
-    def put(
-        self, *, feedback_id: str, request_hash: str, event: dict[str, Any]
-    ) -> dict[str, Any]:
-        normalized = parse_feedback_event(event)
-        if feedback_id != normalized["feedback"]["id"]:
-            raise ValueError("feedback_id does not match the durable event")
-        validate_request_hash(request_hash)
-        expected_hash = feedback_event_request_hash(normalized)
-        if request_hash != expected_hash:
-            raise ValueError("request_hash does not match the durable feedback event")
-        event_json = canonical_event_bytes(normalized).decode("utf-8").rstrip("\n")
-        with self._connect() as conn:
-            conn.execute("BEGIN IMMEDIATE")
-            row = conn.execute(
-                "SELECT request_hash, event_json FROM feedback_events WHERE feedback_id = ?",
-                (feedback_id,),
-            ).fetchone()
-            if row is not None:
-                if row[0] != request_hash or row[1] != event_json:
-                    conn.rollback()
-                    raise FeedbackConflictError(
-                        "feedback_id was already used for different feedback content"
-                    )
-                conn.commit()
-                return {
-                    "status": "replay",
-                    "provider": "sqlite",
-                    "feedback_id": feedback_id,
-                    "request_hash": request_hash,
-                }
-            conn.execute(
-                "INSERT INTO feedback_events(feedback_id, request_hash, site_id, page_id, event_json) VALUES(?,?,?,?,?)",
-                (
-                    feedback_id,
-                    request_hash,
-                    normalized["site_id"],
-                    normalized["page_id"],
-                    event_json,
-                ),
-            )
-            conn.commit()
-            return {
-                "status": "accepted",
-                "provider": "sqlite",
-                "feedback_id": feedback_id,
-                "request_hash": request_hash,
-            }
-
-    def events(self, *, site_id: str | None = None) -> list[dict[str, Any]]:
-        if site_id is not None:
-            site_id = normalize_site_id(site_id)
-        with self._connect() as conn:
-            if site_id is None:
-                rows = conn.execute(
-                    "SELECT feedback_id, request_hash, site_id, page_id, event_json "
-                    "FROM feedback_events ORDER BY site_id, page_id, feedback_id"
-                ).fetchall()
-            else:
-                rows = conn.execute(
-                    "SELECT feedback_id, request_hash, site_id, page_id, event_json "
-                    "FROM feedback_events WHERE site_id = ? ORDER BY page_id, feedback_id",
-                    (site_id,),
-                ).fetchall()
-        events: list[dict[str, Any]] = []
-        for stored_id, stored_hash, stored_site, stored_page, raw_json in rows:
-            try:
-                decoded = json.loads(raw_json)
-                event = parse_feedback_event(decoded)
-                canonical = canonical_event_bytes(event).decode("utf-8").rstrip("\n")
-                expected_hash = feedback_event_request_hash(event)
-            except (json.JSONDecodeError, ValueError, RecursionError) as exc:
-                raise sqlite3.DatabaseError(
-                    "stored feedback event is invalid",
-                ) from exc
-            if canonical != raw_json:
-                raise sqlite3.DatabaseError(
-                    "stored feedback event is not canonical",
-                )
-            if (
-                stored_id != event["feedback"]["id"]
-                or stored_site != event["site_id"]
-                or stored_page != event["page_id"]
-                or stored_hash != expected_hash
-            ):
-                raise sqlite3.DatabaseError(
-                    "stored feedback event metadata does not match canonical event content"
-                )
-            events.append(event)
-        return events
-
-    def aggregate(
-        self, *, site_id: str, page_revision: str | None = None
-    ) -> dict[str, dict[str, int]]:
-        site_id = normalize_site_id(site_id)
-        revision_filter: str | None = None
-        if page_revision not in (None, ""):
-            revision_filter = normalize_page_revision(page_revision)
-            if not revision_filter:
-                raise ValueError(
-                    "revision-scoped aggregate requires a non-empty page_revision",
-                )
-        pages: dict[str, dict[str, int]] = {}
-        for event in self.events(site_id=site_id):
-            if (
-                revision_filter is not None
-                and event.get("page_revision", "") != revision_filter
-            ):
-                continue
-            state = pages.setdefault(
-                event["page_id"],
-                {
-                    "count": 0,
-                    "score": 0,
-                    "positive_count": 0,
-                    "negative_count": 0,
-                    "neutral_count": 0,
-                },
-            )
-            rating = int(event["feedback"]["rating"])
-            state["count"] += 1
-            state["score"] += rating
-            if rating > 0:
-                state["positive_count"] += 1
-            elif rating < 0:
-                state["negative_count"] += 1
-            else:
-                state["neutral_count"] += 1
-        return pages

_page_feedback/_service/app.py

@@ -1,690 +0,0 @@
-"""
-Dependency-free ASGI adapter for the generic page-feedback service.
-
-This module intentionally does not depend on FastAPI/Starlette. It is small enough to
-run directly under any ASGI server while the provider-neutral service remains reusable
-inside an existing application. The adapter stores no page-view data and keeps abuse
-control identity separate from durable feedback events.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import hashlib
-import hmac
-import ipaddress
-import json
-import os
-import secrets
-import time
-from collections.abc import Mapping, Sequence
-from typing import Any
-from urllib.parse import urlsplit
-
-from .._contracts import (
-    FeedbackConflictError,
-    FeedbackValidationError,
-    decode_feedback_request,
-    feedback_request_hash,
-)
-from ._config import FeedbackServiceConfig, load_service_config
-from ._core import FeedbackServiceUnavailable, PageFeedbackService
-
-_ENDPOINT = "/v1/feedback"
-_WINDOW_SECONDS = 3600.0
-_MAX_ORIGINS = 64
-_MAX_RATE_IDENTITIES = 10_000
-_MAX_RETRY_KEYS = 50_000
-_MAX_SAME_REQUEST_RETRIES = 3
-_MAX_REQUEST_FRAMES = 256
-_MAX_TRUSTED_PROXY_CIDRS = 64
-_MAX_FORWARDED_HOPS = 32
-_MAX_SECURITY_HEADER_BYTES = 8192
-_RATE_PRUNE_INTERVAL_SECONDS = 60.0
-
-
-class FeedbackASGIConfigError(ValueError):
-    """Raised for unsafe standalone-adapter configuration."""
-
-
-def _origin(value: str) -> str:
-    text = str(value or "").strip()
-    _len = len(text) > 2048  # ruff: ignore[magic-value-comparison]
-    if (
-        not text
-        or _len
-        or any(
-            ord(ch) < 32 or ord(ch) == 127  # ruff: ignore[magic-value-comparison]
-            for ch in text
-        )
-    ):
-        raise FeedbackASGIConfigError(
-            "feedback allowed origin is invalid",
-        )
-    try:
-        parsed = urlsplit(text)
-        port = parsed.port
-    except ValueError as exc:
-        raise FeedbackASGIConfigError(
-            "feedback allowed origin is invalid",
-        ) from exc
-    host = (parsed.hostname or "").lower()
-    local_http = parsed.scheme == "http" and host in {"127.0.0.1", "localhost", "::1"}
-    if parsed.scheme != "https" and not local_http:
-        raise FeedbackASGIConfigError(
-            "feedback allowed origins must use HTTPS or localhost HTTP",
-        )
-    if (
-        not host
-        or parsed.username
-        or parsed.password
-        or parsed.path not in {"", "/"}
-        or parsed.query
-        or parsed.fragment
-    ):
-        raise FeedbackASGIConfigError(
-            "feedback allowed origin must contain only scheme and authority",
-        )
-    if parsed.scheme == "https" and port not in (None, 443):
-        raise FeedbackASGIConfigError(
-            "feedback allowed HTTPS origins must use port 443",
-        )
-    authority = host
-    if ":" in host and not host.startswith("["):
-        authority = f"[{host}]"
-    if (
-        port is not None
-        and not (
-            parsed.scheme == "https"  # lint
-            and port == 443  # ruff: ignore[magic-value-comparison]
-        )
-        and not (
-            parsed.scheme == "http"  # lint
-            and port == 80  # ruff: ignore[magic-value-comparison]
-        )
-    ):
-        authority += f":{port}"
-    return f"{parsed.scheme}://{authority}"
-
-
-def parse_allowed_origins(value: str | Sequence[str] | None) -> tuple[str, ...]:
-    """Normalize an optional exact CORS allowlist; blank means no CORS headers."""
-    if value in (None, ""):
-        return ()
-    if isinstance(value, Sequence) and not isinstance(value, str) and len(value) == 0:
-        return ()
-    if isinstance(value, str):
-        items = [part.strip() for part in value.split(",") if part.strip()]
-    elif isinstance(value, Sequence):
-        items = list(value)
-    else:
-        raise FeedbackASGIConfigError(
-            "feedback allowed origins must be a string or sequence",
-        )
-    if len(items) > _MAX_ORIGINS:
-        raise FeedbackASGIConfigError(
-            "feedback allowed origins contains too many entries",
-        )
-    result: list[str] = []
-    for item in items:
-        normalized = _origin(str(item))
-        if normalized not in result:
-            result.append(normalized)
-    return tuple(result)
-
-
-def _singleton_header(scope: dict[str, Any], name: str) -> str:
-    values: list[str] = []
-    expected = name.lower()
-    headers = scope.get("headers") or []
-    if not isinstance(headers, (list, tuple)):
-        raise FeedbackASGIConfigError("ASGI headers must be a sequence of byte pairs")
-    for item in headers:
-        _len = len(item) != 2  # ruff: ignore[magic-value-comparison]
-        if not isinstance(item, (list, tuple)) or _len:
-            raise FeedbackASGIConfigError("ASGI headers must contain byte pairs")
-        key, value = item
-        if not isinstance(key, bytes) or not isinstance(value, bytes):
-            raise FeedbackASGIConfigError("ASGI header names and values must be bytes")
-        if (
-            len(key) > 256  # ruff: ignore[magic-value-comparison]
-            or len(value) > _MAX_SECURITY_HEADER_BYTES
-            or any(ch in key for ch in (0, 10, 13))
-            or any(ch in value for ch in (0, 10, 13))
-        ):
-            raise FeedbackASGIConfigError("ASGI header framing is invalid")
-        header_name = key.decode("latin-1").lower()
-        header_value = value.decode("latin-1")
-        if header_name == expected:
-            values.append(header_value)
-    if len(values) > 1:
-        raise FeedbackASGIConfigError(f"duplicate {name} header is not allowed")
-    return values[0] if values else ""
-
-
-def parse_trusted_proxy_cidrs(  # ruff: ignore[too-many-branches]
-    value: str | Sequence[str] | None,
-) -> tuple[str, ...]:
-    """
-    Return canonical trusted-proxy CIDRs for X-Forwarded-For processing.
-
-    Forwarded identity is ignored unless the direct ASGI peer belongs to one of
-    these explicitly trusted networks.  Networks must already be canonical (host
-    bits are rejected), which makes deployment policy deterministic and auditable.
-    """
-    if value in (None, ""):
-        return ()
-    if isinstance(value, Sequence) and not isinstance(value, str) and len(value) == 0:
-        return ()
-    if isinstance(value, str):
-        items = [part.strip() for part in value.split(",") if part.strip()]
-    elif isinstance(value, Sequence):
-        items = list(value)
-    else:
-        raise FeedbackASGIConfigError(
-            "trusted proxy CIDRs must be a string or sequence",
-        )
-    if not items or len(items) > _MAX_TRUSTED_PROXY_CIDRS:
-        raise FeedbackASGIConfigError(
-            "trusted proxy CIDRs must contain 1..64 entries",
-        )
-    result: list[str] = []
-    for item in items:
-        if not isinstance(item, str):
-            raise FeedbackASGIConfigError(
-                "trusted proxy CIDR entries must be strings",
-            )
-        text = item.strip()
-        _len = len(text) > 128  # ruff: ignore[magic-value-comparison]
-        if (
-            not text
-            or _len
-            or any(
-                ord(ch) < 33 or ord(ch) > 126  # ruff: ignore[magic-value-comparison]
-                for ch in text
-            )
-        ):
-            raise FeedbackASGIConfigError("trusted proxy CIDR is invalid")
-        try:
-            network = ipaddress.ip_network(text, strict=True)
-        except ValueError as exc:
-            raise FeedbackASGIConfigError("trusted proxy CIDR is invalid") from exc
-        if network.prefixlen == 0:
-            raise FeedbackASGIConfigError(
-                "trusted proxy CIDR must not trust an entire address family"
-            )
-        canonical = network.with_prefixlen
-        if canonical != text:
-            raise FeedbackASGIConfigError(
-                "trusted proxy CIDR must already be canonical",
-            )
-        if canonical not in result:
-            result.append(canonical)
-
-    # Reject a collectively universal trust policy too. A literal /0 is not the
-    # only way to trust an entire address family: two complementary /1 networks
-    # (or several narrower networks) can collapse to the same /0. Treat that as
-    # the same unsafe deployment mistake instead of letting spelling bypass the
-    # guard above. Keep IPv4 and IPv6 separate because collapse_addresses does
-    # not accept mixed address families.
-    for version in (4, 6):
-        networks = [
-            ipaddress.ip_network(item, strict=True)
-            for item in result
-            if ipaddress.ip_network(item, strict=True).version == version
-        ]
-        if networks and any(
-            network.prefixlen == 0 for network in ipaddress.collapse_addresses(networks)
-        ):
-            raise FeedbackASGIConfigError(
-                "trusted proxy CIDRs must not collectively trust an entire address family"
-            )
-    return tuple(result)
-
-
-def _json_bytes(payload: dict[str, Any]) -> bytes:
-    return json.dumps(
-        payload,
-        ensure_ascii=False,
-        separators=(",", ":"),
-    ).encode("utf-8")
-
-
-class FeedbackASGIApp:
-    """Minimal standalone ASGI transport for :class:`PageFeedbackService`."""
-
-    def __init__(
-        self,
-        config: FeedbackServiceConfig,
-        *,
-        allowed_origins: Sequence[str] = (),
-        service: PageFeedbackService | None = None,
-        trusted_proxy_cidrs: Sequence[str] = (),
-    ) -> None:
-        self.config = config
-        self.service = service or PageFeedbackService(config)
-        self.allowed_origins = parse_allowed_origins(allowed_origins)
-        self.trusted_proxy_cidrs = parse_trusted_proxy_cidrs(trusted_proxy_cidrs)
-        self._trusted_proxy_networks = tuple(
-            ipaddress.ip_network(item, strict=True) for item in self.trusted_proxy_cidrs
-        )
-        self._rate_secret = secrets.token_bytes(32)
-        self._rate: dict[str, tuple[int, float]] = {}
-        self._rate_retries: dict[tuple[str, str], tuple[int, float]] = {}
-        self._rate_lock = asyncio.Lock()
-        self._rate_last_prune = 0.0
-
-    def _cors_origin(self, scope: dict[str, Any]) -> str:
-        origin = _singleton_header(scope, "origin").strip()
-        if not origin or not self.allowed_origins:
-            return ""
-        try:
-            normalized = _origin(origin)
-        except FeedbackASGIConfigError:
-            return "!"
-        return normalized if normalized in self.allowed_origins else "!"
-
-    def _is_trusted_proxy(
-        self,
-        address: ipaddress.IPv4Address | ipaddress.IPv6Address,
-    ) -> bool:
-        return any(address in network for network in self._trusted_proxy_networks)
-
-    def _rate_address(  # ruff: ignore[too-many-return-statements]
-        self,
-        scope: dict[str, Any],
-    ) -> str:
-        """
-        Resolve the abuse-control address without trusting caller headers.
-
-        Direct peers are authoritative by default.  ``X-Forwarded-For`` is used
-        only when the immediate peer is explicitly trusted, and the chain is
-        walked from right to left until the first untrusted hop.  Malformed
-        forwarding metadata falls back to the direct peer, which can over-limit
-        but cannot create attacker-controlled identities.
-        """
-        client = scope.get("client")
-        if not isinstance(client, (list, tuple)) or not client:
-            return "unknown"
-        try:
-            peer = ipaddress.ip_address(str(client[0] or ""))
-        except ValueError:
-            return "unknown"
-        if not self._trusted_proxy_networks or not self._is_trusted_proxy(peer):
-            return peer.compressed
-        try:
-            forwarded = _singleton_header(scope, "x-forwarded-for").strip()
-        except FeedbackASGIConfigError:
-            return peer.compressed
-        if not forwarded:
-            return peer.compressed
-        parts = [part.strip() for part in forwarded.split(",")]
-        if (
-            not parts
-            or len(parts) > _MAX_FORWARDED_HOPS
-            or any(not part for part in parts)
-        ):
-            return peer.compressed
-        try:
-            chain = [ipaddress.ip_address(part) for part in parts]
-        except ValueError:
-            return peer.compressed
-        current = peer
-        for candidate in reversed(chain):
-            if not self._is_trusted_proxy(current):
-                break
-            current = candidate
-        return current.compressed
-
-    def _rate_identity(self, scope: dict[str, Any]) -> str:
-        address = self._rate_address(scope)
-        message = f"page-feedback-asgi:v1:submit\0{address}".encode("utf-8", "replace")
-        return hmac.new(self._rate_secret, message, hashlib.sha256).hexdigest()
-
-    def _prune_rate_locked(self, now: float, *, force: bool = False) -> None:
-        if not force and now - self._rate_last_prune < _RATE_PRUNE_INTERVAL_SECONDS:
-            return
-        stale = [
-            key
-            for key, (_count, start) in self._rate.items()
-            if now - start >= _WINDOW_SECONDS
-        ]
-        for key in stale:
-            self._rate.pop(key, None)
-        stale_retries = [
-            key
-            for key, (_count, start) in self._rate_retries.items()
-            if now - start >= _WINDOW_SECONDS
-        ]
-        for key in stale_retries:
-            self._rate_retries.pop(key, None)
-        self._rate_last_prune = now
-
-    async def _consume_rate(self, identity: str, request_hash: str) -> bool:
-        """
-        Admit a new event or a bounded exact retry without tracking people.
-
-        The main quota counts distinct request commitments. Once a commitment has
-        been admitted, up to ``_MAX_SAME_REQUEST_RETRIES`` exact retries may pass
-        without consuming another new-event slot. This preserves broken-pipe
-        recovery even at a 1/hour quota while keeping retries bounded. The cache
-        contains only the HMAC abuse identity plus a request commitment.
-        """
-        now = time.monotonic()
-        async with self._rate_lock:
-            self._prune_rate_locked(now)
-
-            retry_key = (identity, request_hash)
-            retry = self._rate_retries.get(retry_key)
-            if retry is not None:
-                retry_count, retry_start = retry
-                if retry_count >= _MAX_SAME_REQUEST_RETRIES:
-                    return False
-                self._rate_retries[retry_key] = (retry_count + 1, retry_start)
-                return True
-
-            if (
-                identity not in self._rate and len(self._rate) >= _MAX_RATE_IDENTITIES
-            ) or len(self._rate_retries) >= _MAX_RETRY_KEYS:
-                self._prune_rate_locked(now, force=True)
-            if identity not in self._rate and len(self._rate) >= _MAX_RATE_IDENTITIES:
-                return False
-            if len(self._rate_retries) >= _MAX_RETRY_KEYS:
-                # Fail closed rather than allowing attacker-controlled commitment
-                # cardinality to grow process memory without bound.
-                return False
-
-            count, start = self._rate.get(identity, (0, now))
-            if now - start >= _WINDOW_SECONDS:
-                count, start = 0, now
-            if count >= self.config.rate_limit_per_hour:
-                return False
-            self._rate[identity] = (count + 1, start)
-            self._rate_retries[retry_key] = (0, now)
-            return True
-
-    async def _send_json(
-        self,
-        send,
-        status: int,
-        payload: dict[str, Any],
-        *,
-        cors_origin: str = "",
-        extra_headers: Sequence[tuple[bytes, bytes]] = (),
-    ) -> None:
-        body = (
-            b""
-            if status == 204  # ruff: ignore[magic-value-comparison]
-            else _json_bytes(payload)
-        )
-        headers = [
-            (b"cache-control", b"no-store"),
-            (b"content-length", str(len(body)).encode("ascii")),
-            (b"x-content-type-options", b"nosniff"),
-        ]
-        if status != 204:  # ruff: ignore[magic-value-comparison]
-            headers.insert(0, (b"content-type", b"application/json; charset=utf-8"))
-        if cors_origin and cors_origin != "!":
-            headers.extend(
-                [
-                    (b"access-control-allow-origin", cors_origin.encode("ascii")),
-                    (b"vary", b"Origin"),
-                ]
-            )
-        headers.extend(extra_headers)
-        try:
-            await send(
-                {"type": "http.response.start", "status": status, "headers": headers}
-            )
-            await send({"type": "http.response.body", "body": body})
-        except OSError:
-            # ASGI servers are allowed to raise OSError when the client disconnects
-            # after a durable write. The browser retry uses the same event id, so
-            # suppressing this transport-only broken pipe preserves exact replay.
-            return
-
-    async def _read_body(self, receive) -> bytes:
-        body = bytearray()
-        messages = 0
-        while True:
-            message = await receive()
-            messages += 1
-            if messages > _MAX_REQUEST_FRAMES:
-                raise OverflowError("request is too fragmented")
-            if message.get("type") == "http.disconnect":
-                raise ConnectionError("client disconnected")
-            if message.get("type") != "http.request":
-                continue
-            raw_chunk = message.get("body", b"")
-            if not isinstance(raw_chunk, bytes):
-                raise ValueError(  # ruff: ignore[type-check-without-type-error]
-                    "request body frame must contain bytes",
-                )
-            if len(body) + len(raw_chunk) > self.config.max_body_bytes:
-                raise OverflowError("request too large")
-            body.extend(raw_chunk)
-            if not message.get("more_body", False):
-                return bytes(body)
-
-    async def __call__(  # ruff: ignore[too-many-branches, too-many-return-statements]
-        self,
-        scope,
-        receive,
-        send,
-    ) -> None:
-        if scope.get("type") != "http":
-            return
-        path = str(scope.get("path") or "")
-        method = str(scope.get("method") or "GET").upper()
-        try:
-            # These headers are security-relevant singletons. Ambiguous duplicates
-            # are rejected instead of relying on first/last-value proxy behavior.
-            _singleton_header(scope, "content-type")
-            declared_length = _singleton_header(scope, "content-length")
-            cors_origin = self._cors_origin(scope)
-        except FeedbackASGIConfigError:
-            await self._send_json(send, 400, {"detail": "Ambiguous request headers."})
-            return
-        if cors_origin == "!":
-            await self._send_json(
-                send, 403, {"detail": "Origin is not allowed for page feedback."}
-            )
-            return
-        if path != _ENDPOINT:
-            await self._send_json(
-                send,
-                404,
-                {"detail": "Not found."},
-                cors_origin=cors_origin,
-            )
-            return
-        if method == "OPTIONS":
-            if not cors_origin:
-                await self._send_json(
-                    send,
-                    204,
-                    {},
-                    cors_origin="",
-                )
-                return
-            await self._send_json(
-                send,
-                204,
-                {},
-                cors_origin=cors_origin,
-                extra_headers=(
-                    (b"access-control-allow-methods", b"POST, OPTIONS"),
-                    (b"access-control-allow-headers", b"Content-Type, Accept"),
-                    (b"access-control-max-age", b"600"),
-                ),
-            )
-            return
-        if method != "POST":
-            await self._send_json(
-                send, 405, {"detail": "Method not allowed."}, cors_origin=cors_origin
-            )
-            return
-        declared_bytes: int | None = None
-        if declared_length:
-            if not declared_length.isascii() or not declared_length.isdecimal():
-                await self._send_json(
-                    send,
-                    400,
-                    {"detail": "Invalid Content-Length."},
-                    cors_origin=cors_origin,
-                )
-                return
-            try:
-                declared_bytes = int(declared_length, 10)
-            except ValueError:
-                await self._send_json(
-                    send,
-                    400,
-                    {"detail": "Invalid Content-Length."},
-                    cors_origin=cors_origin,
-                )
-                return
-            if declared_bytes < 0:
-                await self._send_json(
-                    send,
-                    400,
-                    {"detail": "Invalid Content-Length."},
-                    cors_origin=cors_origin,
-                )
-                return
-            if declared_bytes > self.config.max_body_bytes:
-                await self._send_json(
-                    send,
-                    413,
-                    {
-                        "detail": (
-                            "Page feedback request exceeds the configured safety limit."
-                        ),
-                    },
-                    cors_origin=cors_origin,
-                )
-                return
-        media_type = (
-            _singleton_header(
-                scope,
-                "content-type",
-            )
-            .split(";", 1)[0]
-            .strip()
-            .lower()
-        )
-        if media_type != "application/json":
-            await self._send_json(
-                send,
-                415,
-                {"detail": "Page feedback requires application/json."},
-                cors_origin=cors_origin,
-            )
-            return
-        try:
-            raw = await self._read_body(receive)
-        except OverflowError:
-            await self._send_json(
-                send,
-                413,
-                {
-                    "detail": (
-                        "Page feedback request exceeds the configured safety limit."
-                    ),
-                },
-                cors_origin=cors_origin,
-            )
-            return
-        except ConnectionError:
-            return
-        except ValueError:
-            await self._send_json(
-                send,
-                400,
-                {"detail": "Malformed page feedback request framing."},
-                cors_origin=cors_origin,
-            )
-            return
-        if declared_bytes is not None and declared_bytes != len(raw):
-            await self._send_json(
-                send,
-                400,
-                {"detail": "Content-Length does not match the request body."},
-                cors_origin=cors_origin,
-            )
-            return
-        try:
-            validated = decode_feedback_request(raw)
-            validated = self.service.validate_request(validated)
-        except FeedbackValidationError as exc:
-            await self._send_json(
-                send, 422, {"detail": str(exc)}, cors_origin=cors_origin
-            )
-            return
-        request_hash = feedback_request_hash(validated)
-        if not await self._consume_rate(self._rate_identity(scope), request_hash):
-            await self._send_json(
-                send,
-                429,
-                {"detail": "Rate limit exceeded for page feedback submissions."},
-                cors_origin=cors_origin,
-                extra_headers=((b"retry-after", b"3600"),),
-            )
-            return
-        try:
-            receipt = await self.service.submit(validated)
-        except FeedbackConflictError as exc:
-            await self._send_json(
-                send, 409, {"detail": str(exc)}, cors_origin=cors_origin
-            )
-            return
-        except FeedbackValidationError as exc:
-            await self._send_json(
-                send, 422, {"detail": str(exc)}, cors_origin=cors_origin
-            )
-            return
-        except FeedbackServiceUnavailable:
-            await self._send_json(
-                send,
-                503,
-                {"detail": "Page feedback storage is temporarily unavailable."},
-                cors_origin=cors_origin,
-            )
-            return
-        await self._send_json(send, 202, receipt, cors_origin=cors_origin)
-
-
-def create_app(
-    env: Mapping[str, str] | None = None,
-    *,
-    allowed_origins: str | Sequence[str] | None = None,
-) -> FeedbackASGIApp:
-    """Create a standalone ASGI app from explicit/process environment config."""
-    source = os.environ if env is None else env
-    config = load_service_config(source)
-    origins = (
-        source.get("FEEDBACK_ALLOWED_ORIGINS", "")
-        if allowed_origins is None
-        else allowed_origins
-    )
-    trusted_proxies = parse_trusted_proxy_cidrs(
-        source.get("FEEDBACK_TRUSTED_PROXY_CIDRS", "")
-    )
-    return FeedbackASGIApp(
-        config,
-        allowed_origins=parse_allowed_origins(origins),
-        service=PageFeedbackService(config, credential_env=source),
-        trusted_proxy_cidrs=trusted_proxies,
-    )
-
-
-app = create_app()
-
-__all__ = [
-    "FeedbackASGIApp",
-    "FeedbackASGIConfigError",
-    "app",
-    "create_app",
-    "parse_allowed_origins",
-    "parse_trusted_proxy_cidrs",
-]