"""Deterministic online-evaluation rollout gate for invented cohort evidence."""

from dataclasses import asdict, dataclass, fields
import hashlib
import json
import math
import re


IDENTITY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,127}")
SHA256 = re.compile(r"[a-f0-9]{64}")


def identity(value):
    if type(value) is not str or not IDENTITY.fullmatch(value):
        raise ValueError("exact bounded identity required")
    return value


def sha256_text(value):
    if type(value) is not str or not SHA256.fullmatch(value):
        raise ValueError("exact 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 non-boolean integer required")
    return value


def finite_float(value, lower, upper, *, lower_inclusive=True):
    if type(value) is not float or not math.isfinite(value):
        raise ValueError("exact finite float required")
    lower_ok = value >= lower if lower_inclusive else value > lower
    if not lower_ok or value > upper:
        raise ValueError("bounded finite float required")
    return value


def exact_bool(value):
    if type(value) is not bool:
        raise ValueError("exact boolean required")
    return value


def sequence(value, lower, upper):
    if type(value) not in (tuple, list):
        raise ValueError("bounded tuple or list required")
    integer(len(value), lower, upper)
    return tuple(value)


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


def seal(record):
    if type(record.content_id) is not str:
        raise ValueError("content digest requires exact string")
    data = asdict(record)
    data.pop("content_id")
    return digest(data)


def validate_record(record, cls):
    if type(record) is not cls:
        raise ValueError("concrete frozen record required")
    try:
        rebuilt = cls(**{field.name: getattr(record, field.name) for field in fields(cls)})
    except (AttributeError, TypeError) as error:
        raise ValueError("malformed record") from error
    if record != rebuilt or record.content_id != rebuilt.content_id:
        raise ValueError("noncanonical or modified record")
    return rebuilt


def scope_tuple(value):
    result = sequence(value, 8, 8)
    for item in result:
        identity(item)
    if len(set(result)) != len(result):
        raise ValueError("duplicate scope identity")
    return result


@dataclass(frozen=True)
class RolloutContract:
    scope: tuple = (
        "rollout-policy-v1",
        "assignment-v1",
        "quality-metric-v1",
        "safety-metric-v1",
        "latency-metric-v1",
        "cost-metric-v1",
        "telemetry-v1",
        "fixture-v1",
    )
    minimum_samples_per_cohort: int = 1_000
    minimum_observation_windows: int = 3
    max_quality_regression: float = 0.01
    max_safety_event_rate: float = 0.005
    max_latency_ratio: float = 1.20
    max_cost_ratio: float = 1.15
    max_next_canary_fraction: float = 0.25
    require_shadow: bool = True
    require_holdback: bool = True
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        integer(self.minimum_samples_per_cohort, 1, 10_000_000)
        integer(self.minimum_observation_windows, 1, 10_000)
        finite_float(self.max_quality_regression, 0.0, 1.0)
        finite_float(self.max_safety_event_rate, 0.0, 1.0)
        finite_float(self.max_latency_ratio, 1.0, 100.0)
        finite_float(self.max_cost_ratio, 0.01, 100.0)
        finite_float(self.max_next_canary_fraction, 0.0, 1.0, lower_inclusive=False)
        if exact_bool(self.require_shadow) is not True:
            raise ValueError("online gate requires shadow evidence")
        if exact_bool(self.require_holdback) is not True:
            raise ValueError("online gate requires a holdback")
        expected = seal(self)
        if self.content_id and self.content_id != expected:
            raise ValueError("rollout contract digest mismatch")
        object.__setattr__(self, "content_id", expected)


@dataclass(frozen=True)
class CohortMetrics:
    scope: tuple
    experiment_id: str
    cohort: str
    sample_size: int
    quality_successes: int
    safety_events: int
    p95_latency_ms: float
    mean_cost_usd: float
    telemetry_complete: bool
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        identity(self.experiment_id)
        if type(self.cohort) is not str or self.cohort not in ("candidate", "control"):
            raise ValueError("cohort must be candidate or control")
        integer(self.sample_size, 1, 100_000_000)
        integer(self.quality_successes, 0, self.sample_size)
        integer(self.safety_events, 0, self.sample_size)
        finite_float(self.p95_latency_ms, 0.0, 86_400_000.0, lower_inclusive=False)
        finite_float(self.mean_cost_usd, 0.0, 1_000_000.0, lower_inclusive=False)
        exact_bool(self.telemetry_complete)
        expected = seal(self)
        if self.content_id and self.content_id != expected:
            raise ValueError("cohort metrics digest mismatch")
        object.__setattr__(self, "content_id", expected)


@dataclass(frozen=True)
class RolloutEvidence:
    scope: tuple
    contract_content_id: str
    candidate: CohortMetrics
    control: CohortMetrics
    shadow_complete: bool
    holdback_preserved: bool
    current_canary_fraction: float
    observation_windows: int
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        sha256_text(self.contract_content_id)
        candidate = validate_record(self.candidate, CohortMetrics)
        control = validate_record(self.control, CohortMetrics)
        if candidate.scope != self.scope or control.scope != self.scope:
            raise ValueError("cohort evidence outside rollout scope")
        if candidate.cohort != "candidate" or control.cohort != "control":
            raise ValueError("candidate and control roles are fixed")
        if candidate.experiment_id != control.experiment_id:
            raise ValueError("cohorts belong to different experiments")
        exact_bool(self.shadow_complete)
        exact_bool(self.holdback_preserved)
        finite_float(self.current_canary_fraction, 0.0, 1.0, lower_inclusive=False)
        integer(self.observation_windows, 1, 10_000)
        object.__setattr__(self, "candidate", candidate)
        object.__setattr__(self, "control", control)
        expected = seal(self)
        if self.content_id and self.content_id != expected:
            raise ValueError("rollout evidence digest mismatch")
        object.__setattr__(self, "content_id", expected)


@dataclass(frozen=True)
class RolloutReport:
    decision: str
    violations: tuple
    candidate_quality: float
    control_quality: float
    safety_rate: float
    latency_ratio: float
    cost_ratio: float
    next_canary_fraction: float
    evidence_id: str
    claim: str = "LOCAL_ROLLOUT_GATE_NOT_CAUSAL_PROOF"


def audit_rollout(contract: RolloutContract, evidence: RolloutEvidence) -> RolloutReport:
    """Apply a declared gate to aggregated, invented candidate and holdback metrics."""
    contract = validate_record(contract, RolloutContract)
    evidence = validate_record(evidence, RolloutEvidence)
    if evidence.scope != contract.scope or evidence.contract_content_id != contract.content_id:
        raise ValueError("evidence belongs to another rollout contract")
    if evidence.current_canary_fraction > contract.max_next_canary_fraction:
        raise ValueError("current canary exceeds this contract's bounded rollout stage")

    candidate = evidence.candidate
    control = evidence.control
    candidate_quality = candidate.quality_successes / candidate.sample_size
    control_quality = control.quality_successes / control.sample_size
    safety_rate = candidate.safety_events / candidate.sample_size
    latency_ratio = candidate.p95_latency_ms / control.p95_latency_ms
    cost_ratio = candidate.mean_cost_usd / control.mean_cost_usd

    readiness = []
    if not evidence.shadow_complete:
        readiness.append("shadow-incomplete")
    if not evidence.holdback_preserved:
        readiness.append("holdback-missing")
    if not candidate.telemetry_complete or not control.telemetry_complete:
        readiness.append("telemetry-incomplete")
    if min(candidate.sample_size, control.sample_size) < contract.minimum_samples_per_cohort:
        readiness.append("sample-size")
    if evidence.observation_windows < contract.minimum_observation_windows:
        readiness.append("observation-windows")

    regressions = []
    if control_quality - candidate_quality > contract.max_quality_regression:
        regressions.append("quality-regression")
    if safety_rate > contract.max_safety_event_rate:
        regressions.append("safety-rate")
    if latency_ratio > contract.max_latency_ratio:
        regressions.append("latency-ratio")
    if cost_ratio > contract.max_cost_ratio:
        regressions.append("cost-ratio")

    violations = tuple(readiness + regressions)
    if regressions:
        decision = "ROLLBACK"
        next_fraction = 0.0
    elif readiness:
        decision = "HOLD"
        next_fraction = evidence.current_canary_fraction
    else:
        decision = "ADVANCE_CANARY"
        next_fraction = min(
            evidence.current_canary_fraction * 2.0,
            contract.max_next_canary_fraction,
        )
    evidence_id = digest(
        {
            "contract": contract.content_id,
            "evidence": evidence.content_id,
            "decision": decision,
            "violations": violations,
            "next_fraction": f"{next_fraction:.12f}",
        }
    )
    return RolloutReport(
        decision,
        violations,
        candidate_quality,
        control_quality,
        safety_rate,
        latency_ratio,
        cost_ratio,
        next_fraction,
        evidence_id,
    )


def illustrative_fixture():
    contract = RolloutContract()
    experiment_id = "support-summary-v7"
    candidate = CohortMetrics(
        contract.scope,
        experiment_id,
        "candidate",
        1_000,
        940,
        2,
        108.0,
        1.05,
        True,
    )
    control = CohortMetrics(
        contract.scope,
        experiment_id,
        "control",
        1_000,
        935,
        1,
        100.0,
        1.00,
        True,
    )
    evidence = RolloutEvidence(
        contract.scope,
        contract.content_id,
        candidate,
        control,
        True,
        True,
        0.05,
        3,
    )
    return contract, evidence


def main():
    report = audit_rollout(*illustrative_fixture())
    print("example=illustrative_only")
    print(f"decision={report.decision}")
    print(
        f"candidate_quality={report.candidate_quality:.3f};"
        f"control_quality={report.control_quality:.3f}"
    )
    print(
        f"safety_rate={report.safety_rate:.3f};"
        f"latency_ratio={report.latency_ratio:.3f};cost_ratio={report.cost_ratio:.3f}"
    )
    print(f"next_canary_fraction={report.next_canary_fraction:.3f}")
    print(f"claim={report.claim}")


if __name__ == "__main__":
    main()