osrs-codex/tools/validate.py

#!/usr/bin/env python3
"""OSRS Codex validator — the gate for every codex/**.yaml entry.

Runs:
  1. JSON-Schema validation (schemas/entry.schema.json)
  2. Cross-field consistency checks the schema can't express:
     - state_machine `next`/transition targets must be declared states
     - `initial` must be a declared state
     - duplicate action ids / step names
     - status:verified entries with `verified:` older than --stale-days
       => WARNING (entry should be re-verified, not dropped)
     - `best_practice` lines whose source starts with "researched"
       => WARNING (citation TODO — file a follow-up issue)
  3. Global id uniqueness across the whole codex tree.

Exit codes: 0 = clean (warnings ok), 1 = validation errors.

Usage: python3 tools/validate.py [--stale-days 30]
"""
import datetime
import json
import re
import sys
from pathlib import Path

import jsonschema
import yaml

ROOT = Path(__file__).resolve().parent.parent
SCHEMA = json.loads((ROOT / "schemas" / "entry.schema.json").read_text())
CODEX = ROOT / "codex"
STALE_DAYS = 30


def load_entries():
    out = []
    for p in CODEX.rglob("*.yaml"):
        if p.name.startswith("_"):  # dataset files (e.g. _f2p-alch-catalog.yaml)
            continue
        out.append(p)
    return sorted(out)


def alch_worked_example_regression(rel, data, raw_text):
    """One-cast model worked-example regression (owner guidance 2026-08-28,
    PR #8 repair): entries with the owner-selected one-cast iteration model
    must pin a worked example with MONOTONIC, INDEPENDENT counters — breaks
    every 550 casts (1,100, 1,650, ...) regardless of restocks, and restocks
    every 901 casts (1,802, ...), so restocking never resets or shifts the
    break counter. The old 'restart the interval at the restock' bug (break
    #2 at cast 1,452 = 901 + restarted 549) must never reappear. Scoped to
    entries that declare the one-cast model AND carry a worked example, so
    unrelated entries are untouched."""
    ruco = data.get("run_control") or {}
    if "one cast" not in str(ruco.get("iteration", "")):
        return []
    if "worked example" not in raw_text.lower():
        return []
    errs = []
    if "1,452" in raw_text:
        errs.append(f"{rel}: worked example resets the break interval at a restock "
                    f"(cast 1,452 = 901 + restarted interval) — breaks must stay "
                    f"monotonic at 550/1,100/1,650 independently of restocks")
    for marker, label in (("550", "break #1 (cast 550)"),
                          ("901", "restock #1 (cast 901)"),
                          ("1,100", "break #2 (cast 1,100)"),
                          ("1,650", "break #3 (cast 1,650)"),
                          ("1,802", "restock #2 (cast 1,802)")):
        if marker not in raw_text:
            errs.append(f"{rel}: worked example missing monotonic counter pin "
                        f"{label} — breaks/restocks must be pinned literally")
    return errs


def main():
    global STALE_DAYS
    if "--stale-days" in sys.argv:
        STALE_DAYS = int(sys.argv[sys.argv.index("--stale-days") + 1])

    errors, warnings = [], []
    entries = load_entries()
    if not entries:
        print("ERROR: no codex entries found under codex/")
        return 1

    ids = {}
    today = datetime.date.today()

    for path in entries:
        rel = str(path.relative_to(ROOT))
        try:
            data = yaml.safe_load(path.read_text())
        except yaml.YAMLError as e:
            errors.append(f"{rel}: YAML parse error: {e}")
            continue

        # 1. schema
        v = jsonschema.Draft7Validator(SCHEMA)
        for err in sorted(v.iter_errors(data), key=lambda e: list(e.path)):
            loc = "/".join(str(p) for p in err.path) or "(root)"
            errors.append(f"{rel} [{loc}]: {err.message}")
        if errors and errors[-1].startswith(rel):
            # schema already broke; cross-checks on a broken doc are noise
            continue

        eid = data.get("id", rel)
        if eid in ids:
            errors.append(f"{rel}: duplicate entry id '{eid}' (also in {ids[eid]})")
        ids[eid] = rel

        # 2. state machine consistency
        sm = data.get("state_machine", {})
        states = sm.get("states", {})
        declared = set(states)
        terminals = set(sm.get("terminal", []))
        for sname, sdef in states.items():
            if sname not in terminals and not sdef.get("next"):
                errors.append(f"{rel}: non-terminal state '{sname}' has no 'next'")
            for nxt in sdef.get("next", []):
                if nxt not in declared:
                    errors.append(f"{rel}: state '{sname}' next->'{nxt}' not declared")
        if sm.get("initial") not in declared:
            errors.append(f"{rel}: initial state '{sm.get('initial')}' not declared")
        for t in sm.get("transitions", []):
            if t["from"] not in declared:
                errors.append(f"{rel}: transition from '{t['from']}' not declared")
            if t["to"] not in declared:
                errors.append(f"{rel}: transition to '{t['to']}' not declared")
        for term in sm.get("terminal", []):
            if term not in declared:
                errors.append(f"{rel}: terminal '{term}' not declared")

        # one-cast worked-example regression (owner guidance 2026-08-28):
        # monotonic break/restock counters, scoped to the one-cast model
        errors += alch_worked_example_regression(rel, data, path.read_text())

        # action id/step uniqueness
        a_ids = [a["id"] for a in data["actions"]]
        a_steps = [a["step"] for a in data["actions"]]
        if len(a_ids) != len(set(a_ids)):
            errors.append(f"{rel}: duplicate action ids")
        if len(a_steps) != len(set(a_steps)):
            errors.append(f"{rel}: duplicate action step names")

        # 3. staleness / research TODOs
        if data.get("status") == "verified":
            try:
                vdate = datetime.date.fromisoformat(data["verified"])
                age = (today - vdate).days
                if age > STALE_DAYS:
                    warnings.append(
                        f"{rel}: status=verified but facts are {age}d old "
                        f"(> {STALE_DAYS}d) — re-verify or drop to status:stale")
            except ValueError:
                pass
        for line in data.get("best_practice", {}).get("equipment", []) + \
                    data.get("best_practice", {}).get("techniques", []):
            if str(line.get("source", "")).startswith("researched"):
                warnings.append(
                    f"{rel}: best_practice '{line.get('item') or line.get('name')}' "
                    f"carries a 'researched' source — citation TODO, file follow-up")

    # report
    for w in warnings:
        print(f"WARN  {w}")
    for e in errors:
        print(f"ERROR {e}")
    print(f"RESULT: {len(entries)} entries, {len(errors)} errors, {len(warnings)} warnings")
    return 1 if errors else 0


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