orchestrata_ultimate.cxp.schema

Reference › orchestrata_ultimate › cxp › schema

schema holds the JSON Schema for CXP packets and a small validator for it (Spec 004 US-007). CXP_PACKET_SCHEMA describes the 11 core fields, using patterns and limits from constants. SchemaRegistry starts with that schema registered as cxp_packet; register_schema, get_schema and list_schemas manage named schemas. validate_against_schema takes a data dictionary and a schema name and returns a pair of validity flag and error strings; an unknown schema name gives an error and not an exception. The check is implemented in the module itself: _validate_object walks the properties and _validate_value applies type, enum, pattern, string-length and numeric-bound checks, recording the path of each failure. validator delegates schema checks to it. See cxp.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

SchemaRegistry

Decorators: traces(spec='004', story='US-007', reqs=['REQ-004-PERF-007', 'REQ-004-US-007-007', 'REQ-004-US-007-009', 'REQ-004-US-00… · Line 122

Registry for managing and validating against JSON schemas.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/schema.py

# mypy: ignore-errors
"""
CXP JSON Schema Definitions.

Provides JSON schema for CXP packet validation and a lightweight
schema registry for custom schema management per Spec 004 Section US-007.

Author: Mohamed
"""

# Spec: 004 - CXP Protocol

import re
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.cxp.constants import (
    PRIORITY_MAX,
    PRIORITY_MIN,
    SEMVER_PATTERN,
    SUBJECT_MAX_LENGTH,
    UUID_PATTERN,
    VOLK_ALIAS_PATTERN,
)

# ============================================================================
# CXP Packet JSON Schema
# ============================================================================

CXP_PACKET_SCHEMA: dict[str, Any] = {
    "type": "object",
    "description": "CXP packet schema (Spec 004, 11 core fields)",
    "required": [
        "message_id",
        "from_agent",
        "to_agent",
        "thread_id",
        "subject",
        "payload",
        "timestamp",
        "priority",
        "delivery_status",
        "work_status",
        "protocol_version",
    ],
    "properties": {
        "message_id": {
            "type": "string",
            "pattern": UUID_PATTERN,
            "description": "Unique packet identifier (UUID v4)",
        },
        "from_agent": {
            "type": "string",
            "pattern": VOLK_ALIAS_PATTERN,
            "description": "Sender Volk alias (e.g. U:Ox)",
        },
        "to_agent": {
            "type": "string",
            "pattern": VOLK_ALIAS_PATTERN,
            "description": "Recipient Volk alias (e.g. U:Ps)",
        },
        "thread_id": {
            "type": "string",
            "pattern": UUID_PATTERN,
            "description": "Conversation thread identifier (UUID v4)",
        },
        "subject": {
            "type": "string",
            "maxLength": SUBJECT_MAX_LENGTH,
            "description": "Human-readable subject line",
        },
        "payload": {
            "type": ["object", "string", "array", "null"],
            "description": "Message body (object, string, or null)",
        },
        "timestamp": {
            "type": "string",
            "format": "date-time",
            "description": "ISO 8601 UTC timestamp",
        },
        "priority": {
            "type": "integer",
            "minimum": PRIORITY_MIN,
            "maximum": PRIORITY_MAX,
            "description": "Delivery priority (0=background .. 10=urgent)",
        },
        "delivery_status": {
            "type": "string",
            "enum": ["queued", "delivered", "acknowledged", "failed"],
            "description": "Current delivery state",
        },
        "work_status": {
            "type": "string",
            "enum": ["pending", "in_progress", "resolved", "archived"],
            "description": "Current work item state",
        },
        "protocol_version": {
            "type": "string",
            "pattern": SEMVER_PATTERN,
            "description": "CXP protocol version (semver)",
        },
    },
    "additionalProperties": False,
}


# ============================================================================
# Schema Registry
# ============================================================================


@traces(
    spec="004",
    story="US-007",
    reqs=[
        "REQ-004-PERF-007",
        "REQ-004-US-007-007",
        "REQ-004-US-007-009",
        "REQ-004-US-007-010",
    ],
)
class SchemaRegistry:
    """Registry for managing and validating against JSON schemas.

    Provides lightweight validation without requiring the ``jsonschema``
    library.  Checks supported:

    * ``type`` (string, integer, number, boolean, object, array, null,
      and lists of types)
    * ``pattern`` (regex match on string values)
    * ``enum`` (value in allowed set)
    * ``minimum`` / ``maximum`` (numeric bounds)
    * ``maxLength`` / ``minLength`` (string length bounds)
    * ``required`` (presence of keys in objects)
    """

    def __init__(self) -> None:
        self._schemas: dict[str, dict] = {}
        # Pre-register the canonical CXP packet schema
        self._schemas["cxp_packet"] = CXP_PACKET_SCHEMA

    # -- Registry operations ------------------------------------------------

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-010"])
    def register_schema(self, name: str, schema: dict) -> None:
        """Register a schema under *name*, overwriting any existing entry."""
        self._schemas[name] = schema

    def get_schema(self, name: str) -> dict | None:
        """Return the schema registered under *name*, or None."""
        return self._schemas.get(name)

    def list_schemas(self) -> list[str]:
        """Return sorted list of registered schema names."""
        return sorted(self._schemas.keys())

    # -- Validation ---------------------------------------------------------

    @traces(
        spec="004",
        story="US-007",
        reqs=[
            "REQ-004-PERF-007",
            "REQ-004-US-007-009",
            "REQ-004-US-007-010",
        ],
    )
    def validate_against_schema(
        self, data: dict, schema_name: str
    ) -> tuple[bool, list[str]]:
        """Validate *data* against the schema registered as *schema_name*.

        Args:
            data: Dictionary to validate.
            schema_name: Name of a previously registered schema.

        Returns:
            Tuple of (is_valid, list_of_error_strings).
        """
        schema = self._schemas.get(schema_name)
        if schema is None:
            return False, [f"Schema '{schema_name}' not found in registry"]

        errors: list[str] = []
        self._validate_object(data, schema, errors, path="")
        return (len(errors) == 0), errors

    # -- Internal helpers ---------------------------------------------------

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-001"])
    def _validate_object(
        self,
        data: Any,
        schema: dict,
        errors: list[str],
        path: str,
    ) -> None:
        """Recursively validate *data* against *schema*."""

        # --- required fields ---
        required = schema.get("required", [])
        if isinstance(data, dict):
            for field_name in required:
                if field_name not in data:
                    errors.append(
                        f"{path or '(root)'}: missing required field '{field_name}'"
                    )

        # --- per-property checks ---
        properties = schema.get("properties", {})
        if isinstance(data, dict):
            for field_name, field_schema in properties.items():
                if field_name not in data:
                    continue  # missing-required already reported above
                value = data[field_name]
                field_path = f"{path}.{field_name}" if path else field_name
                self._validate_value(value, field_schema, errors, field_path)

        # --- additionalProperties check ---
        additional_properties = schema.get("additionalProperties", True)
        if additional_properties is False and isinstance(data, dict) and properties:
            allowed_keys = set(properties.keys())
            for key in data:
                if key not in allowed_keys:
                    errors.append(
                        f"{path or '(root)'}: unknown property '{key}' "
                        f"(additionalProperties is false)"
                    )

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-002"])
    def _validate_value(
        self,
        value: Any,
        field_schema: dict,
        errors: list[str],
        path: str,
    ) -> None:
        """Validate a single value against its field schema.

        Type is checked first; a type mismatch short-circuits the remaining
        constraints.  Each other constraint group is delegated to a focused
        helper that appends its own errors.
        """

        # --- type check (short-circuits remaining checks on mismatch) ---
        expected_type = field_schema.get("type")
        if expected_type is not None and not self._check_type(value, expected_type):
            errors.append(
                f"{path}: expected type {expected_type}, got {type(value).__name__}"
            )
            return

        self._validate_enum(value, field_schema, errors, path)
        self._validate_pattern(value, field_schema, errors, path)
        self._validate_string_length(value, field_schema, errors, path)
        self._validate_numeric_bounds(value, field_schema, errors, path)

    @staticmethod
    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-004"])
    def _validate_enum(
        value: Any, field_schema: dict, errors: list[str], path: str
    ) -> None:
        """Check *value* against an ``enum`` allow-list, if present."""
        allowed = field_schema.get("enum")
        if allowed is not None and value not in allowed:
            errors.append(f"{path}: value {value!r} not in allowed set {allowed}")

    @staticmethod
    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-003"])
    def _validate_pattern(
        value: Any, field_schema: dict, errors: list[str], path: str
    ) -> None:
        """Check a string *value* against a regex ``pattern``, if present."""
        pattern = field_schema.get("pattern")
        if pattern is not None and isinstance(value, str):
            if not re.search(pattern, value):
                errors.append(
                    f"{path}: value {value!r} does not match pattern {pattern!r}"
                )

    @staticmethod
    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-005"])
    def _validate_string_length(
        value: Any, field_schema: dict, errors: list[str], path: str
    ) -> None:
        """Check a string *value* against ``maxLength``/``minLength``."""
        max_length = field_schema.get("maxLength")
        if max_length is not None and isinstance(value, str):
            if len(value) > max_length:
                errors.append(
                    f"{path}: string length {len(value)} exceeds maxLength {max_length}"
                )

        min_length = field_schema.get("minLength")
        if min_length is not None and isinstance(value, str):
            if len(value) < min_length:
                errors.append(
                    f"{path}: string length {len(value)} is below "
                    f"minLength {min_length}"
                )

    @staticmethod
    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-006"])
    def _validate_numeric_bounds(
        value: Any, field_schema: dict, errors: list[str], path: str
    ) -> None:
        """Check a numeric *value* against ``minimum``/``maximum``."""
        minimum = field_schema.get("minimum")
        if minimum is not None and isinstance(value, (int, float)):
            if value < minimum:
                errors.append(f"{path}: value {value} is below minimum {minimum}")

        maximum = field_schema.get("maximum")
        if maximum is not None and isinstance(value, (int, float)):
            if value > maximum:
                errors.append(f"{path}: value {value} exceeds maximum {maximum}")

    @staticmethod
    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-002"])
    def _check_type(value: Any, expected: Any) -> bool:
        """Check *value* against a JSON-schema type (string or list)."""
        type_map = {
            "string": str,
            "integer": int,
            "number": (int, float),
            "boolean": bool,
            "object": dict,
            "array": list,
            "null": type(None),
        }

        if isinstance(expected, list):
            # Union type, e.g. ["object", "string", "null"]
            return any(SchemaRegistry._check_type(value, t) for t in expected)

        py_type = type_map.get(expected)
        if py_type is None:
            return True  # unknown type keyword -> pass
        # bool is a subtype of int in Python; exclude bools from int/number
        if expected in ("integer", "number") and isinstance(value, bool):
            return False
        return isinstance(value, py_type)