runtime_contract.py

"""Dependency-free boundaries for untrusted model inputs and output."""
import json
import math
MAX_JSON_BYTES = 64_000


def strict_json(text):
    if not isinstance(text, str) or len(text.encode('utf-8')) > MAX_JSON_BYTES:
        raise ValueError('JSON must be text of at most 64,000 bytes')
    def pairs(items):
        result = {}
        for key, value in items:
            if key in result:
                raise ValueError('Duplicate JSON key')
            result[key] = value
        return result
    def constant(value):
        raise ValueError('Non-finite JSON number')
    try:
        result = json.loads(text, object_pairs_hook=pairs, parse_constant=constant)
        pending = [(result, 0)]
        while pending:
            value, depth = pending.pop()
            if depth > 32:
                raise ValueError('JSON nesting exceeds 32 levels')
            if isinstance(value, float) and not math.isfinite(value):
                raise ValueError('Non-finite JSON number')
            if isinstance(value, dict):
                for key in value:
                    key.encode('utf-8')
                pending.extend((v, depth + 1) for v in value.values())
            elif isinstance(value, list):
                pending.extend((v, depth + 1) for v in value)
            elif isinstance(value, str):
                value.encode('utf-8')
        return result
    except (RecursionError, UnicodeError) as exc:
        raise ValueError('Invalid or excessively nested JSON') from exc


def validate_input(quote, schema):
    if not isinstance(quote, str) or not quote.strip() or len(quote) > 4000:
        raise ValueError('Quote must contain 1–4,000 characters')
    if not isinstance(schema, dict) or not 1 <= len(schema) <= 10:
        raise ValueError('Schema must contain 1–10 splits')
    for split, columns in schema.items():
        if not isinstance(split, str) or not split.strip() or len(split) > 128:
            raise ValueError('Split names must contain 1–128 characters')
        if not isinstance(columns, list) or not 1 <= len(columns) <= 100:
            raise ValueError('Each split needs 1–100 columns')
        if any(not isinstance(c, str) or not c.strip() or len(c) > 128 for c in columns):
            raise ValueError('Column names must contain 1–128 characters')
        if len(columns) != len(set(columns)):
            raise ValueError('Duplicate schema columns')


def parse_proposal(raw, quote, schema):
    from data_v2 import grounded
    from training_data import reviewed_candidate
    validate_input(quote, schema)
    target = strict_json(raw)
    grounded(target, schema)
    return reviewed_candidate(quote, quote, target)