osrs-codex/tools/validate.py
Claude 34f36d2275 codex: dynamic SII recipe selection + realized rates + validator crash fix (bounce fix attempt 3, owner guidance)
Owner decision (2026-08-28): Superheat Item uses a dynamic recipe table
governed by Smithing level — highest UNLOCKED recipe selected, primary ore
(target_input, never coal) is the spell target, fallback to the next lower
unlocked recipe, else skip SII and stay on LA.

- magic-alch-handoff: per-recipe target_input; selection block
  (mode highest-unlocked / eligible_rule / fallback); full-lifecycle
  actions (buy-staff, equip-staff w/ fire-rune failure path, buy-supplies,
  cast-alch, cast-superheat, bank-loop, take-break); deterministic restock
  (LA/HLA carried_batch 1,000 + threshold 100 -> 901-cast batches; SII
  24-cast bank cycles, bars >= 24); breaks [400,700] COMPLETED casts,
  independent of banking; worked example pinned (prep -> first restock @24
  -> break @550 -> resume -> restock @552); stop conditions + recovery block
- rates REALIZED incl. mandatory breaks + banking: LA 24,500 / SII 27,500 /
  HLA 51,400 xp/hr (formulas pinned; loop-only 837/hr and 305/hr regimes
  superseded); low-alch-f2p also switched (xp 24,500, gp -110,500..-74,000)
- validate.py: rows now collected ONLY after schema validation (schema-
  invalid MAPPINGS incl. 'economy: not-a-mapping' produce normal ERRORs,
  never AttributeError tracebacks — end-to-end subprocess regression added);
  new gates: recipe_model_errors (target_input in declared inputs, coal
  never targetable, unique smithing gates, selection block + SII/fallback
  transitions) and rate_errors (xp_per_hour must match the realized cadence,
  tolerance 2%); highest_unlocked_recipe helper + 6 eligibility fixtures
- entry.schema.json: optional selection block + per-recipe target_input
- Evidence: validate 2 entries 0 errors 0 warnings; --selftest 31 PASS;
  governor's exact direct repro supply_limit_conflicts([('bad.yaml',
  {'economy':'not-a-mapping'})]) -> [] no crash; synthetic 3-way merge of
  main 79cab78 (incl. issue-6 catalog) + head: clean, gate green; git
  diff --check clean
2026-08-28 03:11:07 -05:00

609 lines
29 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)
- 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
declare a `selection:` block (mode highest-unlocked, eligible rule,
fallback) and the state machine must cover both the SII window and
the stay-on-LA fallback. Every declared recipe must pin a
`target_input` that is one of its declared inputs and never coal
(the spell targets the primary ore), and smithing_level gates must
be unique so "highest unlocked" selection is deterministic.
- Realized-rate consistency: xp_per_hour must derive from the pinned
realized cadence (mandatory additive delays PLUS breaks PLUS
banking), not from the loop-only 837 casts/hr or the superseded
8-14-cast-break regime (~305 casts/hr).
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. The same
holds for schema-INVALID MAPPINGS (e.g. `economy: not-a-mapping`):
shared-gate rows are only collected after schema validation succeeds
(regression: a malformed nested field used to crash
`supply_limit_conflicts` with an AttributeError).
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")
if not isinstance(eco, dict): # schema-invalid nested mapping (e.g. economy: "x")
continue
for sup in eco.get("supplies", []):
if not isinstance(sup, dict):
continue
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)
# --- realized-cadence constants (pinned model; owner guidance 2026-08-28) ---
# docs/action-grammar.md:135: anti_ban.delay is an ADDITIVE post-action sleep,
# so a cast = 1.8s (3 ticks x 0.6s) + [1.6, 3.4]s = 3.4-5.2s (mid 4.3s).
# LA/HLA realized casts/hr: 901-cast restock batch (carried_batch 1,000 -
# restock_threshold 100) with 225s overhead per batch (90s restock trip +
# 135s breaks at every [400,700] casts, mid 550) -> 791.3 casts/hr midpoint.
# SII realized casts/hr: 24-cast bank cycle (non-stackable bars fill the
# inventory) -> 518.0 casts/hr midpoint (550-cast break window: 2,365s
# casting + 22.9 trips x 60s + 82.5s break).
REALIZED_CastsPerHr_ALCH = 791.3
REALIZED_CastsPerHr_SII = 518.0
XP_LA, XP_SII, XP_HLA = 31, 53, 65
RATE_TOLERANCE = 0.02 # pinned values may round to 3 significant figures
def highest_unlocked_recipe(recipes, smithing_level):
"""Deterministic Superheat selection: the declared recipe with the
highest smithing_level <= the account's Smithing level, else None.
Uniqueness of the gates is enforced by recipe_model_errors, so this
helper's ordering cannot be ambiguous. An account can NEVER be
offered a recipe above its Smithing level.
"""
best, best_lvl = None, -1
for rid, recipe in (recipes or {}).items():
lvl = recipe.get("smithing_level") if isinstance(recipe, dict) else None
if isinstance(lvl, int) and lvl <= smithing_level and lvl > best_lvl:
best, best_lvl = rid, lvl
return best
def recipe_model_errors(rel, data):
"""SII recipe-model + selection gates (owner guidance 2026-08-28).
- every declared recipe pins a `target_input` that is one of its
declared inputs and is never coal (the spell targets the primary
ore — Superheat cannot be cast on coal);
- smithing_level gates are unique, so "highest unlocked" selection
is deterministic;
- an entry casting Superheat Item must declare a `selection:` block
(mode highest-unlocked + eligible_rule + fallback) and its state
machine must cover both the SII window (43-55) and the fallback
(skip SII / stay on LA / next unlocked recipe).
"""
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"]
if isinstance(recipes, dict):
seen_levels = {}
for rid, recipe in recipes.items():
if not isinstance(recipe, dict):
continue # schema reports the shape
inputs = recipe.get("inputs") or []
in_names = [str(i.get("item", "")).lower()
for i in inputs if isinstance(i, dict)]
ti = recipe.get("target_input")
if not isinstance(ti, str) or not ti:
errors.append(
f"{rel}: recipe '{rid}' has no target_input — the spell "
f"targets the primary ore and it must be pinned per recipe")
elif ti.lower() not in in_names:
errors.append(
f"{rel}: recipe '{rid}' target_input '{ti}' is not one of "
f"its declared inputs")
if ti and (ti.lower() == "coal" or ti.strip().isdigit()
and int(ti) in {453}):
errors.append(
f"{rel}: recipe '{rid}' target_input '{ti}' is coal — coal "
f"can never be the Superheat spell target")
lvl = recipe.get("smithing_level")
if isinstance(lvl, int):
if lvl in seen_levels:
errors.append(
f"{rel}: recipes '{seen_levels[lvl]}' and '{rid}' share "
f"smithing_level {lvl} — highest-unlocked selection "
f"would be ambiguous")
seen_levels[lvl] = rid
has_sii = any(
"superheat" in (str(a.get("params") or {}).lower()
+ str(a.get("target") or "").lower()
+ str(a.get("step") or "").lower())
for a in data.get("actions", []))
if has_sii:
sel = data.get("selection")
if not isinstance(sel, dict):
errors.append(
f"{rel}: a Superheat Item action requires a 'selection' block "
f"(mode, eligible_rule, fallback)")
else:
if sel.get("mode") != "highest-unlocked":
errors.append(
f"{rel}: selection.mode must be 'highest-unlocked'")
if not (isinstance(sel.get("eligible_rule"), str)
and len(sel["eligible_rule"]) >= 10):
errors.append(
f"{rel}: selection.eligible_rule must state the eligibility "
f"rule (smithing gate + complete-recipe margin)")
if not (isinstance(sel.get("fallback"), str)
and len(sel["fallback"]) >= 10):
errors.append(
f"{rel}: selection.fallback must state the fallback path "
f"(next unlocked recipe; else skip SII / stay on LA)")
whens = [str(t.get("when", ""))
for t in data.get("state_machine", {}).get("transitions", [])]
if not any("43-55" in w for w in whens):
errors.append(
f"{rel}: no transition covers the SII window (43-55) — the "
f"selected recipe tier is unreachable")
if not any(("stay on la" in w or "skip" in w or "fallback" in w)
for w in whens):
errors.append(
f"{rel}: no transition covers the SII fallback (skip SII / "
f"stay on LA) — selection is not deterministic")
return errors
def rate_errors(rel, data):
"""xp_per_hour must derive from the REALIZED cadence: mandatory
additive delays PLUS the pinned mandatory breaks PLUS banking.
Regression guard: the loop-only figure (837 casts/hr) and the
superseded 8-14-cast-break regime (~305 casts/hr) both sit outside
the tolerance band around the realized rate.
"""
errors = []
eco = data.get("economy")
if not isinstance(eco, dict):
return errors
xph = eco.get("xp_per_hour")
if not isinstance(xph, (int, float)):
return errors
has_sii = any(
"superheat" in (str(a.get("params") or {}).lower()
+ str(a.get("target") or "").lower()
+ str(a.get("step") or "").lower())
for a in data.get("actions", []))
per_cast = XP_HLA if has_sii else XP_LA
expected = per_cast * REALIZED_CastsPerHr_ALCH
if abs(xph - expected) / expected > RATE_TOLERANCE:
errors.append(
f"{rel}: xp_per_hour {xph} is not the realized cadence rate "
f"(expected ~{expected:.0f} = {per_cast} xp/cast x "
f"{REALIZED_CastsPerHr_ALCH} casts/hr including mandatory breaks "
f"and banking; tolerance {RATE_TOLERANCE:.0%})")
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)]) == [])
check("supply_limit_conflicts ignores schema-invalid nested mappings (economy: not-a-mapping) without raising",
supply_limit_conflicts([("bad.yaml", {"economy": "not-a-mapping"})]) == [])
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)
check("end-to-end: schema-invalid MAPPING -> exit 1, clean ERROR, no Traceback",
_end_to_end_bad_mapping())
# --- deterministic multi-input recipe fixture: steel-bar (1 iron + 2 coal) ---
steel = {"bar": "Steel bar", "bar_item_id": 2353, "smithing_level": 30,
"target_input": "Iron ore",
"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)
# --- recipe-model fixtures: target_input, unique gates, selection block ---
f2p_bars = {
"bronze-bar": {"bar": "Bronze bar", "bar_item_id": 2349, "smithing_level": 1,
"smithing_xp": 6.2, "target_input": "Copper ore",
"inputs": [{"item": "Copper ore", "item_id": 436, "qty": 1, "source": "ge"},
{"item": "Tin ore", "item_id": 438, "qty": 1, "source": "ge"}]},
"iron-bar": {"bar": "Iron bar", "bar_item_id": 2351, "smithing_level": 15,
"smithing_xp": 12.5, "target_input": "Iron ore",
"inputs": [{"item": "Iron ore", "item_id": 440, "qty": 1, "source": "ge"}]},
"silver-bar": {"bar": "Silver bar", "bar_item_id": 2355, "smithing_level": 20,
"smithing_xp": 13.7, "target_input": "Silver ore",
"inputs": [{"item": "Silver ore", "item_id": 442, "qty": 1, "source": "ge"}]},
"steel-bar": steel,
"runite-bar": {"bar": "Runite bar", "bar_item_id": 2363, "smithing_level": 85,
"smithing_xp": 50, "target_input": "Runite ore",
"inputs": [{"item": "Runite ore", "item_id": 451, "qty": 1, "source": "ge"},
{"item": "Coal", "item_id": 453, "qty": 8, "source": "ge"}]},
}
# Smithing-level eligibility: highest unlocked, never an unmet gate
check("smithing 1 -> bronze-bar (highest unlocked)",
highest_unlocked_recipe(f2p_bars, 1) == "bronze-bar")
check("smithing 15 -> iron-bar",
highest_unlocked_recipe(f2p_bars, 15) == "iron-bar")
check("smithing 20 -> silver-bar (steel at 30 stays locked)",
highest_unlocked_recipe(f2p_bars, 20) == "silver-bar")
check("smithing 30 -> steel-bar",
highest_unlocked_recipe(f2p_bars, 30) == "steel-bar")
check("smithing 85 -> runite-bar",
highest_unlocked_recipe(f2p_bars, 85) == "runite-bar")
check("smithing 0 -> None (no recipe selected below the lowest gate)",
highest_unlocked_recipe(f2p_bars, 0) is None)
superheat_action = {
"id": "a6", "step": "cast-superheat", "verb": "use_item_on",
"target": "the selected recipe's declared target_input in inventory (primary ore — never coal)",
"params": {"spell": "Superheat Item",
"selected_recipe": "runtime selection per selection.mode: highest-unlocked recipe with a positive complete-cost margin",
"target_input": "the selected recipe's declared target_input field"},
"on_fail": "skip"}
selection_ok = {"mode": "highest-unlocked",
"eligible_rule": "recipes entries with smithing_level <= current Smithing and a positive complete-cost margin",
"fallback": "next lower unlocked recipe when materials cannot be obtained; else skip SII and stay on LA"}
transitions_ok = [
{"from": "cast", "when": "level 43-55 and the SII window is chosen (selection.mode picks the highest unlocked recipe with positive margin)", "to": "cast"},
{"from": "cast", "when": "no unlocked recipe has a positive margin this window — skip SII and stay on LA", "to": "cast"}]
entry_model_ok = {"actions": [superheat_action], "recipes": f2p_bars,
"selection": selection_ok,
"state_machine": {"transitions": transitions_ok}}
check("recipe model with declared target_inputs + selection block passes",
recipe_model_errors("ok.yaml", entry_model_ok) == [])
# coal can never be the spell target
entry_coal = {"actions": [superheat_action],
"recipes": {"steel-bar": dict(steel, target_input="Coal")},
"selection": selection_ok,
"state_machine": {"transitions": transitions_ok}}
coal_errs = recipe_model_errors("bad.yaml", entry_coal)
check("steel-bar target_input 'Coal' is rejected (coal is never the spell target)",
any("is coal" in e for e in coal_errs))
# target_input must be a declared input
entry_bad_ti = {"actions": [superheat_action],
"recipes": {"steel-bar": dict(steel, target_input="Runite ore")},
"selection": selection_ok,
"state_machine": {"transitions": transitions_ok}}
bad_ti = recipe_model_errors("bad.yaml", entry_bad_ti)
check("target_input not among the recipe's declared inputs is rejected",
any("not one of" in e for e in bad_ti))
# duplicate smithing gates would make highest-unlocked ambiguous
entry_dup = {"actions": [superheat_action],
"recipes": {"steel-bar": steel,
"steel-bar-2": dict(steel, bar="Steel bar 2", bar_item_id=2353)},
"selection": selection_ok,
"state_machine": {"transitions": transitions_ok}}
dup_errs = recipe_model_errors("bad.yaml", entry_dup)
check("duplicate smithing_level across recipes is rejected (ambiguous highest-unlocked)",
any("share smithing_level" in e for e in dup_errs))
# Superheat action requires the selection block + fallback transitions
entry_no_sel = {"actions": [superheat_action], "recipes": f2p_bars,
"state_machine": {"transitions": transitions_ok}}
no_sel = recipe_model_errors("bad.yaml", entry_no_sel)
check("Superheat action without a selection block is rejected",
any("'selection' block" in e for e in no_sel))
entry_no_fb = {"actions": [superheat_action], "recipes": f2p_bars,
"selection": selection_ok,
"state_machine": {"transitions": [
{"from": "cast", "when": "level 43-55 and the SII window is chosen", "to": "cast"}]}}
no_fb = recipe_model_errors("bad.yaml", entry_no_fb)
check("Superheat entry without an SII fallback transition is rejected",
any("SII fallback" in e for e in no_fb))
# --- realized-rate fixtures: breaks + banking included ---
check("LA-only xp_per_hour 24,500 matches the realized cadence (31 x 791.3)",
rate_errors("ok.yaml", {"economy": {"xp_per_hour": 24500}}) == [])
check("loop-only LA xp_per_hour 25,953 fails the realized-cadence gate",
any("realized cadence" in e
for e in rate_errors("bad.yaml", {"economy": {"xp_per_hour": 25953}})))
check("8-14-cast-break regime LA rate (31 x ~305 = 9,455) fails the realized-cadence gate",
any("realized cadence" in e
for e in rate_errors("bad.yaml", {"economy": {"xp_per_hour": 9455}})))
check("SII entry xp_per_hour 51,400 matches the HLA-tier realized cadence (65 x 791.3)",
rate_errors("ok2.yaml", {"actions": [superheat_action],
"economy": {"xp_per_hour": 51400}}) == [])
check("SII entry with the old loop-only HLA rate 54,419 fails the realized-cadence gate",
any("realized cadence" in e
for e in rate_errors("bad2.yaml", {"actions": [superheat_action],
"economy": {"xp_per_hour": 54419}})))
return errors
def _end_to_end_bad_mapping():
"""Regression (governor bounce 3, finding 1): a schema-invalid MAPPING
with a malformed nested field — `economy: not-a-mapping` — must yield
exit 1 with a normal ERROR line and NO traceback.
Writes the fixture into the real codex tree and shells out to the real
validator so the full main() path is exercised, then removes the file.
Returns True when the gate behaves.
"""
import subprocess
bad = ROOT / "codex" / "selftest-bad-mapping.yaml"
try:
bad.write_text("id: selftest-bad-mapping\neconomy: not-a-mapping\n")
proc = subprocess.run(
[sys.executable, str(Path(__file__).resolve()), "--stale-days", "9999"],
capture_output=True, text=True, cwd=str(ROOT), timeout=120)
out = (proc.stdout or "") + (proc.stderr or "")
return (proc.returncode == 1 and "selftest-bad-mapping" in out
and "ERROR" in out and "Traceback" not in out)
finally:
bad.unlink(missing_ok=True)
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
# 1. schema — BEFORE the shared row set. Schema-invalid documents
# (including mappings with malformed nested fields like
# `economy: not-a-mapping`) must never reach the cross-entry gates:
# regression — rows.append() used to run first, and
# supply_limit_conflicts() then crashed with an AttributeError.
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
rows.append((rel, data))
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 recipe-model + selection gates + realized-rate consistency
errors += recipe_model_errors(rel, data)
errors += rate_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())