d3cb6e9045
Schemas, examples and a flag registry for the twelve HUX cards, a dependency-free validator, the governance rules (memory ledger, autonomy matrix, friendly modes mapped to real Switchyard routes, privacy defaults, suggestion gating, release state machine) and the contract doc UI work codes against. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM
238 lines
9.3 KiB
Python
238 lines
9.3 KiB
Python
"""Load and validate the HUX (Hermes user-experience) contract schemas.
|
|
|
|
The schemas under ``services/hermes/contracts/hux`` are plain JSON Schema
|
|
2020-12 so browser and Go consumers can validate with their usual libraries.
|
|
CI has no ``jsonschema`` package, so this module carries a small validator for
|
|
the keyword subset the contracts actually use. Unsupported keywords fail
|
|
loudly rather than silently passing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
CONTRACT_DIR = Path(__file__).resolve().parents[1] / "contracts" / "hux"
|
|
SCHEMA_FILES = (
|
|
"common.schema.json",
|
|
"event.schema.json",
|
|
"memory.schema.json",
|
|
"project.schema.json",
|
|
"artifact.schema.json",
|
|
"permission.schema.json",
|
|
"mode.schema.json",
|
|
"citation.schema.json",
|
|
"suggestion.schema.json",
|
|
"privacy.schema.json",
|
|
"release.schema.json",
|
|
)
|
|
SUPPORTED_KEYWORDS = frozenset(
|
|
{
|
|
"$schema", "$id", "$defs", "$ref", "title", "description",
|
|
"type", "const", "enum", "required", "properties",
|
|
"additionalProperties", "items", "minItems", "maxItems",
|
|
"uniqueItems", "minLength", "maxLength", "pattern",
|
|
"minimum", "maximum", "oneOf",
|
|
}
|
|
)
|
|
_TYPE_CHECKS = {
|
|
"object": lambda v: isinstance(v, dict),
|
|
"array": lambda v: isinstance(v, list),
|
|
"string": lambda v: isinstance(v, str),
|
|
"boolean": lambda v: isinstance(v, bool),
|
|
"integer": lambda v: isinstance(v, int) and not isinstance(v, bool),
|
|
"number": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
|
|
"null": lambda v: v is None,
|
|
}
|
|
|
|
|
|
class ContractError(ValueError):
|
|
"""Raised when a schema uses something this validator does not support."""
|
|
|
|
|
|
def load_schema(name: str, directory: Path = CONTRACT_DIR) -> dict[str, Any]:
|
|
"""Read one schema file by name."""
|
|
return json.loads((directory / name).read_text(encoding="utf-8"))
|
|
|
|
|
|
def load_all(directory: Path = CONTRACT_DIR) -> dict[str, dict[str, Any]]:
|
|
"""Read every contract schema keyed by file name."""
|
|
return {name: load_schema(name, directory) for name in SCHEMA_FILES}
|
|
|
|
|
|
def load_flags(directory: Path = CONTRACT_DIR) -> dict[str, Any]:
|
|
"""Read the feature flag registry."""
|
|
return json.loads((directory / "flags.json").read_text(encoding="utf-8"))
|
|
|
|
|
|
def _walk(node: Any, path: str, problems: list[str]) -> None:
|
|
if isinstance(node, dict):
|
|
for key, value in node.items():
|
|
if path.endswith("/properties") or path.endswith("/$defs"):
|
|
_walk(value, f"{path}/{key}", problems)
|
|
continue
|
|
if key not in SUPPORTED_KEYWORDS:
|
|
problems.append(f"{path}/{key}")
|
|
continue
|
|
_walk(value, f"{path}/{key}", problems)
|
|
elif isinstance(node, list):
|
|
for index, value in enumerate(node):
|
|
_walk(value, f"{path}/{index}", problems)
|
|
|
|
|
|
def unsupported_keywords(schema: dict[str, Any]) -> list[str]:
|
|
"""Return JSON-pointer style paths of keywords the validator ignores."""
|
|
problems: list[str] = []
|
|
_walk(schema, "#", problems)
|
|
return problems
|
|
|
|
|
|
def _resolve_ref(ref: str, current: str, schemas: dict[str, dict[str, Any]]) -> tuple[dict[str, Any], str]:
|
|
file_part, _, pointer = ref.partition("#")
|
|
file_name = file_part or current
|
|
if file_name not in schemas:
|
|
raise ContractError(f"unknown schema reference {ref!r}")
|
|
node: Any = schemas[file_name]
|
|
for token in [t for t in pointer.split("/") if t]:
|
|
if not isinstance(node, dict) or token not in node:
|
|
raise ContractError(f"unresolvable pointer {ref!r}")
|
|
node = node[token]
|
|
return node, file_name
|
|
|
|
|
|
def _check_type(schema: dict[str, Any], value: Any, path: str, errors: list[str]) -> bool:
|
|
expected = schema.get("type")
|
|
if expected is None:
|
|
return True
|
|
if expected not in _TYPE_CHECKS:
|
|
raise ContractError(f"unsupported type {expected!r} at {path}")
|
|
if not _TYPE_CHECKS[expected](value):
|
|
errors.append(f"{path}: expected {expected}")
|
|
return False
|
|
return True
|
|
|
|
|
|
def _check_scalars(schema: dict[str, Any], value: Any, path: str, errors: list[str]) -> None:
|
|
if "const" in schema and value != schema["const"]:
|
|
errors.append(f"{path}: expected constant {schema['const']!r}")
|
|
if "enum" in schema and value not in schema["enum"]:
|
|
errors.append(f"{path}: {value!r} not in enum")
|
|
if isinstance(value, str):
|
|
if "minLength" in schema and len(value) < schema["minLength"]:
|
|
errors.append(f"{path}: shorter than {schema['minLength']}")
|
|
if "maxLength" in schema and len(value) > schema["maxLength"]:
|
|
errors.append(f"{path}: longer than {schema['maxLength']}")
|
|
if "pattern" in schema and not re.search(schema["pattern"], value):
|
|
errors.append(f"{path}: does not match {schema['pattern']!r}")
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
if "minimum" in schema and value < schema["minimum"]:
|
|
errors.append(f"{path}: below minimum {schema['minimum']}")
|
|
if "maximum" in schema and value > schema["maximum"]:
|
|
errors.append(f"{path}: above maximum {schema['maximum']}")
|
|
|
|
|
|
def _check_object(schema, value, path, errors, current, schemas) -> None:
|
|
properties = schema.get("properties", {})
|
|
for key in schema.get("required", []):
|
|
if key not in value:
|
|
errors.append(f"{path}: missing required {key!r}")
|
|
for key, item in value.items():
|
|
if key in properties:
|
|
_validate(properties[key], item, f"{path}/{key}", errors, current, schemas)
|
|
elif schema.get("additionalProperties") is False:
|
|
errors.append(f"{path}: unexpected property {key!r}")
|
|
|
|
|
|
def _check_array(schema, value, path, errors, current, schemas) -> None:
|
|
if "minItems" in schema and len(value) < schema["minItems"]:
|
|
errors.append(f"{path}: fewer than {schema['minItems']} items")
|
|
if "maxItems" in schema and len(value) > schema["maxItems"]:
|
|
errors.append(f"{path}: more than {schema['maxItems']} items")
|
|
if schema.get("uniqueItems"):
|
|
seen = [json.dumps(item, sort_keys=True) for item in value]
|
|
if len(set(seen)) != len(seen):
|
|
errors.append(f"{path}: items are not unique")
|
|
if "items" in schema:
|
|
for index, item in enumerate(value):
|
|
_validate(schema["items"], item, f"{path}/{index}", errors, current, schemas)
|
|
|
|
|
|
def _check_one_of(schema, value, path, errors, current, schemas) -> None:
|
|
attempts: list[list[str]] = []
|
|
for option in schema["oneOf"]:
|
|
sub: list[str] = []
|
|
_validate(option, value, path, sub, current, schemas)
|
|
attempts.append(sub)
|
|
matches = sum(not sub for sub in attempts)
|
|
if matches != 1:
|
|
errors.append(f"{path}: matched {matches} oneOf branches, expected exactly 1")
|
|
if matches == 0:
|
|
errors.extend(min(attempts, key=len))
|
|
|
|
|
|
def _validate(schema, value, path, errors, current, schemas) -> None:
|
|
if "$ref" in schema:
|
|
target, file_name = _resolve_ref(schema["$ref"], current, schemas)
|
|
_validate(target, value, path, errors, file_name, schemas)
|
|
return
|
|
if not _check_type(schema, value, path, errors):
|
|
return
|
|
_check_scalars(schema, value, path, errors)
|
|
if isinstance(value, dict):
|
|
_check_object(schema, value, path, errors, current, schemas)
|
|
if isinstance(value, list):
|
|
_check_array(schema, value, path, errors, current, schemas)
|
|
if "oneOf" in schema:
|
|
_check_one_of(schema, value, path, errors, current, schemas)
|
|
|
|
|
|
def validate(
|
|
schema_name: str,
|
|
value: Any,
|
|
schemas: dict[str, dict[str, Any]] | None = None,
|
|
pointer: str = "",
|
|
) -> list[str]:
|
|
"""Validate ``value`` against a schema file, or a ``#/$defs/...`` pointer inside it.
|
|
|
|
Returns a list of human-readable problems; an empty list means valid.
|
|
"""
|
|
schemas = schemas or load_all()
|
|
schema, file_name = _resolve_ref(f"{schema_name}#{pointer}", schema_name, schemas)
|
|
errors: list[str] = []
|
|
_validate(schema, value, "$", errors, file_name, schemas)
|
|
return errors
|
|
|
|
|
|
def record_schema_names(schemas: dict[str, dict[str, Any]] | None = None) -> dict[str, str]:
|
|
"""Map every ``schema`` constant (e.g. ``hux.event.v1``) to its file name."""
|
|
schemas = schemas or load_all()
|
|
found: dict[str, str] = {}
|
|
|
|
def visit(node: Any, file_name: str) -> None:
|
|
if isinstance(node, dict):
|
|
const = node.get("properties", {}).get("schema", {}).get("const")
|
|
if isinstance(const, str):
|
|
found[const] = file_name
|
|
for child in node.values():
|
|
visit(child, file_name)
|
|
elif isinstance(node, list):
|
|
for child in node:
|
|
visit(child, file_name)
|
|
|
|
for file_name, schema in schemas.items():
|
|
visit(schema, file_name)
|
|
return found
|
|
|
|
|
|
def validate_record(value: Any, schemas: dict[str, dict[str, Any]] | None = None) -> list[str]:
|
|
"""Validate a record by its own ``schema`` field."""
|
|
schemas = schemas or load_all()
|
|
if not isinstance(value, dict) or not isinstance(value.get("schema"), str):
|
|
return ["$: record has no string 'schema' field"]
|
|
file_name = record_schema_names(schemas).get(value["schema"])
|
|
if file_name is None:
|
|
return [f"$: unknown record schema {value['schema']!r}"]
|
|
return validate(file_name, value, schemas)
|