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)
     - Cross-entry supply-fact consistency: the same (supply item,
       source) across entries must carry the same ge_limit. A diverging
       limit (e.g. nature rune "12500" in one entry vs "18000" in
       another) is a silent fact error, not a per-entry choice — GE
       buy limits are item-global.
     - SII translatability: an action that casts Superheat Item must
       name a concrete recipe id declared in the entry's `recipes:`
       table (all inputs + quantities, smithing gate, bar output). The
       client translates a bar from its exact recipe, not from a vague
       "positive margin ore" rule.
  3. Global id uniqueness across the whole codex tree.

Malformed/scalar YAML documents (an empty doc, a bare string, a list)
are rejected with a normal validator ERROR — they never reach the
cross-entry gates and never crash the tool with a traceback.

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

Usage: python3 tools/validate.py [--stale-days 30] [--selftest]
      --selftest: run the regression fixtures only
"""
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 document_error(rel, data):
    """Shape check: an entry must parse to a YAML mapping.

    Returns an error string, or None for a well-formed mapping. The
    caller keeps malformed/scalar documents out of the shared row set
    so downstream gates never see None/scalars (regression: a scalar
    document used to crash `supply_limit_conflicts` with an
    AttributeError instead of reporting a validation error).
    """
    if isinstance(data, dict):
        return None
    if data is None:
        got = "an empty document"
    elif isinstance(data, (list, tuple)):
        got = f"a {type(data).__name__} document"
    else:
        got = f"{type(data).__name__} ({data!r})"
    return f"{rel}: document is not a YAML mapping (parsed as {got})"


def supply_limit_conflicts(rows):
    """Cross-entry consistency for duplicate supply facts.

    rows: iterable of (rel, data). The same (item, source) appearing in
    more than one entry must agree on ge_limit — GE buy limits are a
    property of the item, so a divergence is a silent fact error.
    Returns a list of error strings.
    """
    facts = {}
    for rel, data in rows:
        if not isinstance(data, dict):  # never crash on malformed rows
            continue
        eco = data.get("economy") or {}
        for sup in eco.get("supplies", []):
            lim = sup.get("ge_limit")
            if lim is None:
                continue
            key = (sup.get("item"), sup.get("source"))
            facts.setdefault(key, {}).setdefault(lim, set()).add(rel)
    errors = []
    for (item, source), by_lim in sorted(facts.items()):
        if len(by_lim) > 1:
            detail = "; ".join(
                f"limit {lim} in {', '.join(sorted(rs))}"
                for lim, rs in sorted(by_lim.items()))
            errors.append(
                f"conflicting ge_limit for supply '{item}' (source {source}): {detail}")
    return errors


def recipe_total_input_qty(recipe):
    """Total input count of a recipe — every input and quantity counts.

    A multi-input recipe (e.g. steel-bar = 1 iron ore + 2 coal) must
    account for all inputs, not just the headline ore.
    """
    return sum(int(inp.get("qty", 0)) for inp in recipe.get("inputs", []))


def recipe_margin(recipe, bar_price, input_prices, nature_price):
    """Complete-recipe margin for Superheat Item.

    bar GE price minus (ALL inputs at live GE buy prices + one nature
    rune). Negative/zero => the bar is not worth producing with SII.
    This is the exact formula the entries pin; tests use it with fixed
    prices so the fixture is deterministic.
    """
    cost = sum(
        int(inp.get("qty", 0)) * int(input_prices.get(inp.get("item"), 0))
        for inp in recipe.get("inputs", []))
    return bar_price - (cost + nature_price)


def recipe_errors(rel, data):
    """Entry-level SII translatability checks. Returns error strings."""
    errors = []
    recipes = data.get("recipes")
    if recipes is not None and not isinstance(recipes, dict):
        return [
            f"{rel}: 'recipes' must be a mapping of recipe id -> recipe definition"]
    for action in data.get("actions", []):
        params = action.get("params") or {}
        spell = str(params.get("spell", ""))
        if "superheat" not in spell.lower():
            continue
        rid = params.get("recipe")
        if not isinstance(rid, str) or not rid:
            errors.append(
                f"{rel}: action {action.get('id')} casts Superheat Item but has no "
                f"params.recipe naming a concrete recipes: entry (the client cannot "
                f"translate an ore choice into a bar without the full recipe)")
        elif recipes is None or rid not in recipes:
            errors.append(
                f"{rel}: action {action.get('id')} references recipe '{rid}' which "
                f"is not declared in recipes:")
    return errors


def run_selftest(verbose=False):
    """Regression fixtures for the cross-field gates; error strings."""
    errors = []

    def check(name, passed):
        if verbose:
            print(f"SELFTEST {'PASS' if passed else 'FAIL'}  {name}")
        if not passed:
            errors.append(f"selftest: {name}")

    # --- supply-consistency fixtures ---
    fixtures = [
        ("identical (item, source) with diverging ge_limit (18000 vs 12500)",
         [("e1.yaml", {"economy": {"supplies": [
              {"item": "nature rune", "source": "ge", "ge_limit": 18000}]}}),
          ("e2.yaml", {"economy": {"supplies": [
              {"item": "nature rune", "source": "ge", "ge_limit": 12500}]}})], True),
        ("identical (item, source) with agreeing ge_limit (18000 both)",
         [("e1.yaml", {"economy": {"supplies": [
              {"item": "nature rune", "source": "ge", "ge_limit": 18000}]}}),
          ("e2.yaml", {"economy": {"supplies": [
              {"item": "nature rune", "source": "ge", "ge_limit": 18000}]}})], False),
        ("no supplies / missing ge_limit are not gated",
         [("e1.yaml", {"economy": {}}),
          ("e2.yaml", {"economy": {"supplies": [
              {"item": "gold necklace", "source": "ge"}]}})], False),
    ]
    for name, rows, expect_error in fixtures:
        got = bool(supply_limit_conflicts(rows))
        check(name, got == expect_error)

    # --- malformed-document regression: no traceback, normal errors ---
    check("supply_limit_conflicts handles (bad.yaml, None) rows without crashing",
          supply_limit_conflicts([("bad.yaml", None)]) == [])
    for val in (None, "just a scalar", [1, 2, 3]):
        msg = document_error("bad.yaml", val)
        check(
            f"document_error rejects {type(val).__name__} docs with an ERROR, no raise",
            isinstance(msg, str) and "not a YAML mapping" in msg)
    check("document_error accepts mapping documents",
          document_error("ok.yaml", {"id": "x"}) is None)

    # --- deterministic multi-input recipe fixture: steel-bar (1 iron + 2 coal) ---
    steel = {"bar": "Steel bar", "bar_item_id": 2353, "smithing_level": 30,
             "inputs": [
                 {"item": "Iron ore", "item_id": 440, "qty": 1, "source": "ge"},
                 {"item": "Coal", "item_id": 453, "qty": 2, "source": "ge"}]}
    check("steel-bar recipe totals 3 inputs (1 iron ore + 2 coal)",
          recipe_total_input_qty(steel) == 3)
    # cost = 100*1 + 150*2 = 400; bar 600 - (400 + nature 145) = +55
    check("steel-bar complete-recipe margin = 55 (bar 600 - (iron 100 + 2*coal 150 + nature 145))",
          recipe_margin(steel, bar_price=600,
                        input_prices={"Iron ore": 100, "Coal": 150},
                        nature_price=145) == 55)
    check("steel-bar margin goes negative when the bar price no longer covers the full recipe",
          recipe_margin(steel, bar_price=500,
                        input_prices={"Iron ore": 100, "Coal": 150},
                        nature_price=145) < 0)

    # --- SII translatability fixtures ---
    entry_ok = {"actions": [
        {"id": "a4", "step": "cast-superheat", "verb": "use_item_on",
         "params": {"spell": "Superheat Item", "recipe": "steel-bar"},
         "on_fail": "skip"}],
        "recipes": {"steel-bar": steel}}
    entry_undeclared = {"actions": [
        {"id": "a4", "step": "cast-superheat", "verb": "use_item_on",
         "params": {"spell": "Superheat Item", "recipe": "steel-bar"},
         "on_fail": "skip"}]}
    entry_unpinned = {"actions": [
        {"id": "a4", "step": "cast-superheat", "verb": "use_item_on",
         "params": {"spell": "Superheat Item"}, "on_fail": "skip"}],
        "recipes": {"steel-bar": steel}}
    check("Superheat action naming a declared recipes: id passes",
          recipe_errors("ok.yaml", entry_ok) == [])
    undeclared = recipe_errors("bad.yaml", entry_undeclared)
    check("Superheat action naming an undeclared recipe id errors",
          any("not declared in recipes:" in e for e in undeclared))
    unpinned = recipe_errors("bad.yaml", entry_unpinned)
    check("Superheat action without params.recipe errors",
          any("no params.recipe" in e for e in unpinned))

    return errors


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

    errors, warnings = [], []
    if "--selftest" in sys.argv:
        errors += run_selftest(verbose=True)
        print(f"RESULT: selftest, {len(errors)} errors, 0 warnings")
        return 1 if errors else 0
    entries = load_entries()
    if not entries:
        print("ERROR: no codex entries found under codex/")
        return 1

    ids = {}
    today = datetime.date.today()
    rows = []

    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
        shape_err = document_error(rel, data)
        if shape_err:
            errors.append(shape_err)
            continue
        rows.append((rel, data))

        # 1. schema
        v = jsonschema.Draft7Validator(SCHEMA)
        schema_errors = sorted(v.iter_errors(data), key=lambda e: list(e.path))
        for err in schema_errors:
            loc = "/".join(str(p) for p in err.path) or "(root)"
            errors.append(f"{rel} [{loc}]: {err.message}")
        if schema_errors:
            # cross-checks on a schema-invalid 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")

        # 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")

        # SII translatability (concrete recipes: entry per Superheat action)
        errors += recipe_errors(rel, data)

        # 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")

    # 2b. cross-entry supply-fact consistency (GE buy limits are item-global)
    errors += supply_limit_conflicts(rows)

    # 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())