00bfc358cb
Owner decision 2026-08-28: single global GE tax formula for all entries: min(floor(0.02 * sell_price), 5_000_000). Apply the per-item cap inside ge_floor_tax_error(); the margin-tag scope (flipping/merchanting) and the generic-ge non-margin exemption are unchanged. Regression fixtures now cover: - fractional-tax flipping entry must FAIL (kept) - non-margin entry with only ge + processing must PASS (kept) - correctly-floored flipping entry must PASS (kept) - high-value flip above the 5,000,000 GP cap (sell 300,000,000 / buy 294,000,000 / net 1,000,000) must PASS — the old uncapped rule computed 6,000,000 tax, demanded net 0, and rejected this valid entry (new) Closes the governor's attempt-3 blocker on PR !7.
246 lines
10 KiB
Python
246 lines
10 KiB
Python
#!/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)
|
|
- GE-margin entries (tags include flipping and/or merchanting —
|
|
the unambiguous margin signals): the 2% GE convenience fee is
|
|
rounded DOWN to whole coins per item and capped at 5,000,000 GP
|
|
per item, so net_per_cycle MUST equal
|
|
income - min(floor(0.02*income), 5_000_000) - cost.
|
|
Fractional-tax arithmetic (0.98*sell - buy) and uncapped tax
|
|
(6M on a 300M sale) are errors.
|
|
NOTE: the generic `ge` tag alone does NOT gate this rule — a
|
|
training/processing/procurement entry may legitimately carry
|
|
`ge` while its income/cost model has nothing to do with flip tax.
|
|
- Regression fixtures (--selftest): a fractional-tax flipping
|
|
entry must FAIL; an above-cap high-value flip with the
|
|
correctly capped-tax net must PASS; a non-margin entry carrying
|
|
only `ge` plus a non-flip tag must PASS.
|
|
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] [--selftest]
|
|
--selftest: run the GE floor-tax regression fixtures only (no tree scan)
|
|
"""
|
|
import datetime
|
|
import json
|
|
import math
|
|
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)
|
|
|
|
|
|
MARGIN_TAGS = {"flipping", "merchanting"}
|
|
|
|
|
|
def ge_floor_tax_error(rel, data):
|
|
"""GE floor-tax gate for a single entry.
|
|
|
|
Applies ONLY to the unambiguous margin tags (flipping/merchanting).
|
|
The generic `ge` tag is not proof that income_per_cycle is a taxed
|
|
GE resale: a training/processing/procurement entry may carry `ge`
|
|
while modeling unrelated income/cost, so it must not be gated by
|
|
this rule. Returns an error message, or None when the entry is not
|
|
gated or its floored-tax arithmetic is correct.
|
|
|
|
Wiki: "The 2% tax rounds down to the nearest whole number, so items
|
|
sold for below 50 coins have a tax of less than one coin" — per-item
|
|
net MUST use the floored tax, never 0.98*sell - buy. The fee is also
|
|
capped at 5,000,000 GP per item (per-item cap, documented in the
|
|
flip-candidates best_practice/risk), so tax = min(floor(0.02*sell),
|
|
5_000_000). Owner decision 2026-08-28: one global formula for all
|
|
entries (option A), no per-entry tax-model discriminator.
|
|
"""
|
|
tags = {str(t).lower() for t in data.get("tags", [])}
|
|
if not tags & MARGIN_TAGS:
|
|
return None
|
|
eco = data.get("economy") or {}
|
|
if not all(k in eco for k in ("income_per_cycle", "cost_per_cycle", "net_per_cycle")):
|
|
return None
|
|
sell, buy = eco["income_per_cycle"], eco["cost_per_cycle"]
|
|
tax = min(math.floor(sell * 0.02), 5_000_000)
|
|
expected = sell - tax - buy
|
|
if eco["net_per_cycle"] != expected:
|
|
return (
|
|
f"{rel}: net_per_cycle {eco['net_per_cycle']} != capped floored-tax "
|
|
f"model {expected} (sell {sell} - min(floor({sell}*0.02), 5_000_000)"
|
|
f"={tax} - buy {buy}); "
|
|
f"GE tax rounds down per item and is capped at 5,000,000 GP per item "
|
|
f"— fractional 0.98*sell - buy and uncapped tax are invalid")
|
|
return None
|
|
|
|
|
|
def run_selftest(verbose=False):
|
|
"""Regression fixtures for the GE floor-tax gate; returns error strings."""
|
|
fixtures = [
|
|
# name, tags, economy, expect_error
|
|
("fractional-tax flipping entry", ["flipping"],
|
|
{"income_per_cycle": 244, "cost_per_cycle": 223, "net_per_cycle": 16.12}, True),
|
|
# 5000 - floor(5000*0.02) - 3000 = 1900 != 2000: this fixture's
|
|
# economics deliberately do NOT satisfy the flip equation, so the
|
|
# old `ge`-tag scoping would false-positive on it (2000 != 1900).
|
|
("non-margin entry with only 'ge' + 'processing'", ["ge", "processing"],
|
|
{"income_per_cycle": 5000, "cost_per_cycle": 3000, "net_per_cycle": 2000}, False),
|
|
("correctly-floored flipping entry", ["flipping", "merchanting"],
|
|
{"income_per_cycle": 244, "cost_per_cycle": 223, "net_per_cycle": 17}, False),
|
|
# sell 300,000,000 -> uncapped floor tax is 6,000,000; the per-item
|
|
# cap makes it 5,000,000, so the correct net is 1,000,000. The old
|
|
# uncapped rule demanded net=0 and rejected this valid entry.
|
|
("high-value flip above the 5,000,000 GP per-item tax cap",
|
|
["flipping"],
|
|
{"income_per_cycle": 300_000_000, "cost_per_cycle": 294_000_000,
|
|
"net_per_cycle": 1_000_000}, False),
|
|
]
|
|
errors = []
|
|
for name, tags, eco, expect_error in fixtures:
|
|
rel = f"selftest/{name}"
|
|
got_error = ge_floor_tax_error(rel, {"tags": tags, "economy": eco}) is not None
|
|
passed = got_error == expect_error
|
|
if verbose:
|
|
print(f"SELFTEST {'PASS' if passed else 'FAIL'} {name}")
|
|
if not passed:
|
|
errors.append(
|
|
f"selftest: '{name}' expected "
|
|
f"{'an error' if expect_error else 'no error'} but got "
|
|
f"{'an error' if got_error else 'no error'}")
|
|
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
|
|
|
|
# The fix itself is gated: regression fixtures run on every invocation.
|
|
errors += run_selftest(verbose=False)
|
|
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")
|
|
|
|
# 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")
|
|
|
|
# GE-margin floor-tax arithmetic (see ge_floor_tax_error): scoped to
|
|
# the unambiguous margin tags flipping/merchanting. The generic `ge`
|
|
# tag alone must not gate this rule — alchemy/processing/procurement
|
|
# entries can carry `ge` with income unrelated to a taxed GE resale.
|
|
err = ge_floor_tax_error(rel, data)
|
|
if err:
|
|
errors.append(err)
|
|
|
|
# 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())
|