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


_LOAD_FAILURE_WARNED = False


def _load_declarations() -> None:
    # KNOWN-BROKEN (2026-07 review, finding C24): `.ref.boundary_parser` was
    # moved to hosts/python/boundary_parser.py AND its tier-1 declaration
    # source (shared/sx/ref/boundary.sx) was deleted, so this ImportError
    # fires on every call and boundary validation has been a silent no-op —
    # including under SX_BOUNDARY_STRICT=1, which production compose sets.
    # Do NOT "fix" the import in isolation: reviving validation while strict
    # mode is live in production requires first recreating the core
    # declarations and proving zero violations across all services
    # (remediation plan, Phase 2 "Python boundary"). Until then we make the
    # dead state visible instead of silent.
    global _DECLARED_PURE, _DECLARED_IO, _DECLARED_HELPERS, _LOAD_FAILURE_WARNED
    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:
        # Don't cache failure — parser may not be ready yet (circular import
        # during startup). Will retry on next call. Validation functions
        # skip checks when declarations aren't loaded.
        if not _LOAD_FAILURE_WARNED:
            _LOAD_FAILURE_WARNED = True
            logger.warning(
                "SX boundary validation is INACTIVE (declarations failed to "
                "load: %s). All validate_* calls are no-ops, even under "
                "SX_BOUNDARY_STRICT=1.", e,
            )
        else:
            logger.debug("Boundary declarations not ready yet: %s", e)


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()
    if _DECLARED_PURE is None:
        return  # Not ready yet (circular import during startup), skip
    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()
    if _DECLARED_IO is None:
        return  # Not ready yet, skip
    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()
    if _DECLARED_HELPERS is None:
        return  # Not ready yet, skip
    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.
    NOT allowed: datetime, ORM models, Quart objects, raw callables.
    """
    from .types import NIL
    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, 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()
    return _DECLARED_PURE or frozenset()


def declared_io() -> frozenset[str]:
    _load_declarations()
    return _DECLARED_IO or frozenset()


def declared_helpers() -> dict[str, frozenset[str]]:
    _load_declarations()
    return dict(_DECLARED_HELPERS) if _DECLARED_HELPERS else {}