"""Audit bounded ANN evidence against declared gates; do not simulate an ANN index."""
from dataclasses import asdict, dataclass, fields
import hashlib
import json
import math
import sys


def text(value):
    if type(value) is not str or not value.strip() or len(value) > 4096:
        raise ValueError("nonempty bounded built-in string required")
    return value


def sha256_hex(value):
    if type(value) is not str or len(value) != 64 or any(c not in "0123456789abcdef" for c in value):
        raise ValueError("lowercase SHA-256 required")
    return value


def integer(value, lower, upper):
    if type(value) is not int or not lower <= value <= upper:
        raise ValueError("bounded integer, not bool, required")
    return value


def number(value, lower, upper):
    if type(value) not in (int, float) or not math.isfinite(value):
        raise ValueError("finite real, not bool, required")
    if value != 0 and abs(value) < sys.float_info.min:
        raise ValueError("subnormal rejected")
    if not lower <= value <= upper:
        raise ValueError("number out of range")
    return float(value)


def digest(value):
    return hashlib.sha256(
        json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
    ).hexdigest()


def _identity_sequence(value, label):
    if type(value) not in (tuple, list) or not 1 <= len(value) <= 100:
        raise ValueError(f"bounded {label} sequence required")
    copied = tuple(text(item) for item in value)
    if len(set(copied)) != len(copied):
        raise ValueError(f"duplicate {label} identity")
    return copied


@dataclass(frozen=True)
class QueryEvidence:
    query_id: str
    slice_name: str
    exact_ids: tuple[str, ...]
    returned_ids: tuple[str, ...]
    latency_ms: float

    def __post_init__(self):
        text(self.query_id)
        text(self.slice_name)
        if self.slice_name not in ("unfiltered", "filtered"):
            raise ValueError("unsupported query slice")
        object.__setattr__(self, "exact_ids", _identity_sequence(self.exact_ids, "exact neighbor"))
        object.__setattr__(self, "returned_ids", _identity_sequence(self.returned_ids, "returned neighbor"))
        number(self.latency_ms, 0, 1_000_000_000)


@dataclass(frozen=True)
class IndexEvidence:
    candidate_id: str
    algorithm: str
    index_revision: str
    index_digest: str
    corpus_digest: str
    embedding_contract_digest: str
    parameter_digest: str
    build_seconds: float
    memory_bytes: int
    update_p95_seconds: float
    deletion_pass: bool
    queries: tuple[QueryEvidence, ...]

    def __post_init__(self):
        for name in ("candidate_id", "algorithm", "index_revision"):
            text(getattr(self, name))
        if self.algorithm not in ("hnsw", "ivf-flat", "ivf-pq"):
            raise ValueError("unsupported ANN family")
        for name in ("index_digest", "corpus_digest", "embedding_contract_digest", "parameter_digest"):
            sha256_hex(getattr(self, name))
        number(self.build_seconds, 0, 10**12)
        integer(self.memory_bytes, 1, 10**15)
        number(self.update_p95_seconds, 0, 10**12)
        if type(self.deletion_pass) is not bool:
            raise ValueError("deletion_pass must be bool")
        if type(self.queries) not in (tuple, list) or not 1 <= len(self.queries) <= 100_000:
            raise ValueError("bounded query evidence required")
        copied = []
        for item in self.queries:
            if type(item) is not QueryEvidence:
                raise ValueError("concrete QueryEvidence required")
            copied.append(QueryEvidence(**{field.name: getattr(item, field.name) for field in fields(QueryEvidence)}))
        object.__setattr__(self, "queries", tuple(copied))
        if len({item.query_id for item in copied}) != len(copied):
            raise ValueError("duplicate query identity")


def _query_payload(queries):
    return [
        {"query_id": item.query_id, "slice_name": item.slice_name, "exact_ids": item.exact_ids}
        for item in queries
    ]


@dataclass(frozen=True)
class AnnDecisionContract:
    contract_id: str
    scope: str
    version: str
    corpus_digest: str
    embedding_contract_digest: str
    exact_baseline_revision: str
    query_set_digest: str
    k: int
    min_queries_per_slice: int
    min_recall: float
    min_filtered_recall: float
    max_p95_latency_ms: float
    max_memory_bytes: int
    max_build_seconds: float
    max_update_p95_seconds: float
    require_deletion_pass: bool
    selection_rule: str
    owner: str
    rollback: str
    candidates: tuple[IndexEvidence, ...]

    def __post_init__(self):
        for name in ("contract_id", "scope", "version", "exact_baseline_revision", "selection_rule", "owner", "rollback"):
            text(getattr(self, name))
        if self.selection_rule != "lowest-memory-then-latency":
            raise ValueError("unsupported selection rule")
        for name in ("corpus_digest", "embedding_contract_digest", "query_set_digest"):
            sha256_hex(getattr(self, name))
        integer(self.k, 1, 100)
        integer(self.min_queries_per_slice, 1, 100_000)
        number(self.min_recall, 0, 1)
        number(self.min_filtered_recall, 0, 1)
        number(self.max_p95_latency_ms, 0, 10**12)
        integer(self.max_memory_bytes, 1, 10**15)
        number(self.max_build_seconds, 0, 10**12)
        number(self.max_update_p95_seconds, 0, 10**12)
        if type(self.require_deletion_pass) is not bool:
            raise ValueError("require_deletion_pass must be bool")
        if type(self.candidates) not in (tuple, list) or not 1 <= len(self.candidates) <= 100:
            raise ValueError("bounded candidate evidence required")
        copied = []
        for item in self.candidates:
            if type(item) is not IndexEvidence:
                raise ValueError("concrete IndexEvidence required")
            copied.append(IndexEvidence(**{field.name: getattr(item, field.name) for field in fields(IndexEvidence)}))
        object.__setattr__(self, "candidates", tuple(copied))
        if len({item.candidate_id for item in copied}) != len(copied):
            raise ValueError("duplicate candidate identity")
        baseline_payload = _query_payload(copied[0].queries)
        if digest(baseline_payload) != self.query_set_digest:
            raise ValueError("query set content mismatch")
        for candidate in copied:
            if candidate.corpus_digest != self.corpus_digest or candidate.embedding_contract_digest != self.embedding_contract_digest:
                raise ValueError("candidate data or embedding binding mismatch")
            if _query_payload(candidate.queries) != baseline_payload:
                raise ValueError("exact baseline mismatch across candidates")
            if any(len(item.exact_ids) != self.k or len(item.returned_ids) != self.k for item in candidate.queries):
                raise ValueError("every query must bind exact top-k and returned top-k")
            for slice_name in ("unfiltered", "filtered"):
                if sum(item.slice_name == slice_name for item in candidate.queries) < self.min_queries_per_slice:
                    raise ValueError("insufficient slice evidence")


@dataclass(frozen=True)
class CandidateReport:
    candidate_id: str
    algorithm: str
    recall: float
    filtered_recall: float
    p95_latency_ms: float
    memory_bytes: int
    qualified: bool
    blockers: tuple[str, ...]


@dataclass(frozen=True)
class AnnDecision:
    contract_digest: str
    selected: str
    reports: tuple[CandidateReport, ...]
    claim: str


def _mean_recall(queries, slice_name, k):
    selected = [item for item in queries if item.slice_name == slice_name]
    return sum(len(set(item.exact_ids).intersection(item.returned_ids)) / k for item in selected) / len(selected)


def _p95(values):
    ordered = sorted(values)
    return ordered[math.ceil(0.95 * len(ordered)) - 1]


def audit(contract):
    if type(contract) is not AnnDecisionContract:
        raise ValueError("concrete AnnDecisionContract required")
    contract = AnnDecisionContract(**{field.name: getattr(contract, field.name) for field in fields(AnnDecisionContract)})
    reports = []
    for candidate in contract.candidates:
        recall = _mean_recall(candidate.queries, "unfiltered", contract.k)
        filtered_recall = _mean_recall(candidate.queries, "filtered", contract.k)
        p95 = _p95([item.latency_ms for item in candidate.queries])
        blockers = []
        if recall < contract.min_recall:
            blockers.append("recall")
        if filtered_recall < contract.min_filtered_recall:
            blockers.append("filtered-recall")
        if p95 > contract.max_p95_latency_ms:
            blockers.append("latency")
        if candidate.memory_bytes > contract.max_memory_bytes:
            blockers.append("memory")
        if candidate.build_seconds > contract.max_build_seconds:
            blockers.append("build")
        if candidate.update_p95_seconds > contract.max_update_p95_seconds:
            blockers.append("update")
        if contract.require_deletion_pass and not candidate.deletion_pass:
            blockers.append("deletion")
        reports.append(CandidateReport(candidate.candidate_id, candidate.algorithm, recall, filtered_recall, p95,
                                       candidate.memory_bytes, not blockers, tuple(blockers)))
    qualified = [report for report in reports if report.qualified]
    selected = min(qualified, key=lambda item: (item.memory_bytes, item.p95_latency_ms, item.candidate_id)).candidate_id if qualified else "BLOCK"
    return AnnDecision(digest(asdict(contract)), selected, tuple(reports), "FIXTURE_DECISION_NOT_A_BENCHMARK")


def _queries(latencies, returned):
    exact = (
        ("u-1", "unfiltered", ("a", "b")),
        ("u-2", "unfiltered", ("c", "d")),
        ("f-1", "filtered", ("e", "f")),
        ("f-2", "filtered", ("g", "h")),
    )
    return tuple(QueryEvidence(query_id, slice_name, gold, found, latency)
                 for (query_id, slice_name, gold), found, latency in zip(exact, returned, latencies))


def example_contract():
    corpus = digest({"corpus": "synthetic-v1"})
    embeddings = digest({"embedding-contract": "example-v1"})
    hnsw_queries = _queries((4.0, 5.0, 4.5, 5.0), (("a", "b"), ("c", "d"), ("e", "f"), ("g", "h")))
    ivf_queries = _queries((7.0, 8.0, 7.5, 8.0), (("a", "x"), ("c", "d"), ("e", "x"), ("g", "x")))
    candidates = (
        IndexEvidence("hnsw-m16", "hnsw", "index-1", digest({"index": "hnsw"}), corpus, embeddings,
                      digest({"M": 16, "efSearch": 64}), 20.0, 8_000_000, 2.0, True, hnsw_queries),
        IndexEvidence("ivf-64", "ivf-flat", "index-2", digest({"index": "ivf"}), corpus, embeddings,
                      digest({"nlist": 64, "nprobe": 4}), 10.0, 3_000_000, 8.0, True, ivf_queries),
    )
    return AnnDecisionContract(
        "support-ann-v1", "illustrative support retrieval", "1", corpus, embeddings, "exact-flat-v1",
        digest(_query_payload(hnsw_queries)), 2, 2, 0.75, 0.75, 10.0, 10_000_000, 30.0, 5.0, True,
        "lowest-memory-then-latency", "example-search-team", "restore exact-flat-v1", candidates,
    )


def main():
    result = audit(example_contract())
    selected = next(report for report in result.reports if report.candidate_id == result.selected)
    print("example=illustrative_only")
    print("selected=" + result.selected)
    print(f"qualified={sum(report.qualified for report in result.reports)}/{len(result.reports)}")
    print(f"selected_recall={selected.recall:.3f}")
    print(f"selected_p95_latency_ms={selected.p95_latency_ms:.3f}")
    print("claim=" + result.claim)


if __name__ == "__main__":
    main()