"""Deterministic audit for a typed tool authority contract.

This module is an illustrative, local validation fixture.  It does not call a
tool, authenticate a principal, evaluate a production policy engine, or certify
that an action is safe.  A production runtime still has to bind authenticated
identity, current resource state, authorization, approval, idempotency storage,
execution, and audit evidence at the real side-effect boundary.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass
from hashlib import sha256
import json
from math import isfinite
import re
from typing import Literal


JsonType = Literal["string", "integer", "number", "boolean"]
Effect = Literal["read", "write"]
ApprovalMode = Literal["none", "per_call"]
IdempotencyMode = Literal["not_applicable", "required"]

NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
VERSION_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
SCOPE_PATTERN = re.compile(r"^[a-z][a-z0-9:_-]{1,95}$")
IDEMPOTENCY_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
MAX_SAFE_INTEGER = 2**53 - 1


def _exact_text(name: str, value: object, *, maximum: int = 512) -> str:
    if type(value) is not str or not value.strip():
        raise ValueError(f"{name} must be non-empty exact str")
    if value != value.strip():
        raise ValueError(f"{name} must not have surrounding whitespace")
    if len(value) > maximum or any(ord(character) < 32 for character in value):
        raise ValueError(f"{name} is outside the bounded text contract")
    return value


def _exact_bool(name: str, value: object) -> bool:
    if type(value) is not bool:
        raise ValueError(f"{name} must be exact bool")
    return value


def _exact_int(name: str, value: object, *, minimum: int, maximum: int) -> int:
    if type(value) is not int:
        raise ValueError(f"{name} must be exact int")
    if not minimum <= value <= maximum:
        raise ValueError(f"{name} must be between {minimum} and {maximum}")
    return value


def _exact_safe_integer(name: str, value: object) -> int:
    if type(value) is not int:
        raise ValueError(f"{name} must be exact int")
    if not -MAX_SAFE_INTEGER <= value <= MAX_SAFE_INTEGER:
        raise ValueError(f"{name} must be a JSON-safe integer")
    return value


def _optional_finite_number(name: str, value: object) -> int | float | None:
    if value is None:
        return None
    if type(value) is int:
        return _exact_safe_integer(name, value)
    if type(value) is not float:
        raise ValueError(f"{name} must be a non-boolean number or None")
    if not isfinite(value):
        raise ValueError(f"{name} must be finite")
    return value


def _stable_digest(payload: object) -> str:
    encoded = json.dumps(
        payload,
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
    return sha256(encoded).hexdigest()


@dataclass(frozen=True)
class ParameterSpec:
    name: str
    json_type: JsonType
    required: bool
    minimum: int | float | None = None
    maximum: int | float | None = None
    max_length: int | None = None

    def __post_init__(self) -> None:
        _exact_text("parameter.name", self.name, maximum=64)
        if not NAME_PATTERN.fullmatch(self.name):
            raise ValueError("parameter.name must be a canonical snake_case name")
        _exact_text("parameter.json_type", self.json_type, maximum=16)
        if self.json_type not in {"string", "integer", "number", "boolean"}:
            raise ValueError("parameter.json_type is unsupported")
        _exact_bool(f"{self.name}.required", self.required)
        minimum = _optional_finite_number(f"{self.name}.minimum", self.minimum)
        maximum = _optional_finite_number(f"{self.name}.maximum", self.maximum)
        object.__setattr__(self, "minimum", minimum)
        object.__setattr__(self, "maximum", maximum)
        if minimum is not None and maximum is not None and minimum > maximum:
            raise ValueError(f"{self.name} minimum exceeds maximum")
        if self.max_length is not None:
            _exact_int(f"{self.name}.max_length", self.max_length, minimum=1, maximum=4096)
        if self.json_type == "string":
            if minimum is not None or maximum is not None:
                raise ValueError("string parameters cannot declare numeric bounds")
            if self.max_length is None:
                raise ValueError("string parameters require max_length")
        elif self.max_length is not None:
            raise ValueError("only string parameters can declare max_length")


@dataclass(frozen=True)
class ToolContract:
    version: str
    name: str
    description: str
    effect: Effect
    parameters: tuple[ParameterSpec, ...]
    required_scopes: tuple[str, ...]
    approval_mode: ApprovalMode
    idempotency_mode: IdempotencyMode
    max_calls_per_run: int
    error_codes: tuple[str, ...]

    def __post_init__(self) -> None:
        _exact_text("version", self.version, maximum=64)
        if not VERSION_PATTERN.fullmatch(self.version):
            raise ValueError("version must be a canonical identifier")
        _exact_text("name", self.name, maximum=64)
        if not NAME_PATTERN.fullmatch(self.name):
            raise ValueError("name must be a canonical snake_case name")
        _exact_text("description", self.description, maximum=512)
        _exact_text("effect", self.effect, maximum=8)
        if self.effect not in {"read", "write"}:
            raise ValueError("effect must be read or write")

        try:
            parameters = tuple(self.parameters)
            scopes = tuple(self.required_scopes)
            error_codes = tuple(self.error_codes)
        except TypeError as error:
            raise ValueError("contract collections must be iterable") from error
        object.__setattr__(self, "parameters", parameters)
        object.__setattr__(self, "required_scopes", scopes)
        object.__setattr__(self, "error_codes", error_codes)
        if not parameters or any(type(item) is not ParameterSpec for item in parameters):
            raise ValueError("parameters must contain exact ParameterSpec records")
        names = [parameter.name for parameter in parameters]
        if len(names) != len(set(names)):
            raise ValueError("parameter names must be unique")
        if not scopes:
            raise ValueError("required_scopes must not be empty")
        for scope in scopes:
            _exact_text("required_scope", scope, maximum=96)
            if not SCOPE_PATTERN.fullmatch(scope):
                raise ValueError("required_scope must be canonical")
        if len(scopes) != len(set(scopes)):
            raise ValueError("required_scopes must be unique")
        if tuple(sorted(scopes)) != scopes:
            raise ValueError("required_scopes must be sorted")
        _exact_text("approval_mode", self.approval_mode, maximum=16)
        if self.approval_mode not in {"none", "per_call"}:
            raise ValueError("approval_mode is unsupported")
        _exact_text("idempotency_mode", self.idempotency_mode, maximum=16)
        if self.idempotency_mode not in {"not_applicable", "required"}:
            raise ValueError("idempotency_mode is unsupported")
        _exact_int("max_calls_per_run", self.max_calls_per_run, minimum=1, maximum=1000)
        if not error_codes:
            raise ValueError("error_codes must not be empty")
        for code in error_codes:
            _exact_text("error_code", code, maximum=64)
            if not re.fullmatch(r"[A-Z][A-Z0-9_]{2,63}", code):
                raise ValueError("error_code must be canonical upper snake case")
        if len(error_codes) != len(set(error_codes)):
            raise ValueError("error_codes must be unique")
        if tuple(sorted(error_codes)) != error_codes:
            raise ValueError("error_codes must be sorted")
        if self.effect == "write" and (
            self.approval_mode != "per_call" or self.idempotency_mode != "required"
        ):
            raise ValueError("write contracts require per-call approval and idempotency")
        if self.effect == "read" and self.idempotency_mode != "not_applicable":
            raise ValueError("read contracts use not_applicable idempotency in this fixture")

    @property
    def content_id(self) -> str:
        return "tool-contract@sha256:" + _stable_digest(asdict(self))


@dataclass(frozen=True)
class ToolInvocation:
    tool_name: str
    contract_content_id: str
    arguments: tuple[tuple[str, object], ...]
    granted_scopes: tuple[str, ...]
    approved: bool
    approval_binding: str | None
    idempotency_key: str | None
    call_index: int

    def __post_init__(self) -> None:
        _exact_text("tool_name", self.tool_name, maximum=64)
        _exact_text("contract_content_id", self.contract_content_id, maximum=96)
        if not re.fullmatch(r"tool-contract@sha256:[0-9a-f]{64}", self.contract_content_id):
            raise ValueError("contract_content_id must be a tool contract digest")
        try:
            raw_arguments = tuple(self.arguments)
            scopes = tuple(self.granted_scopes)
        except TypeError as error:
            raise ValueError("invocation collections must be iterable") from error

        arguments: list[tuple[str, object]] = []
        for index, pair in enumerate(raw_arguments):
            if type(pair) not in {tuple, list} or len(pair) != 2:
                raise ValueError(f"arguments[{index}] must be a name/value pair")
            name, value = pair
            _exact_text(f"arguments[{index}].name", name, maximum=64)
            arguments.append((name, value))
        object.__setattr__(self, "arguments", tuple(arguments))
        object.__setattr__(self, "granted_scopes", scopes)
        argument_names = [name for name, _ in arguments]
        if len(argument_names) != len(set(argument_names)):
            raise ValueError("argument names must be unique")
        for scope in scopes:
            _exact_text("granted_scope", scope, maximum=96)
            if not SCOPE_PATTERN.fullmatch(scope):
                raise ValueError("granted_scope must be canonical")
        if len(scopes) != len(set(scopes)):
            raise ValueError("granted_scopes must be unique")
        if tuple(sorted(scopes)) != scopes:
            raise ValueError("granted_scopes must be sorted")
        _exact_bool("approved", self.approved)
        if self.approval_binding is not None:
            _exact_text("approval_binding", self.approval_binding, maximum=64)
            if not re.fullmatch(r"[0-9a-f]{64}", self.approval_binding):
                raise ValueError("approval_binding must be lowercase SHA-256")
        if self.idempotency_key is not None:
            _exact_text("idempotency_key", self.idempotency_key, maximum=128)
            if not IDEMPOTENCY_PATTERN.fullmatch(self.idempotency_key):
                raise ValueError("idempotency_key must be a bounded opaque identifier")
        _exact_int("call_index", self.call_index, minimum=1, maximum=1_000_000)


@dataclass(frozen=True)
class AuditResult:
    contract: ToolContract
    invocation: ToolInvocation
    delegated_scopes: tuple[str, ...]
    decision: Literal[
        "ALLOW",
        "DENY_SCOPE",
        "DENY_APPROVAL",
        "DENY_IDEMPOTENCY",
        "DENY_CALL_BUDGET",
    ]


def approval_binding(
    *,
    contract_content_id: str,
    tool_name: str,
    arguments: tuple[tuple[str, object], ...],
    idempotency_key: str | None,
) -> str:
    """Bind an approval receipt to one exact logical tool proposal."""

    _exact_text("contract_content_id", contract_content_id, maximum=96)
    if not re.fullmatch(r"tool-contract@sha256:[0-9a-f]{64}", contract_content_id):
        raise ValueError("contract_content_id must be a tool contract digest")
    _exact_text("tool_name", tool_name, maximum=64)
    if not NAME_PATTERN.fullmatch(tool_name):
        raise ValueError("tool_name must be canonical")
    try:
        normalized_arguments = tuple(tuple(pair) for pair in arguments)
    except TypeError as error:
        raise ValueError("arguments must be name/value pairs") from error
    if idempotency_key is not None:
        _exact_text("idempotency_key", idempotency_key, maximum=128)
    return _stable_digest(
        {
            "arguments": normalized_arguments,
            "contract_content_id": contract_content_id,
            "idempotency_key": idempotency_key,
            "tool_name": tool_name,
        }
    )


def _rebuild_parameter(value: object) -> ParameterSpec:
    if type(value) is not ParameterSpec:
        raise ValueError("parameters must contain exact ParameterSpec records")
    return ParameterSpec(
        value.name,
        value.json_type,
        value.required,
        value.minimum,
        value.maximum,
        value.max_length,
    )


def _rebuild_contract(value: object) -> ToolContract:
    if type(value) is not ToolContract:
        raise ValueError("contract must be exact ToolContract")
    try:
        parameters = tuple(_rebuild_parameter(item) for item in value.parameters)
    except (AttributeError, TypeError) as error:
        raise ValueError("contract failed reconstruction") from error
    return ToolContract(
        value.version,
        value.name,
        value.description,
        value.effect,
        parameters,
        value.required_scopes,
        value.approval_mode,
        value.idempotency_mode,
        value.max_calls_per_run,
        value.error_codes,
    )


def _rebuild_invocation(value: object) -> ToolInvocation:
    if type(value) is not ToolInvocation:
        raise ValueError("invocation must be exact ToolInvocation")
    try:
        return ToolInvocation(
            value.tool_name,
            value.contract_content_id,
            value.arguments,
            value.granted_scopes,
            value.approved,
            value.approval_binding,
            value.idempotency_key,
            value.call_index,
        )
    except (AttributeError, TypeError) as error:
        raise ValueError("invocation failed reconstruction") from error


def _validate_argument(parameter: ParameterSpec, value: object) -> None:
    label = f"argument {parameter.name}"
    if parameter.json_type == "string":
        _exact_text(label, value, maximum=parameter.max_length or 1)
        return
    if parameter.json_type == "boolean":
        _exact_bool(label, value)
        return
    if parameter.json_type == "integer":
        numeric = _exact_safe_integer(label, value)
    else:
        if type(value) is int:
            numeric = _exact_safe_integer(label, value)
        elif type(value) is float:
            numeric = value
            if not isfinite(numeric):
                raise ValueError(f"{label} must be finite")
        else:
            raise ValueError(f"{label} must be a non-boolean number")
    if parameter.minimum is not None and numeric < parameter.minimum:
        raise ValueError(f"{label} is below minimum")
    if parameter.maximum is not None and numeric > parameter.maximum:
        raise ValueError(f"{label} exceeds maximum")


def audit(contract: object, invocation: object) -> AuditResult:
    """Reconstruct and audit one proposal without executing or mutating it."""

    checked_contract = _rebuild_contract(contract)
    checked_invocation = _rebuild_invocation(invocation)
    if checked_invocation.tool_name != checked_contract.name:
        raise ValueError("tool_name does not match contract")
    if checked_invocation.contract_content_id != checked_contract.content_id:
        raise ValueError("invocation is not bound to this contract content")

    supplied = dict(checked_invocation.arguments)
    expected = {parameter.name for parameter in checked_contract.parameters}
    required = {
        parameter.name for parameter in checked_contract.parameters if parameter.required
    }
    missing = required - set(supplied)
    unexpected = set(supplied) - expected
    if missing or unexpected:
        details: list[str] = []
        if missing:
            details.append("missing=" + ",".join(sorted(missing)))
        if unexpected:
            details.append("unexpected=" + ",".join(sorted(unexpected)))
        raise ValueError("argument schema mismatch: " + "; ".join(details))
    for parameter in checked_contract.parameters:
        if parameter.name in supplied:
            _validate_argument(parameter, supplied[parameter.name])

    granted = set(checked_invocation.granted_scopes)
    required_scopes = set(checked_contract.required_scopes)
    delegated_scopes = tuple(scope for scope in checked_contract.required_scopes if scope in granted)
    if not required_scopes.issubset(granted):
        decision = "DENY_SCOPE"
    elif (
        checked_contract.idempotency_mode == "required"
        and checked_invocation.idempotency_key is None
    ):
        decision = "DENY_IDEMPOTENCY"
    elif checked_contract.approval_mode == "per_call" and (
        not checked_invocation.approved
        or checked_invocation.approval_binding
        != approval_binding(
            contract_content_id=checked_invocation.contract_content_id,
            tool_name=checked_invocation.tool_name,
            arguments=checked_invocation.arguments,
            idempotency_key=checked_invocation.idempotency_key,
        )
    ):
        decision = "DENY_APPROVAL"
    elif checked_invocation.call_index > checked_contract.max_calls_per_run:
        decision = "DENY_CALL_BUDGET"
    else:
        decision = "ALLOW"
    return AuditResult(checked_contract, checked_invocation, delegated_scopes, decision)


ILLUSTRATIVE_CONTRACT = ToolContract(
    version="tool-contract-v1",
    name="create_support_credit",
    description="Create one bounded illustrative support-credit request.",
    effect="write",
    parameters=(
        ParameterSpec("amount_cents", "integer", True, 1, 50_000),
        ParameterSpec("customer_id", "string", True, max_length=64),
        ParameterSpec("reason", "string", True, max_length=160),
    ),
    required_scopes=("support:credits:write",),
    approval_mode="per_call",
    idempotency_mode="required",
    max_calls_per_run=1,
    error_codes=("CONFLICT", "INVALID_ARGUMENT", "NOT_AUTHORIZED"),
)

ILLUSTRATIVE_ARGUMENTS = (
    ("amount_cents", 2500),
    ("customer_id", "cust_demo_42"),
    ("reason", "Illustrative service recovery"),
)

ILLUSTRATIVE_INVOCATION = ToolInvocation(
    tool_name="create_support_credit",
    contract_content_id=ILLUSTRATIVE_CONTRACT.content_id,
    arguments=ILLUSTRATIVE_ARGUMENTS,
    granted_scopes=("support:credits:read", "support:credits:write"),
    approved=True,
    approval_binding=approval_binding(
        contract_content_id=ILLUSTRATIVE_CONTRACT.content_id,
        tool_name="create_support_credit",
        arguments=ILLUSTRATIVE_ARGUMENTS,
        idempotency_key="run-0042:credit:1",
    ),
    idempotency_key="run-0042:credit:1",
    call_index=1,
)


def format_example() -> str:
    result = audit(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_INVOCATION)
    return "\n".join(
        (
            "example=illustrative_only",
            f"contract_id={result.contract.content_id}",
            f"tool={result.contract.name}",
            f"effect={result.contract.effect}",
            "schema=VALID",
            "delegated_scopes=" + ",".join(result.delegated_scopes),
            "approval=BOUND_TO_CALL",
            "idempotency=REQUIRED_PRESENT",
            f"decision={result.decision}",
            "claim=LOCAL_CONTRACT_CHECK_ONLY",
            "certification=NOT_A_PRODUCTION_AUTHORIZATION",
        )
    )


def main() -> None:
    print(format_example())


if __name__ == "__main__":
    main()