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 _loss_reserve_stop(initial_capital, total_assets, reserve):
    """Loss-reserve stop comparator in the pinned direction (owner guidance
    2026-08-29, governor review 10361, PR #8): stop fires only when the
    loss EXCEEDS the reserve, i.e. when (initial capital - total assets)
    > reserve. A loss makes the LHS positive; a profit (total assets >
    initial capital) makes it negative and can never fire this stop. The
    reversed 'total assets - initial capital' form is invalid because it
    is negative exactly when a loss exists."""
    return (initial_capital - total_assets) > reserve


def alch_accounting_regression(rel, data, raw_text):
    """Loss-reserve + fill-price accounting regression (owner guidance
    2026-08-29 for governor review 10361, PR #8): the low-alch-members
    entry must express the loss-reserve stop in the correct direction
    EVERYWHERE ('initial capital - total assets > 162,990'), the reversed
    sign form must never reappear, and the realized GP/hour bounds must be
    derived from the declared fill-price scenarios (A live-low -3.00/cast
    vs B max allowed +5% fill -30.15/cast), exposing the +5% fill loss
    instead of recycling the live-low margin. Includes below/at/above
    boundary fixtures and a profit-side negative regression for the
    comparator itself. Scoped to the members low-alchemy entry only."""
    if data.get("id") != "low-alch-members":
        return []
    errs = []

    # 1. comparator boundary fixtures (pure function, entry's own constants)
    INIT, RESERVE = 1_308_290, 162_990  # entry capital / 6-batch reserve
    fixtures = [
        ("below (loss 162,989)", INIT - 162_989, False),
        ("at boundary (loss 162,990, strict >)", INIT - 162_990, False),
        ("above (loss 162,991)", INIT - 162_991, True),
        ("profit side (assets 50,000 above initial)", INIT + 50_000, False),
        ("breakeven (assets == initial)", INIT, False),
    ]
    for label, total_assets, expect in fixtures:
        got = _loss_reserve_stop(INIT, total_assets, RESERVE)
        if got != expect:
            errs.append(f"{rel}: loss-reserve comparator fixture '{label}' "
                        f"returned {got}, expected {expect}")

    # 2. entry direction: correct form present, reversed form absent
    if "initial capital - total assets > 162,990" not in raw_text:
        errs.append(f"{rel}: loss-reserve stop must be expressed as "
                    f"'initial capital - total assets > 162,990' everywhere")
    if "total assets - initial capital" in raw_text:
        errs.append(f"{rel}: loss-reserve stop uses the reversed sign form — "
                    f"on a loss it is negative and can never exceed the "
                    f"reserve; use 'initial capital - total assets > 162,990'")

    # 3. realized GP bounds must come from the fill-price scenarios
    eco = data.get("economy", {})
    if eco.get("gp_per_hour_low") != -29463:
        errs.append(f"{rel}: gp_per_hour_low must be -29,463 (max-fill "
                    f"scenario B -30.15/cast x session-fast 977.2 casts/hr "
                    f"= -29,462.6), got {eco.get('gp_per_hour_low')}")
    if eco.get("gp_per_hour_high") != -1758:
        errs.append(f"{rel}: gp_per_hour_high must be -1,758 (live-low "
                    f"scenario A -3.00/cast x session-slow 585.9 casts/hr "
                    f"= -1,757.7), got {eco.get('gp_per_hour_high')}")
    for pin, label in (("-30.15", "max-fill per-cast loss -30.15"),
                       ("570.15", "max-fill per-cast cost 570.15"),
                       ("423.15", "max-fill candidate price 423.15")):
        if pin not in raw_text:
            errs.append(f"{rel}: economy FILL-PRICE SCENARIOS missing {label} "
                        f"({pin}) — the +5% accepted-fill scenario must be "
                        f"declared so the realized GP bounds cannot silently "
                        f"revert to the live-low-only margin")
    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())

        # loss-reserve direction + fill-price GP bound regression (owner
        # guidance 2026-08-29 for governor 10361), scoped to low-alch-members
        errors += alch_accounting_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())