mail-address-semantics/mas_core/policy.py

"""Pure MAS policy: parse tree + validated statute (+ local registry) -> semantics.

No network access, no clock, no randomness. Without a statute, nothing
beyond syntax is claimed: opaque targets stay `kind = opaque`, resource
syntax stays unconfirmed, interaction preferences stay unconfirmed.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Mapping, Optional

from .model import Facet, Interaction, OpaqueTarget, ParsedAddress, ParsedLocalPart, ResourceTarget, _render
from .statute import Statute

# Kind sources, in the order they are consulted.
SOURCE_STATUTE = "statute"
SOURCE_LOCAL = "local-registry"
SOURCE_NONE = "none"


@dataclass(frozen=True)
class SemanticFailure:
    code: str      # ProfileNotEnabled | InteractionModeNotEnabled | ResourceTypeNotEnabled | UnknownResourceType
    message: str


@dataclass(frozen=True)
class AddressSemantics:
    address: str
    parsed: ParsedLocalPart
    namespace_confirmed: bool
    kind: str
    kind_source: str
    registry_key: Optional[str] = None
    intake_for: Optional[str] = None
    resource_type: Optional[str] = None
    resource_id: Optional[str] = None
    facet: Optional[str] = None
    interaction: Optional[str] = None
    equivalence_key: str = ""
    failures: tuple[SemanticFailure, ...] = ()
    unknowns: tuple[str, ...] = ()
    stand_ins: tuple[str, ...] = ()     # v0.2: "resource" and/or "interaction"

    @property
    def ok(self) -> bool:
        return not self.failures

    def to_dict(self) -> dict:
        return {
            "address": self.address,
            "namespace_confirmed": self.namespace_confirmed,
            "kind": self.kind,
            "kind_source": self.kind_source,
            "registry_key": self.registry_key,
            "intake_for": self.intake_for,
            "resource_type": self.resource_type,
            "resource_id": self.resource_id,
            "facet": self.facet,
            "interaction": self.interaction,
            "required_profile": self.parsed.required_profile.value,
            "equivalence_key": self.equivalence_key,
            "canonical_local_part": str(self.parsed.raw),
            "stand_ins": list(self.stand_ins),
            "failures": [{"code": f.code, "message": f.message} for f in self.failures],
            "unknowns": list(self.unknowns),
        }


def _fold(s: str, case: str, dots: bool) -> str:
    if case == "ascii-insensitive":
        s = s.lower()
    if dots:
        s = s.replace(".", "")
    return s


def equivalence_key(parsed: ParsedLocalPart, statute: Optional[Statute]) -> str:
    """Comparison key for the base target. Exact unless a statute declares folding.

    Dot folding applies to opaque targets only, never to resource identifiers.
    """
    base = parsed.base
    if statute is None:
        return base
    if isinstance(parsed.target, ResourceTarget):
        key = _fold(base, statute.local_part_case, False)
        if statute.resource_id_case == "insensitive":
            key = key.lower()  # resource types are lower-case already (v0.1.1)
        return key
    return _fold(base, statute.local_part_case, statute.dot_folding)


def _split_interaction(value: str, statute: Statute) -> tuple[str, Optional[str]]:
    """`x---MODE` -> (`x`, MODE) when MODE is enabled; split at the last `---`."""
    i = value.rfind("---")
    if i > 0:
        mode = value[i + 3:]
        if mode and not mode.startswith("-") and statute.interaction_mode_enabled(mode):
            return value[:i], mode
    return value, None


def apply_stand_ins(parsed: ParsedLocalPart, statute: Optional[Statute]) -> tuple[ParsedLocalPart, tuple[str, ...]]:
    """Read v0.2 stand-ins as canonical syntax (spec v0.2 section 6.3).

    Only a statute that enables a stand-in gives it meaning. Interaction
    (`---MODE`, enabled mode) is matched first, then resource (`TYPE--ID`,
    declared type, first `--`). Returns the canonical parse and the stand-ins used.
    """
    if statute is None or not (statute.alt_resource or statute.alt_interaction):
        return parsed, ()
    target, facet, interaction = parsed.target, parsed.facet, parsed.interaction
    used: list[str] = []
    if statute.alt_interaction and statute.p2 and interaction is None:
        if facet is not None:
            rest, mode = _split_interaction(facet.value, statute)
            if mode and rest:
                facet, interaction = Facet(rest), Interaction(mode)
        elif isinstance(target, OpaqueTarget):
            rest, mode = _split_interaction(target.value, statute)
            if mode and rest:
                target, interaction = OpaqueTarget(rest), Interaction(mode)
        if interaction is not None:
            used.append("interaction")
    if statute.alt_resource and isinstance(target, OpaqueTarget) and "--" in target.value:
        rtype, rid = target.value.split("--", 1)
        if rtype in statute.resource_types and rid and not rid.startswith("-"):
            target = ResourceTarget(rtype, rid)
            used.append("resource")
    if not used:
        return parsed, ()
    return ParsedLocalPart(raw=_render(target, facet, interaction), target=target,
                           facet=facet, interaction=interaction), tuple(used)


def _lookup(registry: Mapping[str, Mapping], key: str, case: str, dots: bool):
    if key in registry:
        return key, registry[key]
    want = _fold(key, case, dots)
    for k, v in registry.items():
        if _fold(k, case, dots) == want:
            return k, v
    return None, None


def resolve(parsed, statute: Optional[Statute] = None,
            local_registry: Optional[Mapping] = None, *, domain: Optional[str] = None) -> AddressSemantics:
    """Resolve semantics. `local_registry` has the statute's `addresses` /
    `resource_types` shape and applies only where the statute is silent."""
    if isinstance(parsed, ParsedAddress):
        domain, parsed = parsed.domain, parsed.local
    address = f"{parsed.raw}@{domain}" if domain else parsed.raw
    parsed, stand_ins = apply_stand_ins(parsed, statute)
    local = local_registry or {}
    failures: list[SemanticFailure] = []
    unknowns: list[str] = []
    case = statute.local_part_case if statute else "exact"
    dots = statute.dot_folding if statute else False

    kind, source, reg_key, intake_for = "opaque", SOURCE_NONE, None, None
    rtype = rid = None
    target = parsed.target

    if statute is None:
        unknowns.append("namespace: no validated MAS statute; semantics unconfirmed")

    if isinstance(target, ResourceTarget):
        rtype, rid = target.resource_type, target.resource_id
        if statute is not None and rtype in statute.resource_types:
            if statute.resource_type_enabled(rtype):
                kind, source = "resource", SOURCE_STATUTE
            else:
                failures.append(SemanticFailure("ResourceTypeNotEnabled",
                                                f"resource type {rtype!r} is disabled by the statute"))
        elif local.get("resource_types", {}).get(rtype, {}).get("enabled"):
            kind, source = "resource", SOURCE_LOCAL
        elif statute is not None:
            failures.append(SemanticFailure("UnknownResourceType",
                                            f"statute does not declare resource type {rtype!r}"))
        else:
            unknowns.append(f"resource type {rtype!r}: '=' syntax alone confers no application semantics")
    else:
        if statute is not None:
            reg_key, entry = _lookup(statute.addresses, target.value, case, dots)
            if entry:
                kind, source, intake_for = entry["kind"], SOURCE_STATUTE, entry.get("intake_for")
        if source == SOURCE_NONE:
            k, entry = _lookup(local.get("addresses", {}), target.value, case, dots)
            if entry:
                reg_key, kind, source = k, entry["kind"], SOURCE_LOCAL
                intake_for = entry.get("intake_for")
        if source == SOURCE_NONE:
            unknowns.append("kind: no registry entry; punctuation is not a classifier")

    facet = parsed.facet.value if parsed.facet else None
    if facet is not None:
        unknowns.append(f"facet {facet!r}: application-defined; may be stripped in transit")

    mode = parsed.interaction.mode if parsed.interaction else None
    if mode is not None:
        if statute is None:
            unknowns.append(f"interaction {mode!r}: P2 support unconfirmed")
        elif not statute.p2:
            failures.append(SemanticFailure("ProfileNotEnabled",
                                            "address uses '!' (P2) but the statute does not enable P2"))
        elif not statute.interaction_mode_enabled(mode):
            failures.append(SemanticFailure("InteractionModeNotEnabled",
                                            f"interaction mode {mode!r} is not enabled by the statute"))

    return AddressSemantics(
        address=address, parsed=parsed, namespace_confirmed=statute is not None,
        kind=kind, kind_source=source, registry_key=reg_key, intake_for=intake_for,
        resource_type=rtype, resource_id=rid, facet=facet, interaction=mode,
        equivalence_key=equivalence_key(parsed, statute),
        failures=tuple(failures), unknowns=tuple(unknowns), stand_ins=stand_ins,
    )