erp/fleet/profile/scripts/validate.py

#!/usr/bin/env python3
"""Validate the fleet/profile document surface (erp#54).

Checks, in order:
  1. fiscal.yaml   parses (strict YAML subset) and validates against fiscal.schema.json
  2. calendar.yaml parses and validates against calendar.schema.json
  3. Referential integrity:
       - every fiscal rule's `decision: adc-NNN` resolves to exactly one
         decisions/adc-NNN-*.md whose frontmatter status is Accepted
         (agents draft, the operator Accepts - a rule may only rely on an
         Accepted decision)
       - rule/entry ids unique; effective_from <= effective_until
       - calendar entries: exactly one of due|recurrence, unless status is
         pending-definition/conditional; recurrence months are 1..12
  4. ADC register hygiene: filename <-> frontmatter id match, valid status
     (Proposed | Accepted | Superseded-by-NNN), Accepted records carry
     decided + effective_from, Proposed records carry decided: null,
     required body sections present, no duplicate adc numbers.

Stdlib only, by design (PRD agent-catalog "document surface" + erp#54): the
profile YAML is written in a strict subset - block maps, block sequences,
single-line scalars, one-line [flow] lists of scalars, comments; no anchors,
no multi-line scalars, no nested flow - and the schemas use the JSON-Schema
subset {type, properties, required, additionalProperties, items, enum,
pattern}.

Usage:  python3 fleet/profile/scripts/validate.py
Exit 0 = all green (warnings allowed). Exit 1 = at least one error.
"""

import json
import re
import sys
from pathlib import Path

PROFILE = Path(__file__).resolve().parent.parent
DECISIONS = PROFILE / "decisions"

DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
ADC_FILE_RE = re.compile(r"^adc-(\d{3})-[a-z0-9-]+\.md$")
ADC_STATUS_RE = re.compile(r"^(Proposed|Accepted|Superseded-by-\d{3})$")
ADC_REQUIRED_SECTIONS = [
    "## Context",
    "## Decision",
    "## Base légale & doctrine",
    "## Alternatives rejected",
    "## Consequences",
    "## QA & validation",
    "## References",
]


# --------------------------------------------------------------------------
# Strict YAML-subset parser
# --------------------------------------------------------------------------

class YamlSubsetError(Exception):
    def __init__(self, msg, line=None):
        super().__init__(f"line {line}: {msg}" if line else msg)


def _strip_comment(raw, line):
    """Return the value token of a raw value string, comment stripped."""
    s = raw.strip()
    if not s or s.startswith("#"):
        return ""
    if s[0] in "\"'":
        q = s[0]
        end = s.find(q, 1)
        if end == -1:
            raise YamlSubsetError("unterminated quoted scalar", line)
        rest = s[end + 1:].strip()
        if rest and not rest.startswith("#"):
            raise YamlSubsetError(f"trailing content after quoted scalar: {rest!r}", line)
        return s[: end + 1]
    idx = s.find(" #")
    if idx != -1:
        s = s[:idx]
    return s.strip()


def _scalar(tok, line):
    if tok == "" or tok in ("null", "~"):
        return None
    if tok == "true":
        return True
    if tok == "false":
        return False
    if len(tok) >= 2 and tok[0] in "\"'" and tok[-1] == tok[0]:
        return tok[1:-1]
    if tok.startswith("["):
        if not tok.endswith("]"):
            raise YamlSubsetError("unterminated flow list", line)
        inner = tok[1:-1].strip()
        if not inner:
            return []
        return [_scalar(p.strip(), line) for p in inner.split(",")]
    if re.fullmatch(r"-?\d+", tok):
        return int(tok)
    if re.fullmatch(r"-?\d+\.\d+", tok):
        return float(tok)
    return tok  # plain string (ISO dates stay strings)


def parse_yaml(text, name="<yaml>"):
    items = []
    for n, raw in enumerate(text.splitlines(), 1):
        if not raw.strip() or raw.strip().startswith("#"):
            continue
        if "\t" in raw:
            raise YamlSubsetError(f"{name}: tab character (use spaces)", n)
        indent = len(raw) - len(raw.lstrip(" "))
        items.append((indent, raw.strip(), n))
    if not items:
        return {}
    value, nxt = _parse_block(items, 0, items[0][0])
    if nxt != len(items):
        raise YamlSubsetError(f"{name}: trailing content", items[nxt][2])
    return value


def _parse_block(items, i, indent):
    _, content, line = items[i]
    if items[i][0] != indent:
        raise YamlSubsetError("unexpected indent", line)
    if content == "-" or content.startswith("- "):
        return _parse_seq(items, i, indent)
    return _parse_map(items, i, indent)


def _parse_map(items, i, indent):
    result = {}
    while i < len(items):
        ind, content, line = items[i]
        if ind < indent:
            break
        if ind > indent:
            raise YamlSubsetError("unexpected deeper indent", line)
        if content == "-" or content.startswith("- "):
            raise YamlSubsetError("sequence item at mapping level", line)
        m = re.match(r"^([A-Za-z_][A-Za-z0-9_-]*):(?:\s+(.*))?$", content)
        if not m:
            raise YamlSubsetError(f"not a 'key: value' line: {content!r}", line)
        key, rest = m.group(1), m.group(2)
        if key in result:
            raise YamlSubsetError(f"duplicate key {key!r}", line)
        tok = _strip_comment(rest, line) if rest else ""
        if tok:
            result[key] = _scalar(tok, line)
            i += 1
        else:
            i += 1
            if i < len(items) and items[i][0] > indent:
                value, i = _parse_block(items, i, items[i][0])
                result[key] = value
            else:
                result[key] = None
    return result, i


def _parse_seq(items, i, indent):
    result = []
    while i < len(items):
        ind, content, line = items[i]
        if ind < indent:
            break
        if ind > indent:
            raise YamlSubsetError("unexpected deeper indent in sequence", line)
        if not (content == "-" or content.startswith("- ")):
            break
        rest = content[1:].strip()
        if not rest or rest.startswith("#"):
            i += 1
            if i < len(items) and items[i][0] > indent:
                value, i = _parse_block(items, i, items[i][0])
                result.append(value)
            else:
                result.append(None)
        elif re.match(r"^[A-Za-z_][A-Za-z0-9_-]*:(\s|$)", rest):
            # inline first key of a mapping item: reparse at indent+2
            sub = [(indent + 2, rest, line)]
            i += 1
            while i < len(items) and items[i][0] > indent:
                s_ind, s_content, s_line = items[i]
                if s_ind < indent + 2:
                    raise YamlSubsetError("bad indent inside sequence item", s_line)
                sub.append((s_ind, s_content, s_line))
                i += 1
            value, used = _parse_map(sub, 0, indent + 2)
            if used != len(sub):
                raise YamlSubsetError("trailing content in sequence item", sub[used][2])
            result.append(value)
        else:
            tok = _strip_comment(rest, line)
            result.append(_scalar(tok, line))
            i += 1
    return result, i


# --------------------------------------------------------------------------
# JSON-Schema subset checker
# --------------------------------------------------------------------------

def _is_type(v, t):
    return {
        "object": lambda: isinstance(v, dict),
        "array": lambda: isinstance(v, list),
        "string": lambda: isinstance(v, str),
        "integer": lambda: isinstance(v, int) and not isinstance(v, bool),
        "number": lambda: isinstance(v, (int, float)) and not isinstance(v, bool),
        "boolean": lambda: isinstance(v, bool),
        "null": lambda: v is None,
    }[t]()


def schema_check(value, schema, path, errors):
    types = schema.get("type")
    if types:
        if isinstance(types, str):
            types = [types]
        if not any(_is_type(value, t) for t in types):
            errors.append(f"{path}: expected {'/'.join(types)}, got {type(value).__name__}")
            return
    if value is None:
        return
    if "enum" in schema and value not in schema["enum"]:
        errors.append(f"{path}: {value!r} not in {schema['enum']}")
    if isinstance(value, str) and "pattern" in schema:
        if not re.search(schema["pattern"], value):
            errors.append(f"{path}: {value!r} does not match {schema['pattern']!r}")
    if isinstance(value, dict):
        props = schema.get("properties", {})
        for req in schema.get("required", []):
            if req not in value:
                errors.append(f"{path}: missing required key '{req}'")
        if schema.get("additionalProperties") is False:
            for k in value:
                if k not in props:
                    errors.append(f"{path}: unexpected key '{k}'")
        for k, v in value.items():
            if k in props:
                schema_check(v, props[k], f"{path}.{k}", errors)
    if isinstance(value, list) and "items" in schema:
        for idx, item in enumerate(value):
            schema_check(item, schema["items"], f"{path}[{idx}]", errors)


# --------------------------------------------------------------------------
# ADC register
# --------------------------------------------------------------------------

def parse_frontmatter(path):
    text = path.read_text(encoding="utf-8")
    if not text.startswith("---\n"):
        raise YamlSubsetError(f"{path.name}: missing frontmatter")
    end = text.find("\n---", 4)
    if end == -1:
        raise YamlSubsetError(f"{path.name}: unterminated frontmatter")
    return parse_yaml(text[4:end], path.name), text[end + 4:]


def load_adc_register(errors):
    """Return {'adc-NNN': {'file': name, 'status': str, ...}}."""
    register = {}
    if not DECISIONS.is_dir():
        errors.append(f"decisions/ directory missing at {DECISIONS}")
        return register
    for path in sorted(DECISIONS.glob("*.md")):
        if path.name == "adc-template.md":
            continue
        m = ADC_FILE_RE.match(path.name)
        if not m:
            errors.append(f"decisions/{path.name}: name must match adc-NNN-<slug>.md")
            continue
        adc_id = f"adc-{m.group(1)}"
        if adc_id in register:
            errors.append(f"decisions/{path.name}: duplicate id {adc_id} "
                          f"(also {register[adc_id]['file']})")
            continue
        try:
            fm, body = parse_frontmatter(path)
        except YamlSubsetError as e:
            errors.append(f"decisions/{path.name}: {e}")
            continue
        if fm.get("id") != adc_id:
            errors.append(f"decisions/{path.name}: frontmatter id {fm.get('id')!r} != {adc_id}")
        status = fm.get("status")
        if not (isinstance(status, str) and ADC_STATUS_RE.match(status)):
            errors.append(f"decisions/{path.name}: invalid status {status!r} "
                          "(Proposed | Accepted | Superseded-by-NNN)")
        if status == "Accepted":
            if not (isinstance(fm.get("decided"), str) and DATE_RE.match(fm["decided"])):
                errors.append(f"decisions/{path.name}: Accepted requires a 'decided' date")
            if not (isinstance(fm.get("effective_from"), str) and DATE_RE.match(fm["effective_from"])):
                errors.append(f"decisions/{path.name}: Accepted requires 'effective_from'")
        elif status == "Proposed":
            if fm.get("decided") is not None:
                errors.append(f"decisions/{path.name}: Proposed must keep decided: null "
                              "(acceptance is a human act)")
        for section in ADC_REQUIRED_SECTIONS:
            if section not in body:
                errors.append(f"decisions/{path.name}: missing section '{section}'")
        register[adc_id] = dict(fm, file=path.name)
    return register


# --------------------------------------------------------------------------
# Custom (referential) checks
# --------------------------------------------------------------------------

def check_dates_ordered(obj, path, errors):
    frm, until = obj.get("effective_from"), obj.get("effective_until")
    if isinstance(frm, str) and isinstance(until, str) and frm > until:
        errors.append(f"{path}: effective_from {frm} > effective_until {until}")


def check_fiscal(fiscal, register, errors, warnings, resolution):
    seen = set()
    for idx, rule in enumerate(fiscal.get("rules") or []):
        rid = rule.get("id", f"[{idx}]")
        path = f"fiscal.rules.{rid}"
        if rid in seen:
            errors.append(f"{path}: duplicate rule id")
        seen.add(rid)
        check_dates_ordered(rule, path, errors)
        decision = rule.get("decision")
        adc = register.get(decision)
        if adc is None:
            errors.append(f"{path}: decision {decision!r} does not resolve to any "
                          f"decisions/adc-NNN-*.md record")
            resolution.append((rid, decision, "<missing>", "<missing>"))
            continue
        resolution.append((rid, decision, adc["file"], adc.get("status")))
        if adc.get("status") != "Accepted":
            errors.append(f"{path}: decision {decision} has status "
                          f"{adc.get('status')!r} - a rule in force may only cite an "
                          "Accepted decision (agents draft, the operator Accepts)")


def check_calendar(calendar, errors, warnings):
    seen = set()
    for idx, entry in enumerate(calendar.get("entries") or []):
        eid = entry.get("id", f"[{idx}]")
        path = f"calendar.entries.{eid}"
        if eid in seen:
            errors.append(f"{path}: duplicate entry id")
        seen.add(eid)
        check_dates_ordered(entry, path, errors)
        status = entry.get("status", "confirmed")
        has_due = entry.get("due") is not None
        has_rec = entry.get("recurrence") is not None
        if has_due and has_rec:
            errors.append(f"{path}: carries both due and recurrence")
        elif not has_due and not has_rec and status not in ("pending-definition", "conditional"):
            errors.append(f"{path}: needs due or recurrence (status {status!r})")
        rec = entry.get("recurrence")
        months = entry.get("months")
        if rec in ("yearly", "quarterly"):
            if not months:
                errors.append(f"{path}: recurrence {rec!r} requires months")
        elif months is not None:
            errors.append(f"{path}: months only allowed with yearly/quarterly recurrence")
        for mth in months or []:
            if not (isinstance(mth, int) and 1 <= mth <= 12):
                errors.append(f"{path}: month {mth!r} not in 1..12")
        if has_due and re.fullmatch(r"\d{4}-\d{2}", entry["due"]):
            warnings.append(f"{path}: month-precision due ({entry['due']}) - the source "
                            "gives no day; verify on the authority's notice")
        if status in ("estimated", "conditional", "pending-definition"):
            warnings.append(f"{path}: status {status} - operator verification pending")


# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------

def load_and_validate(name, errors):
    yaml_path = PROFILE / f"{name}.yaml"
    schema_path = PROFILE / f"{name}.schema.json"
    try:
        data = parse_yaml(yaml_path.read_text(encoding="utf-8"), yaml_path.name)
    except (OSError, YamlSubsetError) as e:
        errors.append(f"{yaml_path.name}: {e}")
        return None
    try:
        schema = json.loads(schema_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as e:
        errors.append(f"{schema_path.name}: {e}")
        return None
    before = len(errors)
    schema_check(data, schema, name, errors)
    print(f"== {yaml_path.name} ==")
    print(f"  parse: OK")
    print(f"  schema ({schema_path.name}): {'OK' if len(errors) == before else 'FAIL'}")
    return data


def main():
    errors, warnings, resolution = [], [], []

    fiscal = load_and_validate("fiscal", errors)
    calendar = load_and_validate("calendar", errors)

    print("== decisions/ (ADC register) ==")
    register = load_adc_register(errors)
    for adc_id in sorted(register):
        adc = register[adc_id]
        print(f"  {adc['file']:<44} [{adc.get('status')}]")

    if fiscal is not None:
        check_fiscal(fiscal, register, errors, warnings, resolution)
        print("== rule -> ADC resolution ==")
        for rid, decision, fname, status in resolution:
            print(f"  {rid:<32} -> {decision} -> {fname} [{status}]")
    if calendar is not None:
        check_calendar(calendar, errors, warnings)

    if warnings:
        print(f"== warnings ({len(warnings)}) ==")
        for w in warnings:
            print(f"  WARN {w}")
    if errors:
        print(f"== errors ({len(errors)}) ==")
        for e in errors:
            print(f"  ERROR {e}")
    n_rules = len((fiscal or {}).get("rules") or [])
    n_entries = len((calendar or {}).get("entries") or [])
    verdict = "FAIL" if errors else "PASS"
    print(f"== RESULT: {verdict} - {n_rules} rules, {n_entries} calendar entries, "
          f"{len(register)} ADC records, {len(errors)} errors, {len(warnings)} warnings ==")
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main())