core/demos/claude_hybrid_verification/schema.py

"""Typed validation of the MCP-shaped tool payload — the System 1/System 2 boundary.

The proposer's entire contribution arrives as the ``arguments`` object of an
MCP-shaped tool call.  This module validates it against ``tool_schema.json``
**interpretively**: the constraint set (required keys, types, lengths, enum,
pattern, ``additionalProperties: false``) is read from the committed schema file
itself, so the executable boundary cannot silently drift from the documented one.

Fail-closed and dependency-free: only the narrow JSON-Schema subset the tool
schema actually uses is interpreted, and a schema feature outside that subset
raises loudly at load time rather than being silently ignored.  Validation never
executes any derivation — an invalid payload is rejected before CORE runs.
"""

from __future__ import annotations

import json
import re
from functools import lru_cache
from pathlib import Path
from typing import Any, Final

_HERE: Final[Path] = Path(__file__).resolve().parent
TOOL_SCHEMA_PATH: Final[Path] = _HERE / "tool_schema.json"

#: The JSON-Schema keywords this interpreter understands, per property spec.
#: ``description``/``default`` are documentation-only.  Anything else in the
#: committed schema is a contract this validator does not enforce -> loud error.
_SUPPORTED_PROPERTY_KEYWORDS: Final[frozenset[str]] = frozenset(
    {"type", "minLength", "maxLength", "enum", "pattern", "description", "default"}
)
_SUPPORTED_TYPES: Final[dict[str, type]] = {"string": str, "boolean": bool}


#: The top-level inputSchema keywords this interpreter understands.  A top-level
#: combinator (``allOf``, ``patternProperties``, ...) would be silently
#: unenforced, so its presence is a loud error, same as an unknown property
#: keyword.
_SUPPORTED_SCHEMA_KEYWORDS: Final[frozenset[str]] = frozenset(
    {"type", "properties", "required", "additionalProperties"}
)


@lru_cache(maxsize=1)
def load_tool_schema() -> dict[str, Any]:
    """The committed MCP-shaped tool definition.  Parsed once and cached: treat
    the returned mapping as read-only (callers only read or compare it)."""
    return json.loads(TOOL_SCHEMA_PATH.read_text(encoding="utf-8"))


@lru_cache(maxsize=1)
def _input_schema() -> dict[str, Any]:
    schema = load_tool_schema()["inputSchema"]
    if schema.get("type") != "object" or schema.get("additionalProperties") is not False:
        raise ValueError(
            "tool inputSchema must be a closed object "
            "(type: object, additionalProperties: false) — the no-smuggling boundary"
        )
    unsupported_top = set(schema) - _SUPPORTED_SCHEMA_KEYWORDS
    if unsupported_top:
        raise ValueError(
            f"inputSchema uses unsupported top-level keywords {sorted(unsupported_top)}; "
            "extend the validator before extending the schema"
        )
    for name, spec in schema["properties"].items():
        unsupported = set(spec) - _SUPPORTED_PROPERTY_KEYWORDS
        if unsupported:
            raise ValueError(
                f"inputSchema property {name!r} uses unsupported keywords "
                f"{sorted(unsupported)}; extend the validator before extending the schema"
            )
        if spec.get("type") not in _SUPPORTED_TYPES:
            raise ValueError(f"inputSchema property {name!r} has unsupported type")
    return schema


def _clip(name: object) -> str:
    """Bounded repr for echoing an unknown property name — untrusted caller text
    must not round-trip unbounded into surfaces or artifacts."""
    rendered = repr(name)
    return rendered if len(rendered) <= 80 else rendered[:79] + "…"


def validate_payload(arguments: Any) -> tuple[str, ...]:
    """Every way ``arguments`` violates the tool's input schema — empty iff valid.

    Deterministic order (sorted unknown keys first, then schema-declaration
    order), so the same bad payload always yields the same error surface.
    """
    schema = _input_schema()
    if not isinstance(arguments, dict):
        return (f"arguments must be a JSON object, got {type(arguments).__name__}",)

    errors: list[str] = []
    properties: dict[str, Any] = schema["properties"]

    for unknown in sorted(str(key) for key in set(arguments) - set(properties)):
        errors.append(
            f"unexpected property {_clip(unknown)} (additionalProperties is false; "
            "the proposer cannot smuggle answers, derivations, or directives)"
        )

    for required in schema["required"]:
        if required not in arguments:
            errors.append(f"missing required property {required!r}")

    for name, spec in properties.items():
        if name not in arguments:
            continue
        value = arguments[name]
        expected = _SUPPORTED_TYPES[spec["type"]]
        # bool is an int subclass; an exact-type check keeps booleans out of
        # string seats and vice versa.
        if type(value) is not expected:
            errors.append(f"property {name!r} must be {spec['type']}")
            continue
        if isinstance(value, str):
            if "minLength" in spec and len(value) < spec["minLength"]:
                errors.append(f"property {name!r} shorter than {spec['minLength']}")
            if "maxLength" in spec and len(value) > spec["maxLength"]:
                errors.append(f"property {name!r} longer than {spec['maxLength']}")
            if "enum" in spec and value not in spec["enum"]:
                errors.append(f"property {name!r} must be one of {spec['enum']}")
            if "pattern" in spec and re.fullmatch(spec["pattern"], value) is None:
                errors.append(f"property {name!r} does not match {spec['pattern']!r}")

    return tuple(errors)


__all__ = ["TOOL_SCHEMA_PATH", "load_tool_schema", "validate_payload"]