"""Strict transport, normalization, and RFC 8785 primitives for pool-selection.v2."""

from __future__ import annotations

import hashlib
import json
import math
import re
import unicodedata
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation

SCHEMA = "claude-multiacc/pool-selection.v2"
# NOT bumped for the headroom band, and NOT bumped for the session gate either. This
# string is a BUILD FENCE, not a changelog: a queued task records the version that
# reserved its account, and the claim function (agent-sdk
# 0005_delivery_claim_functions.sql) only lets a runner advertising the same string
# claim it — while the runner's value is a hardcoded constant shipped in its bundle.
# With Macs routinely several builds behind, bumping this stops every lagging Mac from
# claiming any new work until it updates. Both knobs are additive and optional: the
# request only grows optional keys, the response only grows fields, and every existing
# consumer keeps reading exactly what it read before.
SELECTOR_VERSION = "2.0.1"
TIME_RE = re.compile(
    r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-5][0-9]"
    r"(\.[0-9]{1,6})?(Z|[+-][0-9]{2}:[0-9]{2})$")
DECIMAL_RE = re.compile(r"^-?(0|[1-9][0-9]*)(\.[0-9]+)?$")
SAFE_INTEGER = 9_007_199_254_740_991
MAX_PLAIN_DECIMAL_PLACES = 10_000
IDENTITY_KEYS = {"runner_id", "runner_generation", "engine", "account_id"}
CANDIDATE_KEYS = IDENTITY_KEYS | {
    "status", "weekly_pct", "session_pct", "resets_at", "limited_until", "seen_at",
    "provider_capable", "reservation_history"}
REQUEST_KEYS = {
    "schema", "database_now", "policy", "required_engine", "producer_identity",
    "excluded_identities", "candidates", "reservation_key"}
# Optional because a caller pinned to an older panel build still sends exactly
# REQUEST_KEYS; absent, the policy applies its own defaults. app-robot's panel sends
# headroom_band from its own setting and never sends session_gate, so the gate default
# below is what every panel launch gets (operator's 2026-09-03 decision: session
# headroom gates FIRST, then the weekly band picks among the survivors).
OPTIONAL_REQUEST_KEYS = {"headroom_band", "session_gate"}
DEFAULT_HEADROOM_BAND = "30"
DEFAULT_SESSION_GATE = "50"
POLICIES = {
    "default_claude", "default_codex", "default_both", "explicit", "producer_retry", "reviewer"}


def timestamp(value: object) -> tuple[datetime, str]:
    if not isinstance(value, str) or not TIME_RE.fullmatch(value):
        raise ValueError("invalid timestamp")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        utc = parsed.astimezone(timezone.utc).replace(tzinfo=None)
    except (ValueError, OverflowError) as error:
        raise ValueError("invalid timestamp") from error
    return parsed, utc.strftime("%Y-%m-%dT%H:%M:%S.%fZ")


def decimal_value(value: object) -> tuple[Decimal | None, str | None]:
    if isinstance(value, bool) or value is None:
        return None, None
    if isinstance(value, str) and not DECIMAL_RE.fullmatch(value):
        return None, None
    try:
        parsed = Decimal(str(value))
        if not parsed.is_finite() or (isinstance(value, float) and not math.isfinite(value)):
            return None, None
        if parsed <= 0:
            clamped = Decimal(0)
        elif parsed >= 100:
            clamped = Decimal(100)
        elif parsed.as_tuple().exponent < -MAX_PLAIN_DECIMAL_PLACES:
            return None, None
        else:
            clamped = parsed
        result = format(clamped, "f") if clamped else "0"
    except (InvalidOperation, ValueError, OverflowError):
        return None, None
    return clamped, result.rstrip("0").rstrip(".") if "." in result else result


def identity(raw: dict) -> tuple[int | None, int | None, str | None, str | None, bool]:
    runner = raw.get("runner_id")
    generation = raw.get("runner_generation")
    engine_value = raw.get("engine")
    account_value = raw.get("account_id")
    engine = engine_value.strip().lower() if isinstance(engine_value, str) else None
    account = unicodedata.normalize("NFC", account_value).strip() \
        if isinstance(account_value, str) else None
    runner = runner if isinstance(runner, int) and not isinstance(runner, bool) and runner > 0 else None
    generation = generation if isinstance(generation, int) \
        and not isinstance(generation, bool) and generation > 0 else None
    engine = engine if engine in {"claude", "codex"} else None
    account = account or None
    return runner, generation, engine, account, all((runner, generation, engine, account))


def collision_key(parts: tuple) -> tuple:
    runner, generation, engine, account = parts
    folded = unicodedata.normalize("NFC", account.casefold()) if account else account
    return runner, generation, engine, folded


def _encode_rfc8785(value: object) -> str:
    if value is None:
        return "null"
    if value is True:
        return "true"
    if value is False:
        return "false"
    if isinstance(value, int):
        if abs(value) > SAFE_INTEGER:
            raise ValueError("integer outside RFC 8785 interoperable range")
        return str(value)
    if isinstance(value, str):
        value.encode("utf-8")
        return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
    if isinstance(value, list):
        return "[" + ",".join(_encode_rfc8785(item) for item in value) + "]"
    if isinstance(value, dict):
        if any(not isinstance(key, str) for key in value):
            raise ValueError("RFC 8785 object keys must be strings")
        keys = sorted(value, key=lambda key: key.encode("utf-16-be", "surrogatepass"))
        pairs = (_encode_rfc8785(key) + ":" + _encode_rfc8785(value[key]) for key in keys)
        return "{" + ",".join(pairs) + "}"
    raise ValueError(f"value outside frozen RFC 8785 digest domain: {type(value).__name__}")


def canonical_sha256(value: object) -> str:
    return hashlib.sha256(_encode_rfc8785(value).encode("utf-8")).hexdigest()


def _identity_error(value: object, field: str) -> str | None:
    if not isinstance(value, dict) or set(value) != IDENTITY_KEYS:
        return field
    runner, generation, _, account, valid = identity(value)
    if not isinstance(value.get("engine"), str) or not valid:
        return field
    if abs(runner or 0) > SAFE_INTEGER or abs(generation or 0) > SAFE_INTEGER or account is None:
        return field
    return None


def _candidate_error(value: object, index: int) -> str | None:
    field = f"candidates[{index}]"
    if not isinstance(value, dict) or set(value) != CANDIDATE_KEYS:
        return field
    for name in ("runner_id", "runner_generation"):
        item = value[name]
        if not isinstance(item, int) or isinstance(item, bool) or not 0 < item <= SAFE_INTEGER:
            return f"{field}.{name}"
    if not all(isinstance(value[name], str) for name in ("engine", "account_id", "status", "seen_at")):
        return field
    if not isinstance(value["provider_capable"], bool):
        return f"{field}.provider_capable"
    scalars = (str, int, float, Decimal, bool, type(None))
    if any(not isinstance(value[name], scalars) for name in ("weekly_pct", "session_pct")):
        return field
    if any(isinstance(value[name], float) and not math.isfinite(value[name])
           for name in ("weekly_pct", "session_pct")):
        return field
    if any(item is not None and not isinstance(item, str)
           for item in (value["resets_at"], value["limited_until"])):
        return field
    history = value["reservation_history"]
    if not isinstance(history, dict) or set(history) != {"active_expires_at", "last_selected_at"}:
        return f"{field}.reservation_history"
    if any(item is not None and not isinstance(item, str) for item in history.values()):
        return f"{field}.reservation_history"
    if history["active_expires_at"] is not None:
        try:
            timestamp(history["active_expires_at"])
        except ValueError:
            return f"{field}.reservation_history.active_expires_at"
    return None


def request_error(request: object) -> tuple[str, dict] | None:
    if not isinstance(request, dict):
        return "invalid_request", {"field": "$"}
    if "schema" not in request:
        return "invalid_request", {"field": "schema"}
    if request.get("schema") != SCHEMA:
        return "unsupported_schema", {"field": "schema"}
    keys = set(request)
    if not REQUEST_KEYS <= keys or keys - REQUEST_KEYS - OPTIONAL_REQUEST_KEYS:
        return "invalid_request", {"field": "$"}
    for knob in ("headroom_band", "session_gate"):
        if knob not in request:
            continue
        value, _text = decimal_value(request[knob])
        # decimal_value clamps to [0,100] and rejects non-numeric shapes; a knob the
        # caller cannot express exactly must fail loudly rather than silently widen
        # selection to the whole pool (band) or wave every session-heavy account
        # through (gate). Both are validated identically so the panel can send either.
        if value is None or isinstance(request[knob], bool):
            return "invalid_request", {"field": knob}
    try:
        timestamp(request["database_now"])
    except ValueError:
        return "invalid_request", {"field": "database_now"}
    policy, required = request["policy"], request["required_engine"]
    if not isinstance(policy, str) or policy not in POLICIES:
        return "invalid_request", {"field": "policy"}
    if required is not None and (not isinstance(required, str) or required not in {"claude", "codex"}):
        return "invalid_request", {"field": "required_engine"}
    needs_engine = policy in {"explicit", "producer_retry", "reviewer"}
    if (needs_engine and required is None) or (not needs_engine and required is not None):
        return "invalid_request", {"field": "required_engine"}
    producer = request["producer_identity"]
    if (policy == "reviewer") != (producer is not None):
        return "invalid_request", {"field": "producer_identity"}
    if producer is not None and (error := _identity_error(producer, "producer_identity")):
        return "invalid_request", {"field": error}
    excluded = request["excluded_identities"]
    if not isinstance(excluded, list):
        return "invalid_request", {"field": "excluded_identities"}
    for index, value in enumerate(excluded):
        if error := _identity_error(value, f"excluded_identities[{index}]"):
            return "invalid_request", {"field": error}
    if not isinstance(request["reservation_key"], str) or not request["reservation_key"].strip():
        return "invalid_request", {"field": "reservation_key"}
    if not isinstance(request["candidates"], list):
        return "invalid_request", {"field": "candidates"}
    for index, value in enumerate(request["candidates"]):
        if error := _candidate_error(value, index):
            return "invalid_request", {"field": error}
    return None