_utils/_shared_logic.py

# scikitplot/_externals/_sphinx_ext/_sphinx_ai_assistant/_hf_spaces_proxy/_utils/_shared_logic.py
#
# flake8: noqa: D213
#
# Authors: The scikit-plots developers
# SPDX-License-Identifier: BSD-3-Clause

# _shared_logic.py  v7.0.0
#
# Single source of truth for shared constants, pure helper functions, and
# type aliases used by the deployed proxy (_hf_spaces_proxy/app.py) and the
# local development proxy (dev_proxy.py).
#
# Import discipline
# -----------------
# Only the Python standard library is imported here.  httpx, fastapi, and
# torch are NOT imported so this module can be sourced by stdlib-only tools
# (dev_proxy) and tested in isolation without any network or GPU environment.
#
# Routing paths (v6.0.0)
# ----------------------
# Three ordered routing paths — each with its own configurable read timeout:
#
#   Path 1 — BACKEND_URL set (explicit override)
#     Forward to BACKEND_URL.  Only BACKEND_AUTH_TOKEN may be attached by callers.
#     Read timeout: proxy_timeout kwarg (env: PROXY_TIMEOUT, default 600 s).
#
#   Path 2 — Model namespace in HF_SPACES_MODEL_NAMESPACES
#     Model owner (e.g. "scikit-plots") matches a custom namespace.
#     Forward to HF_SPACES_MODEL_URL (the ai-model HF Space, CPU inference).
#     These models have no HF Inference Provider → direct HF API returns 404/503.
#     Read timeout: path2_read_timeout kwarg (env: PATH2_TIMEOUT, default 600 s).
#     CPU inference on a 7B model takes 4-5 minutes; 600 s gives safe headroom.
#
#   Path 3 — Standard HF Inference API (default)
#     Model has a registered HF Inference Provider (openai/*, Qwen/*, etc.).
#     Forward to HF_BASE/{model}/v1/chat/completions with HF_TOKEN.
#     Read timeout: path3_read_timeout kwarg (env: PATH3_TIMEOUT, default 120 s).
#     HF Serverless API (GPU-backed) normally responds within 30-90 s.
#
# Breaking changes v4.0.0 → v5.0.0
# ----------------------------------
# + DEFAULT_PROXY_TIMEOUT raised from 120 s to 600 s.
#   Root cause: 120 s was shorter than the 4-5 min CPU inference on the
#   ai-model HF Space, causing every request to return a network error.
# + DEFAULT_PATH2_READ_TIMEOUT added (600 s) — ai-model space per-path timeout.
# + DEFAULT_PATH3_READ_TIMEOUT added (120 s) — HF API per-path timeout.
# + _resolve_upstream_url now accepts path2_read_timeout, path3_read_timeout,
#   and proxy_timeout keyword-only parameters.
# + _resolve_upstream_url return type changed from tuple[str, dict] to
#   tuple[str, dict, float] — the third element is the per-path read timeout.
#   Callers must unpack all three values.
# + load_proxy_env extended with path2_read_timeout and path3_read_timeout.
#
# Breaking changes v5.0.0 → v6.0.0
# ----------------------------------
# + DEFAULT_HF_BASE changed from ``https://api-inference.huggingface.co/models``
#   to ``https://router.huggingface.co``.
#   Root cause: api-inference.huggingface.co was DNS-unresolvable ([Errno -5]
#   EAI_NODATA / EAI_NONAME) from within HF Docker Spaces.
#   router.huggingface.co is the current HF Inference Providers endpoint and
#   resolves correctly in all deployment environments.
#   Callers who hard-code ``HF_BASE`` to the old hostname must migrate to
#   the new router URL.
#
# New in v6.1.0 — Three-type HF token system
# -------------------------------------------
# + ``HFTokenType`` literal type alias added: ``"fine-grained" | "read" |
#   ``"write" | "unknown"``.  Maps directly to the three token types exposed
#   in HF Settings → Tokens.
# + ``HF_TOKEN_TYPE_*`` string constants and ``HF_INFERENCE_TOKEN_TYPES`` /
#   ``HF_WRITE_TOKEN_TYPES`` frozensets added for type-safe comparisons.
# + ``_classify_token_type()`` — classify a token by explicit env-var
#   declaration (``HF_TOKEN_TYPE``, ``HF_WRITE_TOKEN_TYPE``) with a length-
#   based heuristic fallback.
# + ``_token_suitable_for_inference()`` / ``_token_suitable_for_writes()``
#   predicates for principle-of-least-privilege validation.
# + ``_validate_token_config()`` — returns actionable WARNING / ERROR strings
#   for token-type mismatches detected at startup.
# + ``_token_log_fragment()`` gains an optional ``token_type`` parameter so
#   log lines include the token type (e.g. ``hf_abcde...1234 (read)``).
# + ``load_proxy_env()`` extended with ``hf_token_type`` and
#   ``hf_write_token_type`` keys read from the matching env vars.
# + ``_safe_float`` added to ``__all__`` (was importable but unadvertised).

"""
Shared utilities for the sphinx-ai-assistant proxy solutions.

This module provides pure, stateless helper functions and typed constants
that are common to all server-side proxy implementations.  It has **no**
runtime dependencies beyond the Python standard library.

Public API:

PROXY_VERSION : str
    Proxy release version string.
DEFAULT_HF_BASE : str
    HuggingFace Serverless Inference API base URL.
DEFAULT_MODEL : str
    Fallback model ID when the request body omits ``model``.
DEFAULT_PROXY_TIMEOUT : int
    Global upstream read timeout in seconds (Path 1 / backward-compat).
DEFAULT_PATH2_READ_TIMEOUT : float
    Per-path read timeout for Path 2 (ai-model space, CPU inference).
DEFAULT_PATH3_READ_TIMEOUT : float
    Per-path read timeout for Path 3 (HF Serverless Inference API).
DEFAULT_MAX_BODY_BYTES : int
    Maximum accepted request body size.
DEFAULT_HF_SPACES_MODEL_URL : str
    Default URL for the custom ai-model HF Space (Path 2).
DEFAULT_HF_SPACES_MODEL_NAMESPACES : tuple[str, ...]
    Default model owner namespaces routed to the model Space (Path 2).
_safe_int : callable
    Parse an integer environment variable with a safe fallback.
_parse_model : callable
    Extract the ``model`` field from a raw JSON request body.
_is_custom_model_namespace : callable
    Return True when a model's owner namespace is in the custom list.
_build_cors_headers : callable
    Return the CORS response-header mapping.
_token_log_fragment : callable
    Produce a safely-truncated token string for log output.
_resolve_upstream_url : callable
    Centralised three-path routing: choose upstream URL, auth headers,
    and per-path read timeout.
_validate_env : callable
    Fail-fast startup check with actionable error messages.
load_proxy_env : callable
    Read all proxy-relevant environment variables into a typed dict.

Notes
-----
**Developer note** — All functions are pure (no side effects, no I/O).
Tests can import this module without a running event loop or any network.
The proxy (FastAPI / asyncio) and dev_proxy (stdlib HTTPServer) both import
from here so that routing and CORS logic are *never* duplicated.

**Breaking change v5.0.0** — ``_resolve_upstream_url`` now returns a
3-tuple ``(url, headers, read_timeout_s: float)`` instead of the previous
2-tuple ``(url, headers)``.  All callers must unpack the third element or
the per-path timeout falls through to the old flat-timeout behaviour.

**Breaking change v6.0.0** — :data:`DEFAULT_HF_BASE` migrated from
``https://api-inference.huggingface.co/models`` to
``https://router.huggingface.co``.  The old hostname was DNS-unresolvable
([Errno -5] EAI_NONAME) from within HF Docker Spaces.  Deployments that
override ``HF_BASE`` to the legacy hostname must update their configuration.

**Security note** — :func:`_token_log_fragment` ensures the full API token
never appears in log output.  Never widen the exposed fragment beyond the
current 8+4 character window without reviewing log-aggregation policy first.

**Versioning note** — Bump :data:`PROXY_VERSION` on every breaking change so
deployed Spaces and log aggregators can correlate errors to a specific release.
"""

from __future__ import annotations

import ipaddress
import json
import logging
import os
import re
from typing import Any, Literal
from urllib.parse import urlsplit

try:
    from ._telemetry import sanitize_log_text
except ImportError:  # standalone HF Space deployment
    from _utils._telemetry import sanitize_log_text


logger = logging.getLogger(__name__)

__all__ = [  # noqa: RUF022
    # Version
    "PROXY_VERSION",
    # Constants — routing / timeout
    "DEFAULT_HF_BASE",
    "DEFAULT_HF_PROVIDER_MODELS",
    "DEFAULT_HF_SPACES_MODEL_NAMESPACES",
    "DEFAULT_HF_SPACES_MODEL_URL",
    "DEFAULT_MAX_BODY_BYTES",
    "DEFAULT_MODEL",
    "DEFAULT_PATH2_READ_TIMEOUT",
    "DEFAULT_PATH3_READ_TIMEOUT",
    "DEFAULT_PROXY_TIMEOUT",
    # Constants — token type system (v6.1.0)
    "HFTokenType",
    "HF_TOKEN_TYPE_FINE_GRAINED",
    "HF_TOKEN_TYPE_READ",
    "HF_TOKEN_TYPE_WRITE",
    "HF_TOKEN_TYPE_UNKNOWN",
    "HF_INFERENCE_TOKEN_TYPES",
    "HF_WRITE_TOKEN_TYPES",
    # Helpers — general
    "_build_cors_headers",
    "_is_custom_model_namespace",
    "_parse_model",
    "_safe_float",
    "_safe_int",
    "_token_log_fragment",
    # Privacy / log-redaction (v6.2.0)
    "_REDACT_PATTERNS",
    "_RedactingFilter",
    "_mask_ip",
    # Helpers — token type system (v6.1.0)
    "_classify_token_type",
    "_token_suitable_for_inference",
    "_token_suitable_for_writes",
    "_validate_token_config",
    # Helpers — routing / env
    "_resolve_upstream_url",
    "_validate_credential_destination",
    "_validate_env",
    "load_proxy_env",
]


# ─────────────────────────────────────────────────────────────────────────────
# Module-level constants
# ─────────────────────────────────────────────────────────────────────────────

#: Proxy release version — bump on every breaking change.
PROXY_VERSION: str = "7.4.0"

#: HuggingFace Inference Providers router base URL (no trailing slash).
#: Only used for Path 3 (standard provider models) when ``BACKEND_URL`` is
#: empty and the model namespace is not in ``HF_SPACES_MODEL_NAMESPACES``.
#:
#: Migrated from ``https://api-inference.huggingface.co/models`` (v5.0.0) to
#: ``https://router.huggingface.co`` (v6.0.0).
#: Root cause: api-inference.huggingface.co was DNS-unresolvable ([Errno -5]
#: EAI_NODATA / EAI_NONAME) from within HF Docker Spaces; the router hostname
#: resolves correctly and is the current HF Inference Providers endpoint.
DEFAULT_HF_BASE: str = "https://router.huggingface.co"

#: Public Hugging Face Inference Provider models advertised by the bundled
#: example configuration. Keep this default synchronized with the Cloudflare
#: Worker so both bundled proxies accept the same public model choices.
#: Operators can replace the exact set with ``ALLOWED_MODELS``.
DEFAULT_HF_PROVIDER_MODELS: tuple[str, ...] = (
    "Qwen/Qwen2.5-Coder-7B-Instruct",
    "Qwen/Qwen2.5-Coder-32B-Instruct",
    "openai/gpt-oss-20b",
)

#: Fallback model ID when the request body omits the ``model`` field.
#: Must have a registered HF Inference Provider for Path 3.
DEFAULT_MODEL: str = "scikit-plots/Qwen2.5-Coder-7B-Instruct"

#: Global upstream read timeout in seconds (used for Path 1 / backward compat).
#:
#: Raised from 120 s (v4.0.0) to 600 s (v5.0.0).
#:
#: Root cause of the increase: the ai-model HF Space runs a 7B model on CPU
#: basic hardware.  Cold-start inference (model loading + generation) takes
#: 4-5 minutes.  The 120 s ceiling caused every request to the ai-model Space
#: to return ``httpx.ReadTimeout``, which the browser reported as
#: "Sorry, something went wrong: network error".
DEFAULT_PROXY_TIMEOUT: int = 600

#: Per-path read timeout for Path 2 (ai-model HF Space, CPU inference).
#:
#: CPU inference on a 7B model takes 4-5 minutes.  600 s gives 1 minute of
#: additional headroom for cold-start model loading (~50 s tokenizer +
#: ~50 s model load + ~4.5 min generation on the first request).
DEFAULT_PATH2_READ_TIMEOUT: float = 600.0

#: Per-path read timeout for Path 3 (HF Serverless Inference API).
#:
#: The HF Serverless API runs inference on GPU hardware.  Most responses
#: arrive within 30-90 s.  120 s gives a comfortable margin.
DEFAULT_PATH3_READ_TIMEOUT: float = 120.0

#: Maximum accepted request body size in bytes (10 MiB).
#: Prevents memory exhaustion from maliciously oversized POST bodies.
DEFAULT_MAX_BODY_BYTES: int = 10 * 1024 * 1024  # 10 MiB

#: Default URL for the custom ai-model HF Space (Path 2).
#: Requests for models whose namespace is in ``DEFAULT_HF_SPACES_MODEL_NAMESPACES``
#: are forwarded here instead of the HF Serverless Inference API.
#: Overridable via the ``HF_SPACES_MODEL_URL`` environment variable.
DEFAULT_HF_SPACES_MODEL_URL: str = (
    "https://scikit-plots-ai-model.hf.space/v1/chat/completions"
)

#: Default model owner namespaces routed to :data:`DEFAULT_HF_SPACES_MODEL_URL`.
#: Models whose owner (the part before ``/``) matches any entry in this tuple
#: are routed to the ai-model Space (Path 2) rather than the HF API (Path 3).
#: Overridable via the ``HF_SPACES_MODEL_NAMESPACES`` environment variable.
DEFAULT_HF_SPACES_MODEL_NAMESPACES: tuple[str, ...] = ("scikit-plots",)


# ─────────────────────────────────────────────────────────────────────────────
# HuggingFace token type system  (v6.1.0)
# ─────────────────────────────────────────────────────────────────────────────
#
# HuggingFace exposes exactly three token types in
# https://huggingface.co/settings/tokens:
#
#   ① Fine-grained   — New-style token.  Permissions set at creation time:
#                      choose any combination of per-repo access levels and
#                      API capabilities.  Recommended for production because
#                      each token carries only the minimum required scope.
#
#   ② Read (classic) — Legacy read-only token.  Grants read access to all
#                      public repos and any private repos you can access.
#                      Always includes the Serverless Inference API capability.
#                      Cannot push commits or create repos.
#
#   ③ Write (classic)— Legacy read+write token.  All read permissions plus
#                      the ability to push commits, create repos, manage
#                      members, etc.  Over-privileged for inference-only use.
#
# Mapping to proxy env vars
# ─────────────────────────
#   HF_TOKEN       — inference token (Path 2 private Space + Path 3 HF API).
#                    Best practice: fine-grained with inference-api scope only,
#                    OR classic read.  Never use a write token here.
#
#   HF_DATASET_TOKEN — preferred dataset-persistence token. Best practice:
#                      fine-grained scoped to ONE dataset repo. Classic Write
#                      also works; classic Read never does.
#   HF_WRITE_TOKEN   — historical alias for HF_DATASET_TOKEN.
#
# Optional type-declaration env vars (Space → Settings → Repository secrets):
#   HF_TOKEN_TYPE         = fine-grained | read | write (default: auto-detect)
#   HF_DATASET_TOKEN_TYPE = fine-grained | read | write (preferred)
#   HF_WRITE_TOKEN_TYPE   = fine-grained | read | write (legacy alias)
#
# ─────────────────────────────────────────────────────────────────────────────

#: Literal type for HuggingFace token type labels.
#: Use as type annotation and for exhaustive ``isinstance``-free comparisons.
HFTokenType = Literal["fine-grained", "read", "write", "unknown"]

#: New-style fine-grained HF token.  Permissions defined at creation time.
#: Declare via env var: ``HF_TOKEN_TYPE=fine-grained``.
HF_TOKEN_TYPE_FINE_GRAINED: str = "fine-grained"  # noqa: S105

#: Classic HF read token.  Read + Inference API; no write capability.
#: Declare via env var: ``HF_TOKEN_TYPE=read``.
HF_TOKEN_TYPE_READ: str = "read"  # noqa: S105

#: Classic HF write token.  All read permissions + repo push capability.
#: Declare via env var: ``HF_TOKEN_TYPE=write`` or ``HF_WRITE_TOKEN_TYPE=write``.
HF_TOKEN_TYPE_WRITE: str = "write"  # noqa: S105

#: Sentinel: token type not declared and could not be inferred.
#: Runtime operations are not blocked, but :func:`_validate_token_config` omits
#: least-privilege warnings because the type is unknown.
HF_TOKEN_TYPE_UNKNOWN: str = "unknown"  # noqa: S105

#: Token types that are appropriate for HF Serverless Inference API calls
#: (Path 3) and private HF Space access (Path 2).
#:
#: Classic write tokens ARE technically capable of inference (write ⊇ read),
#: but are excluded from this set so :func:`_validate_token_config` can emit
#: a startup warning when a write token is used where a read / fine-grained
#: token is the correct choice.  The ``"unknown"`` sentinel is included so
#: that un-declared tokens do not trigger false-positive warnings.
HF_INFERENCE_TOKEN_TYPES: frozenset[str] = frozenset(
    {
        HF_TOKEN_TYPE_FINE_GRAINED,
        HF_TOKEN_TYPE_READ,
        HF_TOKEN_TYPE_UNKNOWN,
    }
)

#: Token types that can push commits to HuggingFace repos and datasets.
#:
#: Classic read tokens **cannot** write — any ``HfApi.create_commit`` call
#: returns HTTP 403 / 401.  ``"unknown"`` is excluded so that
#: :func:`_validate_token_config` can flag a read token configured as the write
#: token as a hard error rather than silently failing at request time.
HF_WRITE_TOKEN_TYPES: frozenset[str] = frozenset(
    {
        HF_TOKEN_TYPE_FINE_GRAINED,
        HF_TOKEN_TYPE_WRITE,
    }
)


# ─────────────────────────────────────────────────────────────────────────────
# Pure helper functions
# ─────────────────────────────────────────────────────────────────────────────


def _safe_int(value: str | None, default: int) -> int:
    """
    Parse *value* as an integer, returning *default* on any failure.

    Parameters
    ----------
    value : str or None
        String to parse.  Typically the raw value of an environment variable
        (may be ``None`` when the variable is absent).
    default : int
        Returned when *value* is ``None``, empty, or cannot be converted.

    Returns
    -------
    int
        Parsed integer, or *default* on any ``ValueError`` / ``TypeError``.

    Notes
    -----
    **Developer note** — This function is intentionally never-raise.
    A misconfigured ``PROXY_TIMEOUT`` or ``MAX_BODY_BYTES`` must not prevent
    the proxy from starting — the safe default is better than a crash.

    Examples
    --------
    >>> _safe_int("120", 60)
    120
    >>> _safe_int("not-a-number", 60)
    60
    >>> _safe_int(None, 60)
    60
    >>> _safe_int("", 60)
    60
    """
    if value is None:
        return default
    try:
        return int(value)
    except (ValueError, TypeError):
        return default


def _safe_float(value: str | None, default: float) -> float:
    """
    Parse *value* as a float, returning *default* on any failure.

    Parameters
    ----------
    value : str or None
        String to parse.  Typically the raw value of an environment variable.
    default : float
        Returned when *value* is ``None``, empty, or cannot be converted.

    Returns
    -------
    float
        Parsed float, or *default* on any ``ValueError`` / ``TypeError``.

    Notes
    -----
    **Developer note** — Like :func:`_safe_int`, this is intentionally
    never-raise.  A misconfigured ``PATH2_TIMEOUT`` or ``PATH3_TIMEOUT``
    must not crash the proxy at startup.

    Examples
    --------
    >>> _safe_float("600.0", 120.0)
    600.0
    >>> _safe_float("bad", 120.0)
    120.0
    >>> _safe_float(None, 120.0)
    120.0
    """
    if value is None:
        return default
    try:
        return float(value)
    except (ValueError, TypeError):
        return default


def _parse_model(body: bytes, default: str = DEFAULT_MODEL) -> str:
    """
    Extract the ``model`` field from a raw JSON request body.

    Parameters
    ----------
    body : bytes
        Raw HTTP request body forwarded from the browser.  Expected to be
        valid JSON but the function never raises on malformed input.
    default : str, optional
        Fallback model ID when the field is absent or the body cannot be
        decoded.  Defaults to :data:`DEFAULT_MODEL`.

    Returns
    -------
    str
        The ``model`` value from the body, or *default* if the field is
        absent, empty, or the body is not valid JSON.

    Notes
    -----
    **Developer note** — This function is intentionally never-raise.
    A malformed body must not crash the proxy; the upstream model backend
    will return a meaningful error that the browser can display.

    Examples
    --------
    >>> _parse_model(b'{"model": "Qwen/Qwen2.5-Coder-7B-Instruct"}')
    'Qwen/Qwen2.5-Coder-7B-Instruct'
    >>> _parse_model(b"{}")
    'scikit-plots/Qwen2.5-Coder-7B-Instruct'
    >>> _parse_model(b"not-json")
    'scikit-plots/Qwen2.5-Coder-7B-Instruct'
    >>> _parse_model(b'{"model": "  "}')
    'scikit-plots/Qwen2.5-Coder-7B-Instruct'
    """
    try:
        data: Any = json.loads(body)
        candidate = str(data.get("model", "")).strip()
        return candidate or default
    except (json.JSONDecodeError, ValueError, AttributeError, TypeError):
        return default


def _is_custom_model_namespace(
    model: str,
    namespaces: tuple[str, ...] | list[str],
) -> bool:
    """
    Return ``True`` when the model owner namespace is in *namespaces*.

    The owner is the portion of the model ID before the first ``/``.
    An optional HF Router variant suffix (e.g. ``:fastest``) is stripped
    before comparison so ``"scikit-plots/Qwen2.5-Coder-7B-Instruct:fastest"``
    is correctly identified as belonging to the ``"scikit-plots"`` namespace.

    Parameters
    ----------
    model : str
        Model ID string, e.g. ``"scikit-plots/Qwen2.5-Coder-7B-Instruct"``
        or ``"openai/gpt-oss-20b:fastest"``.
    namespaces : tuple[str, ...] or list[str]
        Iterable of owner namespace strings to match against (case-insensitive).
        Typically :data:`DEFAULT_HF_SPACES_MODEL_NAMESPACES` or parsed from
        the ``HF_SPACES_MODEL_NAMESPACES`` environment variable.

    Returns
    -------
    bool
        ``True`` when the model owner is in *namespaces*, ``False`` otherwise.

    Notes
    -----
    **Developer note** — Comparison is case-insensitive and strips leading /
    trailing whitespace from both the model owner and each namespace entry.
    A model string without a ``/`` separator (i.e. no namespace component)
    always returns ``False``; such IDs are routed to Path 3 (HF Inference API).

    Examples
    --------
    >>> _is_custom_model_namespace(
    ...     "scikit-plots/Qwen2.5-Coder-7B-Instruct",
    ...     ("scikit-plots",),
    ... )
    True
    >>> _is_custom_model_namespace(
    ...     "scikit-plots/Qwen2.5-Coder-7B-Instruct:fastest",
    ...     ("scikit-plots",),
    ... )
    True
    >>> _is_custom_model_namespace("openai/gpt-oss-20b", ("scikit-plots",))
    False
    >>> _is_custom_model_namespace("no-slash-model", ("scikit-plots",))
    False
    """
    base = model.split(":", maxsplit=1)[0].strip()
    if not base or "/" not in base:
        return False
    owner = base.split("/", 1)[0].lower().strip()
    normalised = {ns.lower().strip() for ns in namespaces if ns.strip()}
    return owner in normalised


def _build_cors_headers(allowed_origin: str = "*") -> dict[str, str]:
    """
    Return the standard CORS response-header mapping.

    Parameters
    ----------
    allowed_origin : str, optional
        Value for the ``Access-Control-Allow-Origin`` header.
        Defaults to ``"*"`` (allow all origins).

    Returns
    -------
    dict[str, str]
        CORS response headers.

    Examples
    --------
    >>> headers = _build_cors_headers()
    >>> headers["Access-Control-Allow-Origin"]
    '*'
    """
    return {
        "Access-Control-Allow-Origin": allowed_origin,
        "Access-Control-Allow-Methods": "POST, OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type",
    }


def _token_log_fragment(token: str, token_type: str = "") -> str:
    """Return non-secret token configuration state for legacy log call sites.

    The historical implementation exposed an 8+4 character credential
    fragment.  Run 5 deliberately removes that behavior: partial credentials
    are still credentials and may become identifying/correlatable in retained
    logs.  Keep the helper name for source compatibility, but return only
    presence and optional type metadata.
    """
    if not token:
        return "<not-set>"
    label = str(token_type or "").strip().lower()
    return f"<set> ({label})" if label and label != "unknown" else "<set>"


# ─────────────────────────────────────────────────────────────────────────────
# Privacy / log-redaction helpers  (v6.2.0)
# ─────────────────────────────────────────────────────────────────────────────
#
# Design rationale
# ----------------
# Two complementary layers protect PII in log output:
#
#   Layer 1 — call-site masking via :func:`_mask_ip`
#     Every ``json.dumps({..., "ip": ...})`` call in ``app.py`` passes
#     ``client_ip`` through :func:`_mask_ip` before it is serialised.
#     This is the PRIMARY control: the raw IP never enters the log string.
#
#   Layer 2 — defence-in-depth via :class:`_RedactingFilter`
#     Attached to the root logging handler.  Applies :data:`_REDACT_PATTERNS`
#     to the fully formatted message BEFORE it is emitted.  Catches:
#       • HF token strings leaked via exception messages from
#         ``huggingface_hub`` (e.g. ``snapshot_download`` auth failures).
#       • IPv4 addresses emitted by third-party library loggers (httpx,
#         uvicorn) that bypass the call-site masking.
#       • Any future code that forgets to call :func:`_mask_ip` first.
#
# IPv6 is handled exclusively at Layer 1 (:func:`_mask_ip`).  A generic
# IPv6 regex in Layer 2 has unacceptable false-positive rates (e.g. it
# would match ``12:34:56:78`` in log timestamps or MAC addresses).
# ─────────────────────────────────────────────────────────────────────────────


def _mask_ip(ip: str) -> str:
    """Mask a client IP address for privacy-safe log output.

    Preserves enough network context for rate-limit and abuse analysis while
    zeroing the host portion that identifies the individual user.

    * **IPv4** — zero the last octet, retaining the /24 subnet.
      ``"192.168.1.100"`` → ``"192.168.1.0"``
    * **IPv6** — zero the interface identifier (last 64 bits), retaining
      the /64 prefix.  ``"2001:db8:85a3::8a2e:370:7334"`` → ``"2001:db8:85a3::"``
    * **IPv6 scope suffix** (e.g. ``"fe80::1%eth0"``) — stripped before
      parsing (Python's :mod:`ipaddress` does not accept scope identifiers).
    * **Non-IP strings** — returned as ``"<ip-redacted>"``.
    * **Sentinel** ``"unknown"`` — returned unchanged (already non-identifying).

    Parameters
    ----------
    ip : str
        Client IP string extracted from the HTTP request headers.
        May be ``"unknown"`` when the proxy header is absent.

    Returns
    -------
    str
        Masked IP suitable for structured log output.  This function is
        intentionally never-raise — any :exc:`ValueError` from
        :mod:`ipaddress` is caught and replaced by the safe fallback.

    Notes
    -----
    **Security note** — This is the canonical privacy gate for all IP values
    written to log records in ``app.py``.  Every ``json.dumps({..., "ip": …})``
    call must pass ``client_ip`` through :func:`_mask_ip` before serialising.
    Callers must **not** write raw ``client_ip`` values to any log record.

    **Developer note** — Uses :mod:`ipaddress` from the Python standard
    library; no third-party dependencies are introduced.

    Examples
    --------
    >>> _mask_ip("192.168.1.100")
    '192.168.1.0'
    >>> _mask_ip("10.0.0.255")
    '10.0.0.0'
    >>> _mask_ip("2001:db8:85a3::8a2e:370:7334")
    '2001:db8:85a3::'
    >>> _mask_ip("fe80::1%eth0")
    'fe80::'
    >>> _mask_ip("unknown")
    'unknown'
    >>> _mask_ip("not-an-ip")
    '<ip-redacted>'
    """
    if ip in ("unknown", ""):
        return ip
    try:
        # Strip IPv6 zone/scope identifier (e.g. "%eth0") — ipaddress rejects it.
        clean: str = ip.split("%", 1)[0].strip()
        addr = ipaddress.ip_address(clean)
        if isinstance(addr, ipaddress.IPv4Address):
            # Retain /24 (first three octets); zero the host octet.
            return str(ipaddress.ip_network(f"{addr}/24", strict=False).network_address)
        # IPv6: retain /64 prefix; zero the 64-bit interface identifier.
        return str(ipaddress.ip_network(f"{addr}/64", strict=False).network_address)
    except ValueError:
        return "<ip-redacted>"


#: Ordered list of ``(compiled_pattern, replacement)`` tuples applied by
#: :class:`_RedactingFilter` to every log record before emission.
#:
#: **Pattern order matters** — patterns are applied left-to-right; more
#: specific patterns must precede catch-all patterns.  There is no overlap
#: between the current patterns, but this convention must be maintained when
#: extending this list.
#:
#: IPv6 addresses are intentionally **absent** — they are handled at the
#: call-site by :func:`_mask_ip` (Layer 1).  A generic IPv6 regex in a
#: global filter produces too many false positives (hex timestamps, MAC
#: addresses, Docker overlay IDs) to be safe in a production log stream.
_REDACT_PATTERNS: list[tuple[re.Pattern[str], str]] = [
    # HuggingFace API tokens — ``hf_`` prefix followed by ≥ 4 alphanumeric
    # characters.  Classic tokens are ~34 chars; fine-grained tokens are ≥ 52.
    # The {4,} lower bound avoids matching ``hf_`` in legitimate identifiers
    # (e.g. Python identifiers that start with ``hf_``) while still catching
    # any partial token fragment that huggingface_hub may embed in an error
    # message.
    (re.compile(r"\bhf_[a-zA-Z0-9]{4,}\b"), "<token-redacted>"),
    # IPv4 addresses — strict dotted-decimal notation with per-octet range
    # validation (0-255).  Word boundaries prevent partial matches inside
    # longer numeric strings.  This pattern catches IPv4 strings emitted by
    # third-party loggers (httpx, uvicorn) that bypass :func:`_mask_ip`.
    (
        re.compile(
            r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}"
            r"(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b"
        ),
        "<ipv4-redacted>",
    ),
]


class _RedactingFilter(logging.Filter):
    """Scrub sensitive values from log records before emission.

    Applies the regex patterns in :data:`_REDACT_PATTERNS` to the fully
    formatted log message, replacing HuggingFace API tokens and raw IPv4
    addresses with opaque placeholders.

    This class is the **defence-in-depth layer** (Layer 2).  The primary
    control is :func:`_mask_ip` at each call site (Layer 1).  The filter
    catches values that slip through Layer 1 — most importantly, HF token
    strings embedded in exception messages from ``huggingface_hub``.

    Parameters
    ----------
    name : str, optional
        Filter name forwarded to :class:`logging.Filter`.  Default ``""``.

    Notes
    -----
    **Security note** — This filter materialises the fully formatted message
    via :meth:`logging.LogRecord.getMessage`, applies every pattern in
    :data:`_REDACT_PATTERNS`, then replaces :attr:`~logging.LogRecord.msg`
    with the scrubbed result and clears :attr:`~logging.LogRecord.args`.
    Clearing ``args`` prevents downstream handlers from re-applying ``%``
    formatting to a string that no longer contains positional placeholders.

    **Developer note** — Attach to the root handler immediately after
    construction so **every** handler in the process benefits::

        handler = logging.StreamHandler()
        handler.addFilter(_RedactingFilter())
        logging.root.handlers = [handler]

    To extend the redaction vocabulary, append a ``(pattern, replacement)``
    tuple to :data:`_REDACT_PATTERNS`.

    Examples
    --------
    >>> import logging
    >>> f = _RedactingFilter()
    >>> rec = logging.makeLogRecord(
    ...     {"msg": "token=hf_abc1234defg5678 ip=10.0.1.99", "args": ()}
    ... )
    >>> f.filter(rec)
    True
    >>> rec.msg
    'token=<token-redacted> ip=<ipv4-redacted>'
    """

    def filter(self, record: logging.LogRecord) -> bool:  # noqa: A003
        """Redact sensitive patterns from *record*'s formatted message.

        Parameters
        ----------
        record : logging.LogRecord
            Log record to inspect and mutate in-place.

        Returns
        -------
        bool
            Always ``True`` — this filter never suppresses records, only
            scrubs their message content.
        """
        # Materialise the full %-formatted string first, then scrub it.
        msg: str = sanitize_log_text(record.getMessage())
        # Write the scrubbed text back and clear args so that any subsequent
        # call to getMessage() returns the already-scrubbed string without
        # attempting to re-apply % formatting.
        record.msg = msg
        record.args = ()
        return True


def _classify_token_type(
    token: str,
    declared_type: str | None = None,
) -> HFTokenType:
    """
    Classify a HuggingFace token by its declared type or format heuristics.

    Token type classification is used at startup by :func:`_validate_token_config`
    to enforce the principle of least privilege before any requests arrive.

    Parameters
    ----------
    token : str
        The HuggingFace API token string.
    declared_type : str or None, optional
        Explicitly declared type from an environment variable
        (``HF_TOKEN_TYPE`` or ``HF_WRITE_TOKEN_TYPE``).
        Accepted values: ``"fine-grained"``, ``"read"``, ``"write"``
        (and minor formatting variants: ``"finegrained"``,
        ``"fine_grained"``).  When provided and recognized, it takes
        precedence over all heuristics.

    Returns
    -------
    HFTokenType
        One of ``"fine-grained"``, ``"read"``, ``"write"``, or ``"unknown"``.

    Notes
    -----
    **Security note** — Token type cannot be verified without an authenticated
    call to the HF API (``GET https://huggingface.co/api/whoami-v2``).  This
    function applies lightweight format heuristics only.  For production
    deployments, always declare the type explicitly via ``HF_TOKEN_TYPE`` /
    ``HF_WRITE_TOKEN_TYPE`` so :func:`_validate_token_config` can enforce
    least-privilege at startup without any network calls.

    **Developer note** — As of 2025, classic HF tokens are approximately 34
    characters total (``hf_`` prefix + 30 alphanumeric chars).  Fine-grained
    tokens are substantially longer (≥ 52 characters total as of the HF 2025
    token format).  This length heuristic is imprecise and subject to silent
    change by HF; explicit declaration via env vars is always preferred.

    Examples
    --------
    Explicit declaration takes precedence over heuristics:

    >>> _classify_token_type("hf_" + "a" * 30, declared_type="read")
    'read'
    >>> _classify_token_type("hf_" + "a" * 30, declared_type="write")
    'write'

    Heuristic: token ≥ 52 chars → fine-grained:

    >>> _classify_token_type("hf_" + "a" * 50)
    'fine-grained'

    Short classic token without declaration → unknown:

    >>> _classify_token_type("hf_" + "a" * 28)
    'unknown'

    Empty or malformed token → unknown:

    >>> _classify_token_type("")
    'unknown'
    """
    # Normalise accepted declared-type values (tolerate minor formatting variants).
    _declared_map: dict[str, HFTokenType] = {
        "fine-grained": "fine-grained",
        "finegrained": "fine-grained",
        "fine_grained": "fine-grained",
        "read": "read",
        "write": "write",
    }
    if declared_type:
        normalised = _declared_map.get(declared_type.lower().strip())
        if normalised is not None:
            return normalised

    # Validate basic token format — all HF tokens start with "hf_".
    if not token or not token.startswith("hf_") or len(token) < 10:  # noqa: PLR2004
        return "unknown"

    # Heuristic: fine-grained tokens are substantially longer than classic tokens.
    # Classic tokens: ~34 chars total.  Fine-grained tokens: ≥ 52 chars (HF 2025).
    # Best-effort only; explicit declaration via env vars is always preferred.
    if len(token) >= 52:  # noqa: PLR2004
        return "fine-grained"

    # Cannot distinguish classic read vs write by token string alone.
    return "unknown"


def _token_suitable_for_inference(token_type: str) -> bool:
    """
    Return ``True`` when *token_type* is appropriate for HF Inference API calls.

    This predicate guards inference paths (Path 2 private Space access and
    Path 3 HF Serverless API).  Returning ``False`` for a classic write token
    does not block the token at runtime — it causes :func:`_validate_token_config`
    to emit a startup ``WARNING`` so the operator knows they are running with
    more permission than necessary.

    Parameters
    ----------
    token_type : str
        One of the ``HF_TOKEN_TYPE_*`` constants or a free-form string parsed
        from an environment variable.

    Returns
    -------
    bool
        ``True`` for ``"fine-grained"``, ``"read"``, and ``"unknown"``.
        ``False`` for ``"write"`` (classic write token — over-privileged).

    Notes
    -----
    The recommended configuration is a fine-grained token scoped exclusively
    to ``Make calls to the serverless Inference API``, or a classic read
    token.  Classic write tokens carry unnecessary repo-write permission
    and violate the principle of least privilege.

    Examples
    --------
    >>> _token_suitable_for_inference("read")
    True
    >>> _token_suitable_for_inference("fine-grained")
    True
    >>> _token_suitable_for_inference("write")
    False
    >>> _token_suitable_for_inference("unknown")
    True
    """
    return token_type in HF_INFERENCE_TOKEN_TYPES


def _token_suitable_for_writes(token_type: str) -> bool:
    """
    Return ``True`` when *token_type* can authorize HuggingFace write operations.

    This predicate guards the ``/v1/contribute`` endpoint.  Returning ``False``
    for a classic read or unknown token causes :func:`_validate_token_config`
    to emit a startup ``ERROR`` string because the token WILL fail at
    ``HfApi.create_commit`` time (HTTP 403 / 401 from HF).

    Parameters
    ----------
    token_type : str
        One of the ``HF_TOKEN_TYPE_*`` constants or a free-form string parsed
        from an environment variable.

    Returns
    -------
    bool
        ``True`` for ``"fine-grained"`` and ``"write"``.
        ``False`` for ``"read"`` and ``"unknown"``.

    Notes
    -----
    Fine-grained tokens can write **only if** write permission was granted to
    the target repo at token-creation time.  A fine-grained token created
    with only inference-API scope will also fail on write operations, but the
    proxy cannot verify fine-grained permissions without an authenticated API
    call.  Fine-grained tokens are therefore accepted here and any permission
    failures surface at operation time with a clear HTTP 503 error.

    Examples
    --------
    >>> _token_suitable_for_writes("write")
    True
    >>> _token_suitable_for_writes("fine-grained")
    True
    >>> _token_suitable_for_writes("read")
    False
    >>> _token_suitable_for_writes("unknown")
    False
    """
    return token_type in HF_WRITE_TOKEN_TYPES


def _validate_token_config(
    hf_token: str,
    hf_write_token: str,
    training_dataset_repo: str = "",
    *,
    hf_token_type: str = HF_TOKEN_TYPE_UNKNOWN,
    hf_write_token_type: str = HF_TOKEN_TYPE_UNKNOWN,
) -> list[str]:
    """
    Validate token types and return actionable warning / error strings.

    Enforces the principle of least privilege and detects token-type
    misconfigurations that would cause silent failures at request time.
    Returns a list of strings rather than raising exceptions so the proxy
    can start in degraded mode and surface issues through structured logs.

    Call this at startup **after** :func:`_validate_env` so routing is
    confirmed viable before type checks are run.

    Parameters
    ----------
    hf_token : str
        HuggingFace token used for inference (``HF_TOKEN`` env var).
    hf_write_token : str
        HuggingFace token used for dataset persistence. New deployments pass the
        effective ``HF_DATASET_TOKEN``; legacy callers may still pass
        ``HF_WRITE_TOKEN``. Pass empty string when not configured.
    training_dataset_repo : str, optional
        HuggingFace Dataset repo ID (``TRAINING_DATASET_REPO`` env var).
        Pass empty string when ``/v1/contribute`` is not enabled.
    hf_token_type : str, optional
        Classified type for *hf_token* (from :func:`_classify_token_type`).
        Defaults to ``"unknown"``.
    hf_write_token_type : str, optional
        Classified type for *hf_write_token*.  Defaults to ``"unknown"``.

    Returns
    -------
    list[str]
        Zero or more diagnostic strings.  Each message is prefixed with
        ``"WARNING:"`` or ``"ERROR:"`` so callers can log at the correct
        level.  An empty list means the configuration passes all checks.

    Notes
    -----
    **Security note** — ``"write"`` token used for inference is a WARNING
    (not an error) because it functions correctly at runtime.  The warning
    exists to prompt the operator to apply least-privilege.

    **Security note** — ``"read"`` token used for writes is a hard ERROR:
    the token WILL fail on every ``HfApi.create_commit`` call.  The proxy
    can still start (useful for operators who only need inference), but
    ``/v1/contribute`` will be permanently non-functional until the token is
    replaced.

    Examples
    --------
    Clean configuration — no messages:

    >>> _validate_token_config("hf_readtok", "", hf_token_type="read")
    []

    Write token for inference (overprivileged) → WARNING:

    >>> msgs = _validate_token_config("hf_writetok", "", hf_token_type="write")
    >>> any("WARNING" in m for m in msgs)
    True

    Read token for writes → ERROR:

    >>> msgs = _validate_token_config(
    ...     "hf_tok",
    ...     "hf_readtok",
    ...     training_dataset_repo="org/dataset",
    ...     hf_write_token_type="read",
    ... )
    >>> any("ERROR" in m for m in msgs)
    True
    """
    messages: list[str] = []

    # ── Inference token (HF_TOKEN) type check ────────────────────────────────
    if hf_token and not _token_suitable_for_inference(hf_token_type):
        messages.append(
            f"WARNING: HF_TOKEN type is {hf_token_type!r} (classic write token). "
            "Write tokens carry unnecessary repo-push permission and violate the "
            "principle of least privilege for inference. "
            "Replace HF_TOKEN with: (a) a fine-grained token scoped to "
            "'Make calls to the serverless Inference API' only, or "
            "(b) a classic read token. "
            "See HF Settings → Tokens → New token → Fine-grained. "
            "Set HF_TOKEN_TYPE=read or HF_TOKEN_TYPE=fine-grained after replacing."
        )

    # ── Dataset-persistence token type check ─────────────────────────────────
    if hf_write_token and not _token_suitable_for_writes(hf_write_token_type):
        messages.append(
            f"ERROR: dataset persistence token type is {hf_write_token_type!r}. "
            "Read tokens cannot push commits to Hugging Face repositories. "
            "Use HF_DATASET_TOKEN with a fine-grained token scoped to write the "
            "target dataset repo (preferred), or a classic Write token. "
            "Legacy HF_WRITE_TOKEN remains supported as an alias."
        )

    # ── Training repo + effective write token consistency ────────────────────
    if training_dataset_repo:
        # Effective write token is HF_WRITE_TOKEN when set; else falls back to
        # HF_TOKEN.  Check that the effective token type can authorize writes.
        effective_token = hf_write_token or hf_token
        effective_type = hf_write_token_type if hf_write_token else hf_token_type
        if effective_token and not _token_suitable_for_writes(effective_type):
            messages.append(
                "ERROR: TRAINING_DATASET_REPO is configured but the effective "
                "write token type "
                f"({effective_type!r}) cannot push to HuggingFace repositories. "
                "POST /v1/contribute will always fail with HTTP 503. "
                "Set HF_DATASET_TOKEN to a write-capable token (fine-grained with "
                "write access to the dataset repo, or a classic Write token). "
                f"Set HF_DATASET_TOKEN_TYPE accordingly."
            )

    return messages


def _resolve_upstream_url(
    body: bytes,
    *,
    backend_url: str,
    hf_token: str,
    backend_auth_token: str = "",
    hf_spaces_auth_token: str = "",
    hf_base: str = DEFAULT_HF_BASE,
    default_model: str = DEFAULT_MODEL,
    hf_spaces_model_url: str = DEFAULT_HF_SPACES_MODEL_URL,
    hf_spaces_model_namespaces: (
        tuple[str, ...] | list[str]
    ) = DEFAULT_HF_SPACES_MODEL_NAMESPACES,
    proxy_timeout: float = float(DEFAULT_PROXY_TIMEOUT),
    path2_read_timeout: float = DEFAULT_PATH2_READ_TIMEOUT,
    path3_read_timeout: float = DEFAULT_PATH3_READ_TIMEOUT,
) -> tuple[str, dict[str, str], float]:
    """
    Centralised three-path routing — choose upstream endpoint, auth headers,
    and per-path read timeout.

    Priority
    --------
    1. *backend_url* is non-empty → **Path 1**: explicit custom backend.
       Forward to *backend_url* (Docker Model Runner, Ollama, any backend).
       *backend_auth_token* is injected only when explicitly configured.
       Read timeout: *proxy_timeout* (env ``PROXY_TIMEOUT``, default 600 s).

    2. Model namespace is in *hf_spaces_model_namespaces* → **Path 2**: HF model Space.
       Forward to *hf_spaces_model_url* (the ``scikit-plots/ai-model`` Space).
       CPU inference on a 7B model takes 4-5 minutes; *path2_read_timeout*
       (env ``PATH2_TIMEOUT``, default 600 s) prevents premature timeout.
       *hf_spaces_auth_token* is injected only when explicitly configured.

    3. Otherwise → **Path 3**: HF Serverless Inference API (default).
       Build ``{hf_base}/{model}/v1/chat/completions`` and inject *hf_token*
       (always required for the HF API).
       *path3_read_timeout* (env ``PATH3_TIMEOUT``, default 120 s) is
       appropriate for GPU-backed HF API inference.

    Parameters
    ----------
    body : bytes
        Raw JSON request body.  Used to extract the ``model`` field for
        Paths 2 and 3.
    backend_url : str
        Value of the ``BACKEND_URL`` environment variable.  Non-empty string
        triggers Path 1; empty string means "proceed to Path 2 / 3".
    hf_token : str
        HuggingFace inference token.  Used only for Path 3.
    backend_auth_token : str, optional
        Dedicated bearer capability bound to Path 1 ``backend_url``.
    hf_spaces_auth_token : str, optional
        Dedicated bearer capability bound to Path 2 ``hf_spaces_model_url``.
    hf_base : str, optional
        HF Serverless Inference API base URL (no trailing slash).
    default_model : str, optional
        Fallback model ID when the body omits the ``model`` field.
    hf_spaces_model_url : str, optional
        URL of the custom ai-model HF Space (Path 2 target).
    hf_spaces_model_namespaces : tuple[str, ...] or list[str], optional
        Model owner namespaces routed to *hf_spaces_model_url*.
    proxy_timeout : float, optional
        Read timeout (seconds) for Path 1.  Default: 600 s.
    path2_read_timeout : float, optional
        Read timeout (seconds) for Path 2 (ai-model Space).  Default: 600 s.
    path3_read_timeout : float, optional
        Read timeout (seconds) for Path 3 (HF Serverless API).  Default: 120 s.

    Returns
    -------
    url : str
        Fully-qualified upstream endpoint URL.
    headers : dict[str, str]
        HTTP headers for the upstream POST request.
    read_timeout_s : float
        Per-path read timeout in seconds.  Pass to ``httpx.Timeout(read=...)``.

    Notes
    -----
    **Breaking change v5.0.0** — Return type changed from
    ``tuple[str, dict]`` to ``tuple[str, dict, float]``.  All callers must
    unpack the third element.

    **Breaking change v6.0.0** — :data:`DEFAULT_HF_BASE` changed from
    ``https://api-inference.huggingface.co/models`` to
    ``https://router.huggingface.co``.  The old hostname was DNS-unresolvable
    from HF Docker Spaces ([Errno -5] EAI_NONAME).

    **Developer note** — All routing logic lives here.  To add a new backend
    type, add a new branch in this function.  Callers (``app.py``,
    ``dev_proxy.py``) remain unchanged when they already unpack 3 values.

    Examples
    --------
    Path 2 — scikit-plots namespace → ai-model Space:

    >>> url, hdrs, t = _resolve_upstream_url(
    ...     b'{"model":"scikit-plots/Qwen2.5-Coder-7B-Instruct","messages":[]}',
    ...     backend_url="",
    ...     hf_token="",
    ... )
    >>> "scikit-plots-ai-model.hf.space" in url
    True
    >>> t
    600.0

    Path 3 — standard HF Inference API:

    >>> url, hdrs, t = _resolve_upstream_url(
    ...     b'{"model":"openai/gpt-oss-20b","messages":[]}',
    ...     backend_url="",
    ...     hf_token="hf_test_token_abc123",
    ... )
    >>> "router.huggingface.co" in url
    True
    >>> t
    120.0

    Path 1 — explicit BACKEND_URL:

    >>> url, hdrs, t = _resolve_upstream_url(
    ...     b"{}",
    ...     backend_url="https://my-model.hf.space/v1/chat/completions",
    ...     hf_token="",
    ... )
    >>> url
    'https://my-model.hf.space/v1/chat/completions'
    >>> t
    600.0
    """  # noqa: D205
    headers: dict[str, str] = {"Content-Type": "application/json"}

    # ── Path 1: explicit custom backend override ──────────────────────────────
    if backend_url:
        if backend_auth_token:
            headers["Authorization"] = f"Bearer {backend_auth_token}"
        return backend_url, headers, proxy_timeout

    # Extract model ID from request body (needed for Paths 2 and 3).
    model: str = _parse_model(body, default=default_model)

    # ── Path 2: custom model namespace → HF Spaces model backend ─────────────
    if hf_spaces_model_url and _is_custom_model_namespace(
        model, hf_spaces_model_namespaces
    ):
        if hf_spaces_auth_token:
            headers["Authorization"] = f"Bearer {hf_spaces_auth_token}"
        return hf_spaces_model_url, headers, path2_read_timeout

    # ── Path 3: HF Serverless Inference API (provider models) ─────────────────
    # router.huggingface.co is a flat OpenAI-compatible endpoint.
    # The model is supplied in the request body (already present in `body`),
    # NOT embedded in the URL path.  The old api-inference.huggingface.co/models
    # API DID embed the model in the path as /{model}/v1/chat/completions, but
    # router.huggingface.co uses a single endpoint for all models:
    #   POST https://router.huggingface.co/v1/chat/completions
    #   body: {"model": "Qwen/Qwen2.5-Coder-7B-Instruct:nscale", ...}
    # Embedding the model ID in the path produces a 404/422 with no log entry
    # because _forward passes non-2xx upstream responses through transparently.
    url = f"{hf_base.rstrip('/')}/v1/chat/completions"
    # Do not manufacture an empty ``Authorization: Bearer `` header.  Besides
    # being useless, malformed/whitespace-only auth values may be rejected at
    # the local HTTP protocol layer before a request ever reaches Hugging Face.
    # When the token is absent, send no Authorization header and let the caller
    # or upstream return a normal authentication/configuration error.
    if hf_token:
        headers["Authorization"] = f"Bearer {hf_token}"
    return url, headers, path3_read_timeout


def _validate_credential_destination(
    url: str,
    *,
    credential_kind: str,
    allow_local_http: bool = False,
) -> None:
    """Fail closed when a server credential could be sent to an unsafe URL.

    ``credential_kind`` is descriptive and never contains the credential.  HF
    inference tokens are bound to official Hugging Face HTTPS origins; custom
    backend/Space tokens are separately configured and therefore bind to the
    exact operator-selected destination rather than reusing ``HF_TOKEN``.
    """
    if not url:
        raise RuntimeError(f"{credential_kind} is configured without a destination URL")
    try:
        parts = urlsplit(url)
        host = (parts.hostname or "").lower().rstrip(".")
        port = parts.port
    except (TypeError, ValueError) as exc:
        raise RuntimeError(
            f"unsafe destination for {credential_kind}: malformed URL"
        ) from exc
    if parts.username or parts.password or parts.query or parts.fragment:
        raise RuntimeError(
            f"unsafe destination for {credential_kind}: userinfo/query/fragment is not allowed"
        )
    is_local = host in {"localhost", "127.0.0.1", "::1"}
    if parts.scheme != "https" and not (
        allow_local_http and parts.scheme == "http" and is_local
    ):
        raise RuntimeError(
            f"unsafe destination for {credential_kind}: HTTPS is required"
        )
    if credential_kind == "HF_TOKEN":
        if host != "router.huggingface.co" and not host.endswith(".huggingface.co"):
            raise RuntimeError(
                "unsafe destination for HF_TOKEN: token is bound to official Hugging Face origins"
            )
        if port not in (None, 443):
            raise RuntimeError("unsafe destination for HF_TOKEN: non-standard port")


def _validate_env(
    backend_url: str,
    hf_token: str,
    hf_spaces_model_url: str = DEFAULT_HF_SPACES_MODEL_URL,
) -> None:
    """
    Validate the minimum required environment at proxy startup.

    At least one of the three routing paths must be viable:

    * **Path 1** — *backend_url* is non-empty.
    * **Path 2** — *hf_spaces_model_url* is non-empty (serves custom namespace models).
    * **Path 3** — *hf_token* is non-empty (HF Inference API for provider models).

    Parameters
    ----------
    backend_url : str
        Value of the ``BACKEND_URL`` environment variable (may be empty).
    hf_token : str
        Value of the ``HF_TOKEN`` environment variable (may be empty).
    hf_spaces_model_url : str, optional
        Value of the ``HF_SPACES_MODEL_URL`` environment variable.

    Raises
    ------
    RuntimeError
        When all three routing paths are disabled (all parameters are empty).

    Examples
    --------
    >>> _validate_env("https://my-model.hf.space/v1/chat/completions", "", "")
    >>> _validate_env("", "hf_mytoken", "")
    >>> _validate_env(
    ...     "", "", "https://scikit-plots-ai-model.hf.space/v1/chat/completions"
    ... )
    >>> import pytest
    >>> with pytest.raises(RuntimeError, match="no viable routing path"):
    ...     _validate_env("", "", "")
    """
    if not backend_url and not hf_token and not hf_spaces_model_url:
        raise RuntimeError(
            "Proxy configuration error: no viable routing path configured.\n\n"
            "Set at least ONE of the following in Space → Settings → Repository secrets:\n\n"
            "  Option 1 — HF Inference API (standard provider models):\n"
            "    HF_TOKEN      = hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxx\n"
            "    DEFAULT_MODEL = openai/gpt-oss-20b\n\n"
            "  Option 2 — Custom ai-model Space (scikit-plots/* models):\n"
            "    HF_SPACES_MODEL_URL = "
            "https://scikit-plots-ai-model.hf.space/v1/chat/completions\n\n"
            "  Option 3 — Explicit custom backend (DMR, Ollama, or any backend):\n"
            "    BACKEND_URL = http://localhost:12434/engines/llama.cpp/v1/chat/completions\n\n"
            "See FREE_PROXY_SOLUTIONS.md for the full path decision tree."
        )


def load_proxy_env() -> dict[str, Any]:
    """
    Read all proxy-relevant environment variables and return a typed dict.

    Returns
    -------
    dict[str, Any]
        Keys and types:

        ``backend_url`` : str
        ``hf_token`` : str
        ``hf_base`` : str
        ``default_model`` : str
        ``hf_spaces_model_url`` : str
        ``hf_spaces_model_namespaces`` : tuple[str, ...]
        ``proxy_timeout`` : int
            Global / Path 1 read timeout (env ``PROXY_TIMEOUT``).
        ``path2_read_timeout`` : float
            Path 2 read timeout (env ``PATH2_TIMEOUT``).
        ``path3_read_timeout`` : float
            Path 3 read timeout (env ``PATH3_TIMEOUT``).
        ``max_body_bytes`` : int
        ``allowed_origins`` : str
        ``allowed_origins_mode`` : str
            Raw deployment composition mode (``additive`` or ``replace``).
        ``hf_token_type`` : str
            Classified token type for *hf_token* (env ``HF_TOKEN_TYPE``).
            One of ``"fine-grained"``, ``"read"``, ``"write"``, ``"unknown"``.
        ``hf_write_token_type`` : str
            Classified type for the legacy ``HF_WRITE_TOKEN`` alias.
        ``hf_dataset_token_type`` : str
            Classified type for the effective dataset-persistence token. One of
            ``"fine-grained"``, ``"read"``, ``"write"``, ``"unknown"``.

    Examples
    --------
    >>> import os
    >>> os.environ["PROXY_TIMEOUT"] = "600"
    >>> cfg = load_proxy_env()
    >>> cfg["proxy_timeout"]
    600
    >>> os.environ["PATH2_TIMEOUT"] = "900"
    >>> cfg = load_proxy_env()
    >>> cfg["path2_read_timeout"]
    900.0
    """
    _raw_namespaces: str = os.environ.get(
        "HF_SPACES_MODEL_NAMESPACES",
        ",".join(DEFAULT_HF_SPACES_MODEL_NAMESPACES),
    )
    _parsed_namespaces: tuple[str, ...] = (
        tuple(ns.strip() for ns in _raw_namespaces.split(",") if ns.strip())
        or DEFAULT_HF_SPACES_MODEL_NAMESPACES
    )

    _hf_token: str = os.environ.get("HF_TOKEN", "").strip()
    _hf_dataset_token_explicit: str = os.environ.get("HF_DATASET_TOKEN", "").strip()
    _hf_write_token: str = os.environ.get("HF_WRITE_TOKEN", "").strip()

    # Classify token types from explicit declarations (preferred) or heuristics.
    # Explicit: set HF_TOKEN_TYPE=read|write|fine-grained in Space secrets.
    # Heuristic: length-based guess (fine-grained tokens are ≥ 52 chars).
    _hf_token_type: str = _classify_token_type(
        _hf_token,
        declared_type=os.environ.get("HF_TOKEN_TYPE"),
    )
    _hf_write_token_type: str = _classify_token_type(
        _hf_write_token,
        declared_type=os.environ.get("HF_WRITE_TOKEN_TYPE"),
    )
    _hf_dataset_token: str = _hf_dataset_token_explicit or _hf_write_token or _hf_token
    _hf_dataset_token_type: str = (
        _classify_token_type(
            _hf_dataset_token_explicit,
            declared_type=os.environ.get("HF_DATASET_TOKEN_TYPE"),
        )
        if _hf_dataset_token_explicit
        else (_hf_write_token_type if _hf_write_token else _hf_token_type)
    )

    return {
        "backend_url": os.environ.get("BACKEND_URL", "").strip(),
        "hf_token": _hf_token,
        # Preferred dataset token + legacy alias. Never forward the effective
        # dataset token to model backends.
        "hf_write_token": _hf_write_token,
        "hf_dataset_token": _hf_dataset_token,
        # Token type metadata — used by startup validation and discovery.
        "hf_token_type": _hf_token_type,
        "hf_write_token_type": _hf_write_token_type,
        "hf_dataset_token_type": _hf_dataset_token_type,
        "hf_base": os.environ.get("HF_BASE", DEFAULT_HF_BASE).rstrip("/"),
        "default_model": (
            os.environ.get("DEFAULT_MODEL", DEFAULT_MODEL).strip() or DEFAULT_MODEL
        ),
        "hf_spaces_model_url": (
            os.environ.get("HF_SPACES_MODEL_URL", DEFAULT_HF_SPACES_MODEL_URL).strip()
        ),
        "hf_spaces_model_namespaces": _parsed_namespaces,
        "proxy_timeout": _safe_int(
            os.environ.get("PROXY_TIMEOUT"),
            DEFAULT_PROXY_TIMEOUT,
        ),
        "path2_read_timeout": _safe_float(
            os.environ.get("PATH2_TIMEOUT"),
            DEFAULT_PATH2_READ_TIMEOUT,
        ),
        "path3_read_timeout": _safe_float(
            os.environ.get("PATH3_TIMEOUT"),
            DEFAULT_PATH3_READ_TIMEOUT,
        ),
        "max_body_bytes": _safe_int(
            os.environ.get("MAX_BODY_BYTES"),
            DEFAULT_MAX_BODY_BYTES,
        ),
        "allowed_origins": os.environ.get("ALLOWED_ORIGINS", "").strip(),
        "allowed_origins_mode": (
            os.environ.get("ALLOWED_ORIGINS_MODE", "additive").strip().lower()
            or "additive"
        ),
    }