"""Deterministic serving-control-plane rollout audit over invented 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 bounded_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):
    try:
        encoded = json.dumps(
            value, sort_keys=True, separators=(",", ":"), allow_nan=False
        ).encode()
    except (TypeError, ValueError, OverflowError) as error:
        raise ValueError("canonical JSON value required") from error
    return hashlib.sha256(encoded).hexdigest()


def token(label):
    identity(label)
    return hashlib.sha256(label.encode()).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, ValueError, OverflowError) 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 = bounded_sequence(value, 6, 6)
    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 = (
        "serving-policy-v1",
        "placement-policy-v1",
        "health-schema-v1",
        "autoscaling-policy-v1",
        "rollback-policy-v1",
        "fixture-v1",
    )
    minimum_target_ready_replicas: int = 4
    minimum_observation_windows: int = 3
    max_total_ready_replicas: int = 8
    minimum_model_load_success_rate: float = 0.99
    max_error_rate: float = 0.01
    max_p95_latency_ms: float = 250.0
    max_next_traffic_fraction: float = 0.50
    require_previous_revision: bool = True
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        integer(self.minimum_target_ready_replicas, 1, 100_000)
        integer(self.minimum_observation_windows, 1, 100_000)
        integer(self.max_total_ready_replicas, 1, 100_000)
        if self.max_total_ready_replicas < self.minimum_target_ready_replicas:
            raise ValueError("ready-replica capacity is internally inconsistent")
        finite_float(self.minimum_model_load_success_rate, 0.0, 1.0)
        finite_float(self.max_error_rate, 0.0, 1.0)
        finite_float(
            self.max_p95_latency_ms, 0.0, 86_400_000.0, lower_inclusive=False
        )
        finite_float(
            self.max_next_traffic_fraction, 0.0, 1.0, lower_inclusive=False
        )
        if exact_bool(self.require_previous_revision) is not True:
            raise ValueError("a rollback revision is required")
        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 ServingRevision:
    scope: tuple
    model_id: str
    revision: str
    model_digest: str
    config_digest: str
    desired_replicas: int
    ready_replicas: int
    model_load_attempts: int
    model_load_successes: int
    request_count: int
    error_count: int
    p95_latency_ms: float
    health_fresh: bool
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        identity(self.model_id)
        identity(self.revision)
        sha256_text(self.model_digest)
        sha256_text(self.config_digest)
        integer(self.desired_replicas, 1, 100_000)
        integer(self.ready_replicas, 0, self.desired_replicas)
        integer(self.model_load_attempts, 1, 100_000_000)
        integer(self.model_load_successes, 0, self.model_load_attempts)
        integer(self.request_count, 1, 1_000_000_000)
        integer(self.error_count, 0, self.request_count)
        finite_float(self.p95_latency_ms, 0.0, 86_400_000.0, lower_inclusive=False)
        exact_bool(self.health_fresh)
        expected = seal(self)
        if self.content_id and self.content_id != expected:
            raise ValueError("serving revision digest mismatch")
        object.__setattr__(self, "content_id", expected)


@dataclass(frozen=True)
class RolloutEvidence:
    scope: tuple
    contract_content_id: str
    target: ServingRevision
    previous: ServingRevision
    placement_verified: bool
    autoscaler_ready: bool
    rollback_route_ready: bool
    current_traffic_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)
        target = validate_record(self.target, ServingRevision)
        previous = validate_record(self.previous, ServingRevision)
        if target.scope != self.scope or previous.scope != self.scope:
            raise ValueError("revision evidence outside rollout scope")
        if target.model_id != previous.model_id:
            raise ValueError("revisions belong to different models")
        if target.revision == previous.revision:
            raise ValueError("target and previous revisions must differ")
        exact_bool(self.placement_verified)
        exact_bool(self.autoscaler_ready)
        exact_bool(self.rollback_route_ready)
        finite_float(
            self.current_traffic_fraction, 0.0, 1.0, lower_inclusive=False
        )
        integer(self.observation_windows, 1, 100_000)
        object.__setattr__(self, "target", target)
        object.__setattr__(self, "previous", previous)
        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 RolloutPlan:
    decision: str
    violations: tuple
    target_revision: str
    previous_revision: str
    target_ready_replicas: int
    target_desired_replicas: int
    model_load_success_rate: float
    error_rate: float
    p95_latency_ms: float
    next_traffic_fraction: float
    evidence_id: str
    claim: str = "LOCAL_CONTROL_PLANE_AUDIT_NOT_DEPLOYMENT"


def plan_rollout(contract: RolloutContract, evidence: RolloutEvidence) -> RolloutPlan:
    """Plan one bounded transition without contacting or mutating a serving system."""
    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_traffic_fraction > contract.max_next_traffic_fraction:
        raise ValueError("current traffic exceeds this contract's rollout boundary")

    target = evidence.target
    previous = evidence.previous
    load_success_rate = target.model_load_successes / target.model_load_attempts
    error_rate = target.error_count / target.request_count

    readiness = []
    if not evidence.placement_verified:
        readiness.append("placement-unverified")
    if not evidence.autoscaler_ready:
        readiness.append("autoscaler-unready")
    if not target.health_fresh or not previous.health_fresh:
        readiness.append("stale-health")
    if (
        target.ready_replicas < contract.minimum_target_ready_replicas
        or target.ready_replicas != target.desired_replicas
    ):
        readiness.append("target-readiness")
    if load_success_rate < contract.minimum_model_load_success_rate:
        readiness.append("model-load-success")
    if evidence.observation_windows < contract.minimum_observation_windows:
        readiness.append("observation-windows")
    if (
        target.desired_replicas + previous.desired_replicas
        > contract.max_total_ready_replicas
    ):
        readiness.append("surge-capacity")
    if contract.require_previous_revision and (
        not evidence.rollback_route_ready
        or previous.ready_replicas != previous.desired_replicas
    ):
        readiness.append("rollback-unready")
    if evidence.current_traffic_fraction == contract.max_next_traffic_fraction:
        readiness.append("traffic-ceiling")

    regressions = []
    if error_rate > contract.max_error_rate:
        regressions.append("error-rate")
    if target.p95_latency_ms > contract.max_p95_latency_ms:
        regressions.append("p95-latency")

    violations = tuple(readiness + regressions)
    if regressions:
        decision = "ROLLBACK"
        next_fraction = 0.0
    elif readiness:
        decision = "HOLD"
        next_fraction = evidence.current_traffic_fraction
    else:
        decision = "ADVANCE_BOUNDED_ROLLOUT"
        next_fraction = min(
            evidence.current_traffic_fraction * 2.0,
            contract.max_next_traffic_fraction,
        )

    evidence_id = digest(
        {
            "contract": contract.content_id,
            "evidence": evidence.content_id,
            "decision": decision,
            "violations": violations,
            "next_fraction": f"{next_fraction:.12f}",
        }
    )
    return RolloutPlan(
        decision,
        violations,
        target.revision,
        previous.revision,
        target.ready_replicas,
        target.desired_replicas,
        load_success_rate,
        error_rate,
        target.p95_latency_ms,
        next_fraction,
        evidence_id,
    )


def illustrative_fixture():
    contract = RolloutContract()
    target = ServingRevision(
        contract.scope,
        "support-summarizer",
        "model-v18",
        token("invented-model-v18"),
        token("invented-config-v18"),
        4,
        4,
        4,
        4,
        10_000,
        20,
        180.0,
        True,
    )
    previous = ServingRevision(
        contract.scope,
        "support-summarizer",
        "model-v17",
        token("invented-model-v17"),
        token("invented-config-v17"),
        4,
        4,
        4,
        4,
        10_000,
        15,
        175.0,
        True,
    )
    evidence = RolloutEvidence(
        contract.scope,
        contract.content_id,
        target,
        previous,
        True,
        True,
        True,
        0.10,
        3,
    )
    return contract, evidence


def main():
    contract, evidence = illustrative_fixture()
    report = plan_rollout(contract, evidence)
    print("example=illustrative_only")
    print(f"decision={report.decision}")
    print(
        f"target_revision={report.target_revision};"
        f"previous_revision={report.previous_revision}"
    )
    print(
        f"target_ready={report.target_ready_replicas}/"
        f"{report.target_desired_replicas};"
        f"load_success_rate={report.model_load_success_rate:.3f}"
    )
    print(
        f"error_rate={report.error_rate:.3f};"
        f"p95_latency_ms={report.p95_latency_ms:.3f}"
    )
    print(f"next_traffic_fraction={report.next_traffic_fraction:.3f}")
    print(f"claim={report.claim}")


if __name__ == "__main__":
    main()