"""Deterministic AI-platform ownership-boundary audit over invented evidence."""

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


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


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 exact_bool(value):
    if type(value) is not bool:
        raise ValueError("exact boolean required")
    return value


def bounded_tuple(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_tuple(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 BoundaryRule:
    capability: str
    expected_plane: str
    interface_id: str
    accountable_owner: str
    runtime_owner: str
    oncall_owner: str
    change_authority: str
    fallback_owner: str
    require_fail_closed: bool

    def __post_init__(self):
        identity(self.capability)
        if type(self.expected_plane) is not str or self.expected_plane not in PLANES:
            raise ValueError("plane must be control or data")
        for value in (
            self.interface_id,
            self.accountable_owner,
            self.runtime_owner,
            self.oncall_owner,
            self.change_authority,
            self.fallback_owner,
        ):
            identity(value)
        exact_bool(self.require_fail_closed)


def boundary_rules(value):
    result = bounded_tuple(value, 1, 32)
    canonical = []
    for rule in result:
        if type(rule) is not BoundaryRule:
            raise ValueError("concrete boundary rule required")
        try:
            rebuilt = BoundaryRule(**asdict(rule))
        except (TypeError, ValueError, OverflowError) as error:
            raise ValueError("malformed boundary rule") from error
        if rule != rebuilt:
            raise ValueError("noncanonical boundary rule")
        canonical.append(rebuilt)
    if len({rule.capability for rule in canonical}) != len(canonical):
        raise ValueError("duplicate boundary rule")
    return tuple(canonical)


@dataclass(frozen=True)
class PlatformBoundaryContract:
    scope: tuple = (
        "platform-charter-v1",
        "capability-map-v1",
        "interface-policy-v1",
        "tenant-policy-v1",
        "operations-policy-v1",
        "fixture-v1",
    )
    rules: tuple = (
        BoundaryRule("identity", "control", "identity-api-v1", "security-platform", "identity-runtime", "identity-oncall", "security-review", "identity-fallback", True),
        BoundaryRule("model-access", "data", "model-access-api-v1", "ai-platform", "model-access-runtime", "model-access-oncall", "ai-platform-review", "model-access-fallback", True),
        BoundaryRule("prompt-policy", "control", "prompt-policy-api-v1", "trust-platform", "prompt-policy-runtime", "prompt-policy-oncall", "policy-review", "prompt-policy-fallback", True),
        BoundaryRule("evaluation", "control", "evaluation-api-v1", "ai-quality", "evaluation-runtime", "evaluation-oncall", "quality-review", "evaluation-fallback", False),
        BoundaryRule("observability", "data", "observability-api-v1", "reliability-platform", "observability-runtime", "observability-oncall", "sre-review", "observability-fallback", False),
        BoundaryRule("incident-response", "control", "incident-response-api-v1", "reliability-platform", "incident-response-runtime", "incident-response-oncall", "incident-command", "incident-response-fallback", False),
    )
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        object.__setattr__(self, "rules", boundary_rules(self.rules))
        expected = seal(self)
        if self.content_id and self.content_id != expected:
            raise ValueError("platform-boundary contract digest mismatch")
        object.__setattr__(self, "content_id", expected)


@dataclass(frozen=True)
class CapabilityBoundary:
    scope: tuple
    capability: str
    plane: str
    interface_id: str
    accountable_owner: str
    runtime_owner: str
    oncall_owner: str
    change_authority: str
    fallback_owner: str
    tenant_context_bound: bool
    least_privilege_reviewed: bool
    telemetry_routed: bool
    fail_closed: bool
    evidence_fresh: bool
    evidence_digest: str
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        identity(self.capability)
        if type(self.plane) is not str or self.plane not in PLANES:
            raise ValueError("plane must be control or data")
        for value in (
            self.interface_id,
            self.accountable_owner,
            self.runtime_owner,
            self.oncall_owner,
            self.change_authority,
            self.fallback_owner,
        ):
            identity(value)
        for value in (
            self.tenant_context_bound,
            self.least_privilege_reviewed,
            self.telemetry_routed,
            self.fail_closed,
            self.evidence_fresh,
        ):
            exact_bool(value)
        sha256_text(self.evidence_digest)
        expected = seal(self)
        if self.content_id and self.content_id != expected:
            raise ValueError("capability-boundary digest mismatch")
        object.__setattr__(self, "content_id", expected)


@dataclass(frozen=True)
class PlatformBoundaryEvidence:
    scope: tuple
    contract_content_id: str
    boundaries: tuple
    review_fresh: bool
    content_id: str = ""

    def __post_init__(self):
        object.__setattr__(self, "scope", scope_tuple(self.scope))
        sha256_text(self.contract_content_id)
        values = bounded_tuple(self.boundaries, 1, 32)
        canonical = []
        for boundary in values:
            rebuilt = validate_record(boundary, CapabilityBoundary)
            if rebuilt.scope != self.scope:
                raise ValueError("capability outside platform scope")
            canonical.append(rebuilt)
        if len({boundary.capability for boundary in canonical}) != len(canonical):
            raise ValueError("duplicate capability boundary")
        exact_bool(self.review_fresh)
        object.__setattr__(self, "boundaries", tuple(canonical))
        expected = seal(self)
        if self.content_id and self.content_id != expected:
            raise ValueError("platform-boundary evidence digest mismatch")
        object.__setattr__(self, "content_id", expected)


@dataclass(frozen=True)
class BoundaryAudit:
    decision: str
    verified_capabilities: tuple
    violations: tuple
    evidence_id: str
    claim: str = "LOCAL_BOUNDARY_AUDIT_NOT_DEPLOYMENT_OR_SECURITY_CERTIFICATION"


def audit_platform_boundaries(
    contract: PlatformBoundaryContract, evidence: PlatformBoundaryEvidence
):
    """Compare declared ownership with immutable invented boundary evidence."""
    contract = validate_record(contract, PlatformBoundaryContract)
    evidence = validate_record(evidence, PlatformBoundaryEvidence)
    if evidence.scope != contract.scope or evidence.contract_content_id != contract.content_id:
        raise ValueError("evidence belongs to another platform-boundary contract")

    rules = {rule.capability: rule for rule in contract.rules}
    actual = {boundary.capability: boundary for boundary in evidence.boundaries}
    violations = []
    for capability in sorted(set(rules) - set(actual)):
        violations.append(f"{capability}/missing-boundary")
    for capability in sorted(set(actual) - set(rules)):
        violations.append(f"{capability}/undeclared-capability")
    for capability in sorted(set(rules) & set(actual)):
        rule = rules[capability]
        boundary = actual[capability]
        if boundary.plane != rule.expected_plane:
            violations.append(f"{capability}/plane")
        if boundary.interface_id != rule.interface_id:
            violations.append(f"{capability}/interface")
        if boundary.accountable_owner != rule.accountable_owner:
            violations.append(f"{capability}/accountable-owner")
        if boundary.runtime_owner != rule.runtime_owner:
            violations.append(f"{capability}/runtime-owner")
        if boundary.oncall_owner != rule.oncall_owner:
            violations.append(f"{capability}/oncall-owner")
        if boundary.change_authority != rule.change_authority:
            violations.append(f"{capability}/change-authority")
        if boundary.fallback_owner != rule.fallback_owner:
            violations.append(f"{capability}/fallback-owner")
        if not boundary.tenant_context_bound:
            violations.append(f"{capability}/tenant-context")
        if not boundary.least_privilege_reviewed:
            violations.append(f"{capability}/least-privilege")
        if not boundary.telemetry_routed:
            violations.append(f"{capability}/telemetry")
        if rule.require_fail_closed and not boundary.fail_closed:
            violations.append(f"{capability}/fail-mode")
        if not boundary.evidence_fresh or not evidence.review_fresh:
            violations.append(f"{capability}/stale-evidence")
    decision = "PASS_BOUNDARY_AUDIT" if not violations else "HOLD_BOUNDARY_AUDIT"
    return BoundaryAudit(
        decision=decision,
        verified_capabilities=tuple(sorted(set(rules) & set(actual))),
        violations=tuple(violations),
        evidence_id=evidence.content_id,
    )


def illustrative_fixture():
    contract = PlatformBoundaryContract()
    boundaries = tuple(
        CapabilityBoundary(
            scope=contract.scope,
            capability=rule.capability,
            plane=rule.expected_plane,
            interface_id=rule.interface_id,
            accountable_owner=rule.accountable_owner,
            runtime_owner=rule.runtime_owner,
            oncall_owner=rule.oncall_owner,
            change_authority=rule.change_authority,
            fallback_owner=rule.fallback_owner,
            tenant_context_bound=True,
            least_privilege_reviewed=True,
            telemetry_routed=True,
            fail_closed=rule.require_fail_closed,
            evidence_fresh=True,
            evidence_digest=token(f"{rule.capability}-evidence"),
        )
        for rule in contract.rules
    )
    return contract, PlatformBoundaryEvidence(
        scope=contract.scope,
        contract_content_id=contract.content_id,
        boundaries=boundaries,
        review_fresh=True,
    )


def main():
    contract, evidence = illustrative_fixture()
    report = audit_platform_boundaries(contract, evidence)
    print("example=illustrative_only")
    print(f"decision={report.decision}")
    print(f"capabilities={','.join(report.verified_capabilities)}")
    print(f"violations={','.join(report.violations)}")
    print(f"claim={report.claim}")


if __name__ == "__main__":
    main()