rose-ash/shared/sx/boundary.py

"""
SX Boundary Enforcement — runtime validation.

Reads declarations from boundary.sx + primitives.sx and validates
that all registered primitives, I/O handlers, and page helpers
are declared in the spec.

Controlled by SX_BOUNDARY_STRICT env var:
  - "1": validation raises errors (fail fast)
  - anything else: validation logs warnings
"""
from __future__ import annotations

import logging
import os
from typing import Any

logger = logging.getLogger("sx.boundary")

# ---------------------------------------------------------------------------
# Lazy-loaded declaration sets (populated on first use)
# ---------------------------------------------------------------------------

_DECLARED_PURE: frozenset[str] | None = None
_DECLARED_IO: frozenset[str] | None = None
_DECLARED_HELPERS: dict[str, frozenset[str]] | None = None


def _load_declarations() -> None:
    global _DECLARED_PURE, _DECLARED_IO, _DECLARED_HELPERS
    if _DECLARED_PURE is not None:
        return
    try:
        from .ref.boundary_parser import parse_primitives_sx, parse_boundary_sx
        _DECLARED_PURE = parse_primitives_sx()
        _DECLARED_IO, _DECLARED_HELPERS = parse_boundary_sx()
        logger.debug(
            "Boundary loaded: %d pure, %d io, %d services",
            len(_DECLARED_PURE), len(_DECLARED_IO), len(_DECLARED_HELPERS),
        )
    except Exception as e:
        logger.warning("Failed to load boundary declarations: %s", e)
        _DECLARED_PURE = frozenset()
        _DECLARED_IO = frozenset()
        _DECLARED_HELPERS = {}


def _is_strict() -> bool:
    return os.environ.get("SX_BOUNDARY_STRICT") == "1"


def _report(message: str) -> None:
    if _is_strict():
        raise RuntimeError(f"SX boundary violation: {message}")
    else:
        logger.warning("SX boundary: %s", message)


# ---------------------------------------------------------------------------
# Validation functions
# ---------------------------------------------------------------------------

def validate_primitive(name: str) -> None:
    """Validate that a pure primitive is declared in primitives.sx."""
    _load_declarations()
    assert _DECLARED_PURE is not None
    if name not in _DECLARED_PURE:
        _report(f"Undeclared pure primitive: {name!r}. Add to primitives.sx.")


def validate_io(name: str) -> None:
    """Validate that an I/O primitive is declared in boundary.sx or boundary-app.sx."""
    _load_declarations()
    assert _DECLARED_IO is not None
    if name not in _DECLARED_IO:
        _report(
            f"Undeclared I/O primitive: {name!r}. "
            f"Add to boundary.sx (core) or boundary-app.sx (deployment)."
        )


def validate_helper(service: str, name: str) -> None:
    """Validate that a page helper is declared in {service}/sx/boundary.sx."""
    _load_declarations()
    assert _DECLARED_HELPERS is not None
    svc_helpers = _DECLARED_HELPERS.get(service, frozenset())
    if name not in svc_helpers:
        _report(
            f"Undeclared page helper: {name!r} for service {service!r}. "
            f"Add to {service}/sx/boundary.sx."
        )


def validate_boundary_value(value: Any, context: str = "") -> None:
    """Validate that a value is an allowed SX boundary type.

    Allowed: int, float, str, bool, None/NIL, list, dict, SxExpr, StyleValue.
    NOT allowed: datetime, ORM models, Quart objects, raw callables.
    """
    from .types import NIL, StyleValue
    from .parser import SxExpr

    if value is None or value is NIL:
        return
    if isinstance(value, (int, float, str, bool)):
        return
    if isinstance(value, SxExpr):
        return
    if isinstance(value, StyleValue):
        return
    if isinstance(value, list):
        for item in value:
            validate_boundary_value(item, context)
        return
    if isinstance(value, dict):
        for k, v in value.items():
            validate_boundary_value(v, context)
        return

    type_name = type(value).__name__
    ctx_msg = f" (in {context})" if context else ""
    _report(
        f"Non-SX type crossing boundary{ctx_msg}: {type_name}. "
        f"Convert to dict/string at the edge."
    )


# ---------------------------------------------------------------------------
# Declaration accessors (for introspection / bootstrapper use)
# ---------------------------------------------------------------------------

def declared_pure() -> frozenset[str]:
    _load_declarations()
    assert _DECLARED_PURE is not None
    return _DECLARED_PURE


def declared_io() -> frozenset[str]:
    _load_declarations()
    assert _DECLARED_IO is not None
    return _DECLARED_IO


def declared_helpers() -> dict[str, frozenset[str]]:
    _load_declarations()
    assert _DECLARED_HELPERS is not None
    return dict(_DECLARED_HELPERS)