osrs-codex/tools/validate.py

1088 lines
55 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 (owner guidance 2026-08-28, bounce 4):
entries that DECLARE a `rate_model` must have xp_per_hour derived
from their OWN declared lifecycle — cast ticks + the casting
action's mandatory anti_ban.delay (additive), run_control breaks,
per-recipe inventory capacity (floor(slots / total input qty)),
restock trips, and startup amortised over the session horizon.
The gate is behind an explicit discriminator: entries without
`rate_model` (e.g. mining) are never treated as alchemy. Every
declared SII recipe must pin `rates` (capacity + min/mid/max
casts/hr + Magic xp/hr) matching the derived model, and capacity
must fit a 28-slot inventory (1 nature stack -> 27 input slots).
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 copy
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)
# --- rate-model constants (owner guidance 2026-08-28, bounce 4) ---
# The cadence is DERIVED from each entry's declared rate_model + actions +
# run_control + recipes — nothing is hard-coded per activity class. These
# are the only universal facts:
# * one game tick = 0.6 s (docs/action-grammar.md);
# * anti_ban.delay is an ADDITIVE post-action sleep
# (docs/action-grammar.md:135), so cycle = cast_ticks*0.6 + delay;
# * LA/SII/HLA Magic XP per cast (wiki infoboxes): 31 / 53 / 65.
XP_LA, XP_SII, XP_HLA = 31, 53, 65
RATE_TOLERANCE = 0.02 # pinned values may round to 3 significant figures
def _mean(pair):
return (pair[0] + pair[1]) / 2.0 if pair else 0.0
def target_reached(level, target):
"""Magic-level stop boundary (governor 10394 finding 3): the run stops
when level >= target. Default sessions target 55 (the LA/SII -> HLA
handoff); a session at level 54 with target 55 CONTINUES (False), and
level 55 hands off (True)."""
return level >= target
def selectable_tier(level, target, sii_skipped=False):
"""Executable tier selection (governor 10477): exactly ONE tier
predicate is true at any (level, target) — the predicates are mutually
exclusive, so no Magic level can match two tiers (the old '43-55' SII
window overlapped HLA at exactly 55 for targets above 55).
Order of evaluation (completion always wins first):
- 'complete' when level >= target (the configured target always
beats any remaining tier predicate);
- 'hla' when Magic >= 55 AND the target is higher
(level >= 55 but < target implies target > 55);
- 'sii' only while 43 <= Magic < 55 (never at 55+);
- 'la' level < 43, OR the explicit SII-skip path while
43 <= Magic < 55.
"""
if level >= target:
return "complete"
if level >= 55:
return "hla" # target > 55 implied by level < target
if 43 <= level < 55 and not sii_skipped:
return "sii"
return "la" # level < 43, or 43-54 with SII skipped
def tier_boundary_errors(rel, data):
"""Executable tier mutual-exclusivity (governor 10477): no Magic level
may match TWO tier predicates at once. The old inclusive '43-55' SII
window overlapped HLA (>= 55) at exactly Magic 55 for targets above
55, so this gate:
- rejects an SII window spelled as the inclusive '43-55';
- requires SII to be declared strictly below 55
('43 <= Magic < 55' / '43-54');
- requires an HLA transition that names Magic >= 55 AND a session
target above 55 (HLA is executable, never declared-but-unreachable);
- requires a completion transition at the configured target.
Only entries that cast Superheat Item are scrutinised here (the
LA-only entry has no tier overlap and is untouched)."""
errors = []
transitions = data.get("state_machine", {}).get("transitions", [])
whens = [str(t.get("when", "")) for t in transitions]
has_sii = "superheat" in (str(data.get("actions", []))).lower()
if not has_sii:
return []
# Reject the inclusive '43-55' spelling (the overlap defect).
if any("43-55" in w for w in whens):
errors.append(
f"{rel}: SII window spelled as the inclusive '43-55' overlaps "
f"HLA at exactly Magic 55 — declare SII strictly below 55 "
f"(43 <= Magic < 55), HLA at >= 55 requiring the target above 55")
# The SII window must exist in strictly-below-55 form.
if not any(("43 <= " in w and "< 55" in w) or "43-54" in w
for w in whens):
errors.append(
f"{rel}: no transition declares the SII window strictly below "
f"55 (43 <= Magic < 55) — SII must never be selectable at 55+")
# HLA requires both >= 55 and a target above 55 in the same clause.
if not any((">= 55" in w and "> 55" in w) for w in whens):
errors.append(
f"{rel}: no mutually-exclusive HLA transition (Magic >= 55 AND "
f"session target > 55) — HLA would overlap SII or be unreachable")
return errors
def stop_loss_exceeded(initial_capital, total_assets, reserve):
"""Sign-corrected capital-loss stop (governor 10394 finding 2):
exceeded when initial capital - total assets > reserve. total_assets -
initial capital goes NEGATIVE on losses and can never exceed the
reserve — the reversed sign would never fire."""
return (initial_capital - total_assets) > reserve
def loadout_verified(carried, required):
"""Build-loadout proof (governor 10394 finding 1): the carried counts
must EQUAL the required carried amounts before the first cast. A
shortfall is not verified (False) — it routes to bank_supplies."""
return all(carried.get(k, 0) >= qty for k, qty in required.items())
def _casting_delay(data):
"""The mandatory post-action delay of the entry's casting action(s),
converted from ms (anti_ban.delay units) to seconds.
Derived from the entry's OWN actions: the first action whose step
starts with 'cast' and carries an anti_ban.delay pair. Returns
(delay_min_s, delay_max_s) or None.
"""
for a in data.get("actions", []):
if not isinstance(a, dict):
continue
if not str(a.get("step", "")).startswith("cast"):
continue
delay = (a.get("anti_ban") or {}).get("delay")
if isinstance(delay, list) and len(delay) == 2:
return float(delay[0]) / 1000.0, float(delay[1]) / 1000.0
return None
def _break_amort(data):
"""Break time per cast from the entry's OWN run_control.break.
break.every counts completed casts (one-cast iteration model), so the
amortised break share = mean(break.for) / mean(break.every). Fast/slow
ends use for_min/every_max and for_max/every_min respectively.
"""
brk = (data.get("run_control") or {}).get("break") or {}
every, for_ = brk.get("every"), brk.get("for")
if not (isinstance(every, list) and len(every) == 2
and isinstance(for_, list) and len(for_) == 2):
return None
e_min, e_max = float(every[0]), float(every[1])
f_min, f_max = float(for_[0]), float(for_[1])
return {
"fast": f_min / e_max, "mid": _mean(for_) / _mean(every),
"slow": f_max / e_min}
def _session_factor(rm, pace):
"""Startup amortisation factor = 1 - startup_s / session_horizon_s.
pace selects the matching end of the startup range: fast cast pace
pairs with the fast (short) startup, slow with the slow one, mid with
the midpoint. Returns None when the horizon is missing or startup
exceeds the horizon (impossible session)."""
startup = rm.get("startup_s") or [300, 600]
horizon = rm.get("session_horizon_s")
if not isinstance(horizon, (int, float)) or horizon <= 0:
return None
if not (isinstance(startup, list) and len(startup) == 2):
return None
s_min, s_max = float(startup[0]), float(startup[1])
pick = {"fast": s_min, "mid": _mean(startup), "slow": s_max}[pace]
if pick >= horizon:
return None
return 1.0 - pick / horizon
def recipe_capacity(recipe, slots):
"""Inventory capacity of a recipe: floor(slots / total input qty).
SII bars are non-stackable and one slot holds the nature rune stack,
so a 28-slot inventory leaves `slots` (=27) for recipe inputs. A
recipe whose inputs exceed the slots cannot exist (capacity 0)."""
qty = recipe_total_input_qty(recipe)
if qty <= 0:
return 0
return int(slots // qty)
def _alch_pace_casts_hr(cycle, trip, break_share, batch_casts):
"""Steady-state realized casts/hr for an alch tier (LA/HLA), derived
from the batch model (carried_batch - restock_threshold casts per
restock trip) plus breaks."""
batch_s = batch_casts * cycle + trip + break_share * batch_casts
if batch_s <= 0:
return 0.0
return batch_casts * 3600.0 / batch_s
def rate_model_errors(rel, data):
"""xp_per_hour and per-recipe rates must derive from the entry's OWN
declared lifecycle (owner guidance 2026-08-28, bounce 4).
The gate is OPT-IN behind an explicit `rate_model` discriminator:
entries without one (e.g. a mining entry) are never treated as
alchemy. For entries that declare it, the model is derived from the
entry's own fields — cast_ticks + the casting action's mandatory
anti_ban.delay, run_control.break, restock trips, the recipes table
(capacity = floor(slots / total input qty)), and startup amortised
over session_horizon_s. Regression coverage (--selftest):
non-alchemy pass, loop-only/old-regime failures, cadence-mutation
failure (one-hour breaks), per-recipe capacity and rate pins."""
def err(msg):
return f"{rel}: {msg}"
rm = data.get("rate_model")
if not isinstance(rm, dict):
return [] # explicit discriminator — unrelated entries pass
errors = []
tick_s = 0.6
cast_ticks = rm.get("cast_ticks")
if not isinstance(cast_ticks, (int, float)) or cast_ticks < 1:
errors.append(err("rate_model.cast_ticks must be >= 1"))
return errors
cast_s = cast_ticks * tick_s
delay = _casting_delay(data)
if delay is None:
errors.append(err("rate_model requires a casting action (step 'cast-*') "
"with anti_ban.delay [min, max] — the rate derives from "
"the entry's own mandatory pacing"))
return errors
cycle = {"fast": cast_s + delay[0], "mid": cast_s + _mean(delay),
"slow": cast_s + delay[1]}
break_amort = _break_amort(data)
if break_amort is None:
errors.append(err("rate_model requires run_control.break with every + for "
"ranges (cast-count iteration model)"))
return errors
trip = rm.get("restock_trip_s")
if not (isinstance(trip, list) and len(trip) == 2):
errors.append(err("rate_model.restock_trip_s must be [min, max] seconds"))
return errors
trip = {"fast": float(trip[0]), "mid": _mean(trip), "slow": float(trip[1])}
slots = rm.get("sii_capacity_slots", 27)
# --- per-recipe SII rates (capacity + min/mid/max), fully derived ---
recipes = data.get("recipes")
if isinstance(recipes, dict):
for rid, recipe in recipes.items():
if not isinstance(recipe, dict):
continue
cap = recipe_capacity(recipe, slots)
rates = recipe.get("rates")
if not isinstance(rates, dict):
errors.append(err(f"recipe '{rid}' must pin a rates block "
f"(capacity, casts_per_hour, magic_xp_per_hour)"))
continue
if cap < 1:
errors.append(err(f"recipe '{rid}' cannot fit the inventory: "
f"total input qty {recipe_total_input_qty(recipe)} "
f"> {slots} input slots (capacity 0)"))
continue
pinned_cap = rates.get("capacity")
if pinned_cap != cap:
errors.append(err(f"recipe '{rid}' rates.capacity {pinned_cap} != "
f"derived floor({slots} / {recipe_total_input_qty(recipe)}) = {cap}"))
casts = {}
for pace in ("min", "mid", "max"):
p = {"min": "fast", "mid": "mid", "max": "slow"}[pace]
per_cast = cycle[p] + trip[p] / cap + break_amort[p]
casts[pace] = 3600.0 / per_cast if per_cast > 0 else 0.0
pin_cph = (rates.get("casts_per_hour") or {})
pin_xp = (rates.get("magic_xp_per_hour") or {})
for pace, computed in casts.items():
expect_xp = computed * XP_SII
pc, px = pin_cph.get(pace), pin_xp.get(pace)
for label, pinned, expected in (("casts_per_hour", pc, computed),
("magic_xp_per_hour", px, expect_xp)):
if not isinstance(pinned, (int, float)):
errors.append(err(f"recipe '{rid}' rates.{label}.{pace} missing"))
elif abs(pinned - expected) / expected > RATE_TOLERANCE:
errors.append(err(
f"recipe '{rid}' rates.{label}.{pace} {pinned} is not the "
f"derived rate ({expected:.1f} from cycle {cycle[p]:.2f}s + "
f"trip {trip[p]:.1f}s/{cap} capacity + break {break_amort[p]:.3f}s; "
f"tolerance {RATE_TOLERANCE:.0%})"))
# --- xp_per_hour anchor: the declared anchor tier, same lifecycle ---
eco = data.get("economy")
xph = eco.get("xp_per_hour") if isinstance(eco, dict) else None
if not isinstance(xph, (int, float)):
errors.append(err("rate_model requires economy.xp_per_hour to pin the "
"anchor tier's realized rate"))
return errors
anchor = rm.get("anchor_tier")
per_cast = {"la": XP_LA, "sii": XP_SII, "hla": XP_HLA}.get(anchor)
if per_cast is None:
errors.append(err("rate_model.anchor_tier must be la, sii or hla"))
return errors
if anchor == "sii":
rid = rm.get("anchor_recipe")
recipe = (recipes or {}).get(rid) if isinstance(recipes, dict) else None
if not isinstance(recipe, dict):
errors.append(err(f"rate_model.anchor_recipe '{rid}' is not a declared "
f"recipe — the anchor must be one of the recipes"))
return errors
cap = recipe_capacity(recipe, slots)
if cap < 1:
return errors # already reported as capacity 0
casts_mid = 3600.0 / (cycle["mid"] + trip["mid"] / cap + break_amort["mid"])
else:
batch = rm.get("alch_carry_batch")
threshold = rm.get("alch_restock_threshold")
if not (isinstance(batch, (int, float)) and batch > 1
and isinstance(threshold, (int, float)) and 0 <= threshold < batch):
errors.append(err("rate_model.alch_carry_batch and alch_restock_threshold "
"are required for la/hla anchors (threshold < batch)"))
return errors
batch_casts = batch - threshold
casts_mid = _alch_pace_casts_hr(cycle["mid"], trip["mid"],
break_amort["mid"], batch_casts)
factor = _session_factor(rm, "mid")
if factor is None:
errors.append(err("rate_model startup_s must be < session_horizon_s "
"(startup cannot exceed the session)"))
return errors
expected = per_cast * casts_mid * factor
if abs(xph - expected) / expected > RATE_TOLERANCE:
errors.append(err(
f"xp_per_hour {xph} is not the anchor tier's derived realized rate "
f"(expected ~{expected:.0f} = {per_cast} xp/cast x {casts_mid:.1f} "
f"steady-state casts/hr x startup factor {factor:.3f} over "
f"{rm.get('session_horizon_s')}s; tolerance {RATE_TOLERANCE:.0%})"))
return errors
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 select_recipe(recipes, smithing_level, bar_prices, input_prices,
nature_price, fill_factor=1.0):
"""Deterministic full selection (owner guidance 2026-08-28, bounce 4):
among the recipes whose Smithing gate is unlocked, pick the HIGHEST-gate
one whose complete-recipe margin is positive at the given prices.
fill_factor scales every input and the nature rune price to the ALLOWED
fill cost (live low + 5% -> 1.05), so eligibility uses the FULL
actual/allowed recipe cost. Returns None when no unlocked recipe has a
positive margin — the caller then skips SII and stays on LA (the
fallback chain: this helper already drops to the next lower unlocked
recipe whenever the highest one is margin-negative)."""
unlocked = []
for rid, recipe in (recipes or {}).items():
if not isinstance(recipe, dict):
continue
lvl = recipe.get("smithing_level")
if isinstance(lvl, int) and lvl <= smithing_level:
cost = sum(
int(inp.get("qty", 0)) * int(input_prices.get(inp.get("item"), 0))
for inp in recipe.get("inputs", []))
margin = (bar_prices.get(recipe.get("bar"), 0)
- (cost * fill_factor + nature_price * fill_factor))
unlocked.append((lvl, rid, margin))
unlocked.sort(key=lambda t: t[0], reverse=True)
for _lvl, rid, margin in unlocked:
if margin > 0:
return rid
return None
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", [])]
# SII window must be declared as 43 <= Magic < 55 (also accept the
# historical "43-55" spelling so legacy fixtures stay readable).
if not any(("43 <= " in w and "< 55" in w) or "43-55" in w
for w in whens):
errors.append(
f"{rel}: no transition covers the SII window (43 <= Magic "
f"< 55) — the 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 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)
# Full selection: Smithing gate + complete-recipe margin at actual and
# ALLOWED fill prices, falling back to the next lower unlocked recipe
bar_prices = {"Bronze bar": 200, "Iron bar": 380, "Silver bar": 450,
"Steel bar": 600, "Runite bar": 12500}
input_prices = {"Copper ore": 50, "Tin ore": 30, "Iron ore": 100,
"Silver ore": 150, "Coal": 150, "Runite ore": 11000}
nature = 145
check("full selection at live prices: smithing 30 + steel margin +55 -> steel-bar",
select_recipe(f2p_bars, 30, bar_prices, input_prices, nature) == "steel-bar")
check("allowed-fill eligibility (x1.05): steel margin +27.75 -> steel-bar still selected",
select_recipe(f2p_bars, 30, bar_prices, input_prices, nature,
fill_factor=1.05) == "steel-bar")
check("material availability: steel margin-negative at allowed fills -> next lower unlocked (silver-bar)",
select_recipe(f2p_bars, 30, {**bar_prices, "Steel bar": 570}, input_prices,
nature, fill_factor=1.05) == "silver-bar")
check("no unlocked recipe with a positive margin -> None (skip SII, stay on LA)",
select_recipe(f2p_bars, 30, {"Bronze bar": 10, "Iron bar": 10,
"Silver bar": 10, "Steel bar": 10},
input_prices, nature) is None)
check("an unmet Smithing gate is never selected (smithing 20 -> silver-bar, steel stays locked)",
select_recipe(f2p_bars, 20, bar_prices, input_prices, nature) == "silver-bar")
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": "43 <= level < 55 and the SII window is chosen (selection.mode picks the highest unlocked recipe with positive margin; MUTUALLY EXCLUSIVE — never at 55+)", "to": "cast"},
{"from": "cast", "when": "no unlocked recipe has a positive margin this window — skip SII and stay on LA", "to": "cast"},
{"from": "cast", "when": "level >= 55 and session target > 55 — the HLA tier executes until the target", "to": "cast"},
{"from": "cast", "when": "magic level >= session target — completion at the configured target", "to": "complete"}]
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) == [])
# --- governor comment 10477 regression: mutually exclusive tier boundaries ---
# The old SII window '43-55' overlapped HLA at exactly Magic 55 for
# targets above 55. selectable_tier() must prove exactly ONE tier at
# every (level, target) — completion first, HLA only >= 55 with a
# higher target, SII only 43 <= Magic < 55.
check("tier boundary 10477: Magic 55 with target 56 selects ONLY HLA (not SII/LA/complete)",
selectable_tier(55, 56) == "hla")
check("tier boundary 10477: Magic 55 with target 56 is never SII (the old overlap)",
selectable_tier(55, 56) != "sii")
check("tier boundary 10477: Magic 54 with target 56 still selects SII (below 55)",
selectable_tier(54, 56) == "sii")
check("tier boundary 10477: Magic 54 with SII skipped selects LA (SII-skip path)",
selectable_tier(54, 55, sii_skipped=True) == "la")
check("tier boundary 10477: Magic 55 with target 55 COMPLETES first (target wins)",
selectable_tier(55, 55) == "complete")
check("tier boundary 10477: Magic 56 with target 55 completes too",
selectable_tier(56, 55) == "complete")
check("tier boundary 10477: Magic 56 with target 60 selects HLA (above 55, target higher)",
selectable_tier(56, 60) == "hla")
check("tier boundary 10477: Magic 42 with target 55 selects LA (below 43)",
selectable_tier(42, 55) == "la")
check("tier boundary 10477: Magic 43 with target 55 selects SII (window lower edge)",
selectable_tier(43, 55) == "sii")
check("tier boundary 10477: exactly one tier for every level 40..60 at target 55",
all(len({selectable_tier(l, 55)}) == 1
for l in range(40, 61)))
check("tier boundary 10477: SII is NEVER selectable at Magic 55+ (the old overlap)",
all(selectable_tier(l, 56) != "sii" for l in range(55, 61)))
check("tier boundary 10477: HLA is NEVER selected below Magic 55",
all(selectable_tier(l, 60) != "hla" for l in range(40, 55)))
check("tier boundary 10477: at 55 the mutual-exclusion table is exact",
[selectable_tier(l, 55) for l in (42, 43, 54, 55, 56)] ==
["la", "sii", "sii", "complete", "complete"])
# tier_boundary_errors() gate: the old inclusive '43-55' spelling is rejected
entry_overlap = {"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"},
{"from": "cast", "when": "level >= 55 and session target > 55 — HLA", "to": "cast"}]}}
overlap_errs = tier_boundary_errors("bad.yaml", entry_overlap)
check("tier boundary 10477: SII window '43-55' overlapping HLA is REJECTED",
any("overlaps" in e or "43-55" in e for e in overlap_errs))
check("tier boundary 10477: mutually-exclusive spelling passes the boundary gate",
tier_boundary_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))
# --- rate_model fixtures (owner guidance 2026-08-28, bounce 4) ---
# The gate is OPT-IN: entries without rate_model are never treated as
# alchemy, and the rate is DERIVED from the entry's own declared
# lifecycle (cast action delay, run_control.break, recipes, startup).
cast_action = {"id": "a3", "step": "cast-alch", "verb": "use_item_on",
"params": {"spell": "Low Level Alchemy"},
"on_fail": "retry",
"anti_ban": {"delay": [1600, 3400]}}
base = {
"rate_model": {"anchor_tier": "la", "cast_ticks": 3,
"restock_trip_s": [60, 120], "session_horizon_s": 7200,
"startup_s": [300, 600],
"alch_carry_batch": 1000, "alch_restock_threshold": 100},
"actions": [cast_action],
"run_control": {"break": {"every": [400, 700], "for": [45, 120]}},
"economy": {"xp_per_hour": 23000}}
check("non-alchemy entry without rate_model passes (mining xp 52,500 untouched)",
rate_model_errors("mining.yaml", {"economy": {"xp_per_hour": 52500}}) == [])
check("LA entry declaring rate_model with the derived 23,000 xp/hr passes",
rate_model_errors("ok.yaml", dict(base)) == [])
check("loop-only LA 25,953 fails the derived realized-rate gate",
any("derived realized rate" in e
for e in rate_model_errors("bad.yaml", dict(base, **{"economy": {"xp_per_hour": 25953}}))))
check("pre-startup realized 24,500 fails too (startup amortisation is part of the model)",
any("derived realized rate" in e
for e in rate_model_errors("bad.yaml", dict(base, **{"economy": {"xp_per_hour": 24500}}))))
check("8-14-cast-break regime 9,455 fails the derived rate gate",
any("derived realized rate" in e
for e in rate_model_errors("bad.yaml", dict(base, **{"economy": {"xp_per_hour": 9455}}))))
mutated = copy.deepcopy(base)
mutated["run_control"]["break"]["for"] = [3600, 3600] # one-hour break per cast
check("cadence mutation caught: one-hour break per cast fails the derived rate gate",
any("derived realized rate" in e
for e in rate_model_errors("bad.yaml", mutated)))
# SII: per-recipe capacity (floor(27 / total input qty)) + derived rates
def recipe(rid, bar, bar_id, lvl, ti, inputs, cap, cph, xp):
return {"bar": bar, "bar_item_id": bar_id, "smithing_level": lvl,
"target_input": ti,
"inputs": [{"item": it, "item_id": iid, "qty": q, "source": "ge"}
for it, iid, q in inputs],
"rates": {"capacity": cap,
"casts_per_hour": {"min": cph[0], "mid": cph[1], "max": cph[2]},
"magic_xp_per_hour": {"min": xp[0], "mid": xp[1], "max": xp[2]}}}
steel_recipe = recipe("steel-bar", "Steel bar", 2353, 30, "Iron ore",
[("Iron ore", 440, 1), ("Coal", 453, 2)],
9, (355.3, 249.1, 191.2), (18833, 13204, 10131))
bronze_recipe = recipe("bronze-bar", "Bronze bar", 2349, 1, "Copper ore",
[("Copper ore", 436, 1), ("Tin ore", 438, 1)],
13, (445.6, 316.5, 244.4), (23617, 16776, 12952))
runite_recipe = recipe("runite-bar", "Runite bar", 2363, 85, "Runite ore",
[("Runite ore", 451, 1), ("Coal", 453, 8)],
3, (153.4, 104.5, 79.1), (8131, 5538, 4194))
sii_base = dict(base)
sii_base["rate_model"] = {"anchor_tier": "sii", "anchor_recipe": "steel-bar",
"cast_ticks": 3, "restock_trip_s": [60, 120],
"session_horizon_s": 7200, "startup_s": [300, 600]}
sii_base["economy"] = {"xp_per_hour": 12400}
sii_base["recipes"] = {"steel-bar": steel_recipe, "bronze-bar": bronze_recipe,
"runite-bar": runite_recipe}
check("SII entry passes: capacities floor(27/qty) = steel 9, bronze 13, runite 3, all rate pins derived",
rate_model_errors("ok.yaml", dict(sii_base)) == [])
check("steel anchor xp 12,400 (249.1 x 53 x startup 0.9375) passes the anchor gate",
rate_model_errors("ok2.yaml", dict(sii_base)) == [])
check("old 24-pack SII rate 27,500 fails the derived per-recipe rate",
any("derived realized rate" in e
for e in rate_model_errors("bad.yaml", dict(sii_base, **{"economy": {"xp_per_hour": 27500}}))))
bad_cap = copy.deepcopy(sii_base)
bad_cap["recipes"]["steel-bar"]["rates"]["capacity"] = 8
check("capacity pin mismatch (8 vs floor(27/3)=9) is rejected",
any("rates.capacity 8" in e for e in rate_model_errors("bad.yaml", bad_cap)))
bad_xp = copy.deepcopy(sii_base)
bad_xp["recipes"]["steel-bar"]["rates"]["magic_xp_per_hour"]["mid"] = 20000
check("recipe magic_xp pin far from the derived 13,204 is rejected",
any("magic_xp_per_hour.mid" in e for e in rate_model_errors("bad.yaml", bad_xp)))
oversized = copy.deepcopy(sii_base)
oversized["recipes"]["oversized-bar"] = recipe("oversized-bar", "Oversized bar", 9999, 90,
"Iron ore", [("Iron ore", 440, 28)],
1, (1, 1, 1), (53, 53, 53))
check("recipe needing more input slots than the inventory (qty 28 > 27) is rejected",
any("cannot fit the inventory" in e
for e in rate_model_errors("bad.yaml", oversized)))
# --- governor comment 10394 regression fixtures (bounce-fix 6) ---
# Finding 3: Magic target boundary — the entry may stop at the declared
# target (55 = the LA/SII -> HLA handoff) but must CONTINUE below it.
check("target boundary: level 54 with target 55 is a CONTINUE (not complete)",
target_reached(54, 55) is False)
check("target boundary: level 55 with target 55 HANDS OFF (complete)",
target_reached(55, 55) is True)
check("target boundary: level 56 with target 55 completes too",
target_reached(56, 55) is True)
# Finding 2: sign-corrected capital-loss stop — a loss is when assets
# FALL; the old reversed sign (total_assets - initial_capital) can never
# exceed the reserve on losses.
check("loss stop: assets 800,000 below initial capital (800,000 > 684,400 reserve) FIRES",
stop_loss_exceeded(1_300_000, 500_000, 684_400) is True)
check("loss stop: assets 604,400 below initial capital (604,400 < reserve) does NOT fire",
stop_loss_exceeded(1_300_000, 695_600, 684_400) is False)
check("loss stop reversed-sign trap: a PROFIT of 100,000 must NOT fire the loss stop",
stop_loss_exceeded(1_200_000, 1_300_000, 684_400) is False)
check("loss stop: crossing the reserve boundary exactly (loss == reserve) does NOT fire (strict >)",
stop_loss_exceeded(1_300_000, 615_600, 684_400) is False)
# Finding 1: build-loadout proof — carried counts must equal the carried
# amounts before the first cast; a shortfall is not verified.
required = {"Gold necklace": 1000, "Nature rune": 1000}
check("loadout verified at exactly 1,000/1,000 (proven before the first cast)",
loadout_verified({"Gold necklace": 1000, "Nature rune": 1000}, required) is True)
check("loadout NOT verified at 999/1,000 (shortfall routes to bank_supplies)",
loadout_verified({"Gold necklace": 999, "Nature rune": 1000}, required) is False)
check("loadout NOT verified when natures are missing entirely",
loadout_verified({"Gold necklace": 1000}, required) is False)
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 += tier_boundary_errors(rel, data)
errors += rate_model_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())