src/ail_ref/validator.py

Read-only source from the unchanged 0.3.4 kit. Existing license terms apply.

from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal

from .ast import Alias, Atom, Call, Document, ListValue, Number, Record, Value
from .registry import concept_ids, load_registry


@dataclass(frozen=True)
class ValidationIssue:
    code: str
    message: str
    path: str
    severity: str = "error"


ALLOWED_ACTS = {f"A{i:02d}" for i in range(1, 11)}
REQUIRED_FIELDS = {"id", "act", "registry", "body"}
REF_ACTS = {"A05", "A06", "A09", "A10"}
ARITIES: dict[str, tuple[int, int | None]] = {
    "P": (3, 3),
    "AND": (1, None),
    "OR": (1, None),
    "NOT": (1, 1),
    "IF": (2, 2),
    "BIND": (2, 2),
    "VAL": (1, 2),
    "TIME": (1, 1),
    "DUR": (1, 1),
    "ADD": (2, 2),
    "DO": (1, 1),
}


def _walk(value: Value, path: str):
    yield value, path
    if isinstance(value, Alias):
        yield value.target, path + ".target"
    elif isinstance(value, Call):
        for index, arg in enumerate(value.args):
            yield from _walk(arg, f"{path}.args[{index}]")
    elif isinstance(value, ListValue):
        for index, item in enumerate(value.items):
            yield from _walk(item, f"{path}[{index}]")
    elif isinstance(value, Record):
        for key, item in value.fields:
            yield from _walk(item, f"{path}.{key}")


def _validate_message(record: Record, path: str) -> list[ValidationIssue]:
    issues: list[ValidationIssue] = []
    fields = dict(record.fields)
    for key in sorted(REQUIRED_FIELDS - fields.keys()):
        issues.append(ValidationIssue("E02", f"missing required field {key}", path))

    act = fields.get("act")
    if act is not None:
        if not isinstance(act, Atom) or act.value not in ALLOWED_ACTS:
            issues.append(ValidationIssue("E06", "act must be one of A01 through A10", path + ".act"))
        elif act.value in REF_ACTS and "ref" not in fields:
            issues.append(ValidationIssue("E02", f"{act.value} requires ref", path))

    conf = fields.get("conf")
    if conf is not None:
        if not isinstance(conf, Number):
            issues.append(ValidationIssue("E06", "conf must be a number from 0.0 through 1.0", path + ".conf"))
        elif not (Decimal("0") <= conf.value <= Decimal("1")):
            issues.append(ValidationIssue("E06", "conf is outside 0.0 through 1.0", path + ".conf"))
    return issues


def validate(document: Document, strict_registry: bool = False, registry: dict | None = None) -> list[ValidationIssue]:
    issues: list[ValidationIssue] = []
    known = concept_ids(registry or load_registry()) if strict_registry else set()

    for unit_index, unit in enumerate(document.units):
        path = f"units[{unit_index}]"
        expected_header = "AIL-PASSPORT/0.2" if isinstance(unit.value, Record) and unit.value.name == "PASSPORT" else "AIL/0.2"
        if unit.header != expected_header:
            issues.append(ValidationIssue("E04", f"expected header {expected_header}, received {unit.header}", path))

        if isinstance(unit.value, Record) and unit.value.name == "M":
            issues.extend(_validate_message(unit.value, path + ".M"))

        for value, value_path in _walk(unit.value, path):
            if isinstance(value, Call) and value.name in ARITIES:
                minimum, maximum = ARITIES[value.name]
                count = len(value.args)
                if count < minimum or (maximum is not None and count > maximum):
                    expected = str(minimum) if maximum == minimum else f"{minimum}+"
                    issues.append(ValidationIssue("E05", f"{value.name} expects {expected} arguments, received {count}", value_path))
            if strict_registry and isinstance(value, Atom) and value.value.startswith("ail:") and value.value not in known:
                issues.append(ValidationIssue("E03", f"unknown ail core concept {value.value}", value_path))
    return issues