pbwraith/osrs-codex
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
main
osrs-codex/tools/validate.py
Claude 0c2e38c998 Merge origin/main 5ca8324b into issue-3; reconcile validator gates (governor 10637 / owner guidance 2026-08-29) 
Magic/Alchemy PR #10 landed on main; this integrates it into the woodcutting
branch and resolves the predicted tools/validate.py conflict:

- union of all validator gates: magic/alchemy (recipe/rate/tier/loadout,
  supply_limit_conflicts, document_error + _end_to_end_bad_mapping),
  mining (pickaxe suite), GE-flip lifecycle (floor-tax + scheduler +
  conservation + slot-phase + buy-capacity + slot-reuse), and the
  woodcutting gates (canonical axe table, per-mode ladders, per-selected-
  row 27/28 dynamic rates, drop-GP-zero, prepare/failed/drop-gate wiring).
- governor 10637: removed the run_flip_selftest alias duplication — the
  combined GE/Woodcutting suite no longer runs twice. Each suite now runs
  exactly once per invocation: run_selftest (cross-field/alchemy),
  run_flip_selftest (GE lifecycle), run_mining_selftest, and the new
  run_woodcutting_selftest. Truthful unique selftest count: 144 PASS.
- tree scan: 6 entries, 0 errors, 0 warnings; git diff --check clean.
2026-08-29 15:44:34 -05:00

3126 lines
170 KiB
Python
Raw Permalink Blame History

#!/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).
- Mining model (pickaxe gates + run-mode contract) — STRUCTURED
checks on the declared `pickaxe_tiers` / `pickaxe_examples` /
`mining_modes` / `mine_sites` blocks (owner guidance 2026-08-28):
* pickaxe tier values must match the canonical F2P table exactly
(Mining = usability gate, Attack = equip gate only, roll ticks,
and the explicit rank tie-break for equal Mining gates — Mining 1
selects iron over bronze, equipped at Attack 1);
* every pickaxe_examples row is recomputed (select highest tier
whose mining_level <= mining, tie-break by rank; equipped = attack
>= that tier's attack_level; expected_slots = 28 equipped / 27
carried) and must match the declared expectation — the pinned
1-Attack/41-Mining account must select and CARRY rune and the
1-Attack/1-Mining account must select and EQUIP iron;
* both run modes must be declared with the correct inventory
policy (fast-xp = drop, banked-income = bank) and only
reference declared, selectable sites whose policy matches;
* a selectable banked site with respawn >= 120 s MUST model
world hopping (world_hop: true AND a hop_world action in the
entry) — long-respawn sites are not executable otherwise;
* a selectable site whose access is Wilderness is rejected
(PK hazard cannot be hand-waved);
* modeled_ores_per_hour must not exceed the cadence ceiling
3600 / (rolls_per_ore x 2.7s rune delay midpoint);
* the entry's economy.xp_per_hour (fast-xp headline) must sit
within 2% of the fastest declared fast-xp site's modelled
ores/hr x its ore xp.
- 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.
- GE-flip lifecycle (flipping/merchanting entries): the 4h buy-limit
ledger (limit_used / limit_resets_at) is DISTINCT from offer
expiry — a limit-window reset never cancels an offer and cancelling
never renews the limit; every slot is an independent asynchronous
sub-state machine (canonical set: free / buy-open /
buy-filled-awaiting-collect / buy-cancelled-awaiting-collect /
ready-to-sell / sell-open / sell-filled-awaiting-collect /
sell-cancelled-awaiting-collect / ready-to-requote / closed) driven
by a global scheduler pass that never touches unrelated slots;
ge_cancel moves filled items, refunded CASH, unsold units and
proceeds to the GE collection box for a separate ge_collect;
per-slot conservation is SIDE-AWARE (buy: quoted units =
filled + still_open + cancelled_unfilled and acquired items =
filled ONLY — the unfilled/cancelled quantity is cash, never
refunded units; sell: quoted = sold + still_open + returned_unsold)
with phase gates (a full buy cannot sell before collection, a full
sell cannot close before proceeds collection);
closed has ONE outgoing transition — the archival release
``closed -> free`` — so a completed cycle's physical GE slot
returns to free for the next flip instead of stranding forever
(governor 10471); the two-rotation reuse regression proves a slot
completes buy/collect/sell/collect, archives the first ledger, and
places a second buy in the same slot without loss, duplication, or
mutation of the archived record;
economy.ceiling_gp_per_hour is the TYPED economic ceiling and
0 <= gp_per_hour_low <= gp_per_hour_high <= ceiling.
- Woodcutting entries (woodcutting tag, owner 10288 + governor
10398): canonical F2P axe table + equip-vs-carry rule (41 WC to
USE / 40 Attack to EQUIP, a 1-Attack/41-WC account carries and
uses a rune axe), exactly three deterministic modes with
per-SELECTED-row rate derivation (27/28-slot capacity from
axe_equipped, committed woodcutting_fixtures recomputed by this
validator), drop-mode realized GP zero, and state-machine wiring
(executable prepare -> walk_to_trees, no axe -> failed, drop
transition gated to the afk-xp drop mode).
- 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; plus per-slot scheduler
dispatch, conservation, buy-limit and nesting fixtures, the
Mining pickaxe suite, the Woodcutting axe/mode/fixture/wiring
suite, and the alchemy rate-model/tier/loadout suite — each
suite runs exactly once per invocation (no aliases or
duplicated executions).
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 full committed regression fixture suites only (no tree scan)
"""
import copy
import datetime
import json
import math
import os
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
# Canonical F2P pickaxe table (OSRS wiki pickaxe pages, verified 2026-08-28):
# mining_level = which pickaxe can be USED; attack_level = only whether it can
# be EQUIPPED; roll_ticks = ore roll interval (ticks x 0.6 s);
# rank = explicit tier order used as the SELECTION tie-break for equal
# Mining gates (bronze and iron both open at Mining 1 -> Mining 1 = iron,
# per the owner decision on governor review 10476).
CANONICAL_PICKAXES = {
"bronze-pickaxe": {"mining_level": 1, "attack_level": 1, "roll_ticks": 8, "rank": 1},
"iron-pickaxe": {"mining_level": 1, "attack_level": 1, "roll_ticks": 7, "rank": 2},
"steel-pickaxe": {"mining_level": 6, "attack_level": 5, "roll_ticks": 6, "rank": 3},
"mithril-pickaxe": {"mining_level": 21, "attack_level": 20, "roll_ticks": 5, "rank": 4},
"adamant-pickaxe": {"mining_level": 31, "attack_level": 30, "roll_ticks": 4, "rank": 5},
"rune-pickaxe": {"mining_level": 41, "attack_level": 40, "roll_ticks": 3, "rank": 6},
}
# a2 anti_ban delay midpoint [1800, 3600] ms — the paced roll-attempt cadence
# the rate model derives from (rune anchor; see mining-f2p.yaml economy).
RUNE_DELAY_MID_S = 2.7
XP_TOLERANCE = 0.02
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 select_pickaxe(mining_level, tiers):
"""Highest eligible tier by the two-gate rule: among tiers whose
mining_level gate the account meets, the highest RANK wins (rank is
the explicit tie-break for equal Mining gates — bronze and iron both
open at Mining 1, so Mining 1 deterministically selects iron)."""
best, best_key = None, None
for tid, tdef in (tiers or {}).items():
if not isinstance(tdef, dict):
continue
gate = tdef.get("mining_level")
if not isinstance(gate, int) or gate > mining_level:
continue
rank = tdef.get("rank")
key = (rank if isinstance(rank, int) else gate, gate)
if best_key is None or key > best_key:
best, best_key = tid, key
return best
def select_fast_site(mining_level, modes, sites):
"""Deterministic fast-xp site selection (mining_selections recompute):
highest ore XP among eligible selectable fast-xp sites whose
mining_level <= level; tie-break lexicographic site id (copper/tin share
an XP+level tie). Returns the site id or None."""
best_xp, cands = -1, []
for sid in (modes.get("fast-xp", {}).get("sites", []) or []):
sdef = sites.get(sid)
if not isinstance(sdef, dict) or sdef.get("selectable") is not True:
continue
ml, xp = sdef.get("mining_level"), sdef.get("xp")
if isinstance(ml, int) and isinstance(xp, (int, float)) and ml <= mining_level:
if xp > best_xp:
best_xp, cands = xp, [sid]
elif xp == best_xp:
cands.append(sid)
return sorted(cands)[0] if cands else None
def mining_model_errors(rel, data):
"""STRUCTURED mining-model gates (owner guidance 2026-08-28; replaces the
earlier phrase-search checks, which could pass on comments alone)."""
errors = []
tiers = data.get("pickaxe_tiers")
if tiers is None:
return errors # non-mining entry — not gated
if not isinstance(tiers, dict):
return [f"{rel}: 'pickaxe_tiers' must be a mapping of tier id -> gates"]
# --- 1. canonical tier values (exact) ---
for tid, canon in CANONICAL_PICKAXES.items():
got = tiers.get(tid)
if not isinstance(got, dict):
errors.append(
f"{rel}: pickaxe tier '{tid}' missing from pickaxe_tiers "
f"(canonical F2P table: mining {canon['mining_level']} use / "
f"attack {canon['attack_level']} equip / {canon['roll_ticks']} ticks)")
continue
for key, want in canon.items():
if got.get(key) != want:
errors.append(
f"{rel}: pickaxe tier '{tid}' {key}={got.get(key)}; canonical "
f"value is {want} (Mining = usability gate, Attack = equip "
f"gate ONLY, roll_ticks = ore roll interval)")
for tid in tiers:
if tid not in CANONICAL_PICKAXES:
errors.append(f"{rel}: pickaxe_tiers declares unknown tier '{tid}'")
# --- 2. examples recomputed from the DECLARED tiers (two-gate proof) ---
examples = data.get("pickaxe_examples")
if not isinstance(examples, list) or not examples:
errors.append(f"{rel}: pickaxe_examples must declare at least one account "
f"example for the two-gate selection proof")
else:
has_141 = has_11 = False
for ex in examples:
if not isinstance(ex, dict):
continue
atk, mng = ex.get("attack"), ex.get("mining")
if atk == 1 and mng == 41:
has_141 = True
if atk == 1 and mng == 1:
has_11 = True
chosen = select_pickaxe(mng, tiers)
if chosen is None:
errors.append(f"{rel}: example '{ex.get('account')}' (Mining {mng}) "
f"has no eligible pickaxe tier")
continue
tdef = tiers.get(chosen) or {}
should_equip = atk >= tdef.get("attack_level", 10**9)
if ex.get("expected_tier") != chosen:
errors.append(
f"{rel}: example '{ex.get('account')}' (Attack {atk}/Mining "
f"{mng}) expects tier '{ex.get('expected_tier')}' but the "
f"Mining-use gate selects '{chosen}'")
if ex.get("expected_equipped") != should_equip:
errors.append(
f"{rel}: example '{ex.get('account')}' (Attack {atk}/Mining "
f"{mng}) expects equipped={ex.get('expected_equipped')} but "
f"the Attack-equip gate on '{chosen}' says equipped={should_equip}")
# capacity proof: equipped pickaxe -> 28 ore slots, carried -> 27
want_slots = 28 if should_equip else 27
if ex.get("expected_slots") is None:
errors.append(
f"{rel}: example '{ex.get('account')}' must declare "
f"expected_slots ({want_slots} for tier '{chosen}' "
f"equipped={should_equip})")
elif ex.get("expected_slots") != want_slots:
errors.append(
f"{rel}: example '{ex.get('account')}' expects "
f"expected_slots={ex.get('expected_slots')} but the "
f"declared pickaxe '{chosen}' equipped={should_equip} "
f"gives {want_slots} ore slots (28 equipped / 27 carried)")
if not has_141:
errors.append(
f"{rel}: pickaxe_examples must include the owner-mandated "
f"1-Attack/41-Mining account (selects, carries, and uses a rune "
f"pickaxe — equipped=false)")
if not has_11:
errors.append(
f"{rel}: pickaxe_examples must include the 1-Attack/1-Mining "
f"account (owner decision on governor review 10476: bronze and "
f"iron share the Mining-1 gate, the rank tie-break selects "
f"IRON, and Attack 1 equips it — equipped=true, 28 ore slots)")
# --- 3. run-mode contract ---
modes = data.get("mining_modes")
sites = data.get("mine_sites")
if not isinstance(modes, dict) or not isinstance(sites, dict):
errors.append(f"{rel}: a mining entry must declare both 'mining_modes' "
f"and 'mine_sites' blocks")
return errors
if modes.get("fast-xp", {}).get("inventory_policy") != "drop":
errors.append(f"{rel}: mining_modes.fast-xp.inventory_policy must be "
f"'drop' (powermine drops every load — never banks)")
if modes.get("banked-income", {}).get("inventory_policy") != "bank":
errors.append(f"{rel}: mining_modes.banked-income.inventory_policy must "
f"be 'bank'")
def site_errors(site_id, where, want_policy):
out = []
sdef = sites.get(site_id)
if not isinstance(sdef, dict):
out.append(f"{rel}: {where} references '{site_id}' which is not "
f"declared in mine_sites")
return out
if sdef.get("selectable") is not True:
out.append(f"{rel}: {where} references '{site_id}' which is marked "
f"selectable: false (not an eligible target)")
if sdef.get("inventory_policy") != want_policy:
out.append(f"{rel}: {where} site '{site_id}' has inventory_policy "
f"'{sdef.get('inventory_policy')}' but the mode requires "
f"'{want_policy}'")
if sdef.get("selectable") is True and any(
not str(sdef.get(k, "")).strip() for k in ("access", "safety", "route")):
out.append(f"{rel}: selectable site '{site_id}' must carry access, "
f"safety and route descriptions")
return out
for sid in modes.get("fast-xp", {}).get("sites", []) or []:
errors += site_errors(sid, "mining_modes.fast-xp.sites", "drop")
for sid in modes.get("banked-income", {}).get("targets", []) or []:
errors += site_errors(sid, "mining_modes.banked-income.targets", "bank")
# --- 4. per-site execitability: long respawn, wilderness, rate ceiling ---
verbs = [a.get("verb") for a in data.get("actions", [])]
for sid, sdef in sites.items():
if not isinstance(sdef, dict):
continue
if sdef.get("selectable") is True:
access = str(sdef.get("access", "")).lower()
if "wilderness" in access:
errors.append(f"{rel}: selectable site '{sid}' is in the "
f"Wilderness (PK hazard) — it must be selectable: "
f"false, not silently exposed")
respawn = sdef.get("respawn_seconds")
if isinstance(respawn, (int, float)) and respawn >= 120:
if sdef.get("world_hop") is not True:
errors.append(f"{rel}: selectable site '{sid}' has a "
f"{respawn}s respawn — long-respawn sites need "
f"world_hop: true (the wiki's 120+ ores/hr "
f"assumes hopping worlds); without hopping the "
f"ceiling is ~30 ores/hr")
elif "hop_world" not in verbs:
errors.append(f"{rel}: site '{sid}' declares world_hop: true "
f"but the entry has no hop_world action — "
f"world hopping must be executable")
rolls = sdef.get("rolls_per_ore")
modeled = sdef.get("modeled_ores_per_hour")
if isinstance(rolls, (int, float)) and isinstance(modeled, (int, float)):
ceiling = 3600.0 / (rolls * RUNE_DELAY_MID_S)
if modeled > ceiling * (1 + 1e-9):
errors.append(
f"{rel}: site '{sid}' models {modeled} ores/hr but the "
f"cadence ceiling at rolls_per_ore {rolls} is {ceiling:.0f} "
f"(3600 / (rolls_per_ore x 2.7 s)) — the rate exceeds what "
f"the declared click policy can produce")
# --- 4b. GE tool purchase must be executable end to end ---
gev = set(v for v in verbs if v in {
"ge_open", "ge_buy", "ge_offers_done", "ge_collect", "ge_sell"})
if "ge_buy" in gev:
for must, why in (("ge_open", "open the exchange before placing an offer"),
("ge_offers_done", "wait one BOUNDED fill window"),
("ge_collect", "collect the item / refund the gp on timeout")):
if must not in gev:
errors.append(
f"{rel}: ge_buy must be part of an executable GE loop — "
f"missing '{must}' ({why}); a resting offer must never be "
f"retried indefinitely")
elif gev:
if "ge_open" not in gev:
errors.append(
f"{rel}: GE verbs {sorted(gev)} require a ge_open action — "
f"the GE flow must start by opening the exchange")
# --- 4c. below-15 copper/tin path must be structurally arming-able ---
min_level = (data.get("accounts") or {}).get("min_level")
fx_levels = []
for sid in (modes.get("fast-xp", {}).get("sites", []) or []):
sdef = sites.get(sid)
if isinstance(sdef, dict) and sdef.get("selectable") is True:
ml = sdef.get("mining_level")
if isinstance(ml, int):
fx_levels.append(ml)
if isinstance(min_level, int) and min_level < 15:
if not fx_levels:
errors.append(
f"{rel}: min_level {min_level} (<15) but fast-xp has no "
f"structured selectable site")
else:
fmin = min(fx_levels)
if fmin > 14:
errors.append(
f"{rel}: fast-xp sites all require Mining >= {fmin} but "
f"the entry claims a copper/tin below-15 path (min_level "
f"{min_level}) — add a structured low-level (<=14) "
f"fallback site or raise min_level")
if fmin > min_level:
errors.append(
f"{rel}: the lowest fast-xp site requires Mining {fmin} "
f"but min_level is {min_level} — a level-{min_level} "
f"account cannot arm a fast-xp run")
# --- 4d. mining_selections pins (deterministic, recomputed) ---
sel = data.get("mining_selections")
if not isinstance(sel, dict) or not isinstance(sel.get("fast_xp"), list) \
or not sel["fast_xp"]:
errors.append(
f"{rel}: mining entries must declare mining_selections.fast_xp "
f"(level -> expected_site pins, recomputed by the validator)")
else:
rows = {}
for r in sel["fast_xp"]:
lvl = r.get("level") if isinstance(r, dict) else None
if not isinstance(lvl, int) or lvl < 1:
errors.append(
f"{rel}: mining_selections.fast_xp rows need a positive "
f"integer 'level'")
continue
rows[lvl] = r
if isinstance(min_level, int) and min_level <= 14:
for lvl in (1, 14):
if lvl not in rows:
errors.append(
f"{rel}: mining_selections.fast_xp must pin level "
f"{lvl} (a below-15 account must have a structured, "
f"selectable fast-xp site)")
for lvl, r in rows.items():
got = select_fast_site(lvl, modes, sites)
if got is None:
errors.append(
f"{rel}: no eligible selectable fast-xp site at level "
f"{lvl} — the low-level (copper/tin) fallback is missing")
elif r.get("expected_site") != got:
errors.append(
f"{rel}: mining_selections level {lvl} expects site "
f"'{r.get('expected_site')}' but the selection rule "
f"picks '{got}'")
# --- 5. fast-xp headline xp rate consistent with the modelled sites ---
eco = data.get("economy")
xph = eco.get("xp_per_hour") if isinstance(eco, dict) else None
if isinstance(xph, (int, float)):
best = None
for sid in modes.get("fast-xp", {}).get("sites", []) or []:
sdef = sites.get(sid)
if not isinstance(sdef, dict):
continue
rate = sdef.get("modeled_ores_per_hour") or 0
xpv = (sdef.get("xp") or 0) * rate
if best is None or xpv > best:
best = xpv
if best:
if abs(xph - best) / best > XP_TOLERANCE:
errors.append(
f"{rel}: economy.xp_per_hour {xph} is not the fast-xp "
f"modelled policy rate (expected ~{best:.0f} = fastest "
f"fast-xp site's modelled ores/hr x ore xp; tolerance "
f"{XP_TOLERANCE:.0%})")
return errors
def _with_runite(sites):
"""Fixture helper: the `good` site set plus a non-selectable Wilderness
runite row (hyphenated id so the declared reference resolves)."""
out = dict(sites)
out["runite-wilderness"] = {
"ore": "Runite ore", "ore_id": 451, "mining_level": 85, "xp": 125,
"rock_count": 2, "respawn_seconds": 720, "rolls_per_ore": 3.0,
"modeled_ores_per_hour": 0, "world_hop": False, "selectable": False,
"inventory_policy": "bank", "access": "Wilderness (PK risk)",
"safety": "out of scope", "route": "not selectable"}
return out
def run_mining_selftest(verbose=False):
"""Committed regression fixtures for the structured mining gates."""
errors = []
def check(name, passed):
if verbose:
print(f"SELFTEST {'PASS' if passed else 'FAIL'} {name}")
if not passed:
errors.append(f"selftest: {name}")
canon = {k: dict(v) for k, v in CANONICAL_PICKAXES.items()} # deep copy — fixtures below must never mutate the module table
good_examples = [
{"account": "1-Attack/1-Mining", "attack": 1, "mining": 1,
"expected_tier": "iron-pickaxe", "expected_equipped": True,
"expected_slots": 28},
{"account": "1-Attack/41-Mining", "attack": 1, "mining": 41,
"expected_tier": "rune-pickaxe", "expected_equipped": False,
"expected_slots": 27},
{"account": "40-Attack/70-Mining", "attack": 40, "mining": 70,
"expected_tier": "rune-pickaxe", "expected_equipped": True,
"expected_slots": 28},
{"account": "5-Attack/30-Mining", "attack": 5, "mining": 30,
"expected_tier": "mithril-pickaxe", "expected_equipped": False,
"expected_slots": 27},
]
sites = {
"al-kharid-powermine": {"ore": "Iron ore", "ore_id": 440, "mining_level": 15,
"xp": 35, "rock_count": 3, "respawn_seconds": 5.4, "rolls_per_ore": 1.33,
"modeled_ores_per_hour": 803, "world_hop": False, "selectable": True,
"inventory_policy": "drop", "access": "combat 29+", "safety": "gate",
"route": "drop in place"},
"se-varrock-copper": {"ore": "Copper ore", "ore_id": 436, "mining_level": 1,
"xp": 17.5, "rock_count": 2, "respawn_seconds": 5.4, "rolls_per_ore": 1.6,
"modeled_ores_per_hour": 360, "world_hop": False, "selectable": True,
"inventory_policy": "drop", "access": "safe", "safety": "none",
"route": "drop in place"},
"se-varrock-tin": {"ore": "Tin ore", "ore_id": 438, "mining_level": 1,
"xp": 17.5, "rock_count": 2, "respawn_seconds": 5.4, "rolls_per_ore": 1.6,
"modeled_ores_per_hour": 360, "world_hop": False, "selectable": True,
"inventory_policy": "drop", "access": "safe", "safety": "none",
"route": "drop in place"},
"se-varrock-bank": {"ore": "Iron ore", "ore_id": 440, "mining_level": 15,
"xp": 35, "rock_count": 2, "respawn_seconds": 5.4, "rolls_per_ore": 1.33,
"modeled_ores_per_hour": 713, "world_hop": False, "selectable": True,
"inventory_policy": "bank", "access": "safe", "safety": "none",
"route": "east Varrock bank 30s"},
"al-kharid-south-adamantite": {"ore": "Adamantite ore", "ore_id": 449,
"mining_level": 70, "xp": 95, "rock_count": 2, "respawn_seconds": 240,
"rolls_per_ore": 2.0, "modeled_ores_per_hour": 120, "world_hop": True,
"selectable": True, "inventory_policy": "bank", "access": "south rocks",
"safety": "banked locally", "route": "bank 25s + hop_world"},
}
good = {"accounts": {"min_level": 1},
"pickaxe_tiers": canon, "pickaxe_examples": good_examples,
"mining_modes": {"fast-xp": {"inventory_policy": "drop",
"ore_policy": "iron 15+ / copper/tin under 15",
"sites": ["al-kharid-powermine",
"se-varrock-copper",
"se-varrock-tin"]},
"banked-income": {"inventory_policy": "bank",
"targets": ["se-varrock-bank",
"al-kharid-south-adamantite"],
"stop": "gp target"}},
"mining_selections": {"fast_xp": [
{"level": 1, "expected_site": "se-varrock-copper"},
{"level": 14, "expected_site": "se-varrock-copper"}]},
"mine_sites": sites,
"actions": [{"id": "a1", "step": "hop-world", "verb": "hop_world",
"on_fail": "retry"}],
"economy": {"xp_per_hour": 28100}}
check("canonical pickaxe tiers + recomputed examples pass",
mining_model_errors("ok.yaml", good) == [])
# pickaxe tier value mismatch
bad_tiers = dict(canon)
bad_tiers["rune-pickaxe"] = {"mining_level": 41, "attack_level": 45, "roll_ticks": 3}
errs = mining_model_errors("bad.yaml", {**good, "pickaxe_tiers": bad_tiers})
check("rune equip gate value 45 != canonical 40 is rejected",
any("rune-pickaxe" in e and "attack_level" in e for e in errs))
# 1-Attack/41-Mining must carry rune — declared expectation is enforced
bad_141 = {"account": "1-Attack/41-Mining", "attack": 1, "mining": 41,
"expected_tier": "rune-pickaxe", "expected_equipped": True,
"expected_slots": 27}
errs = mining_model_errors("bad2.yaml", {**good, "pickaxe_examples": [bad_141]})
check("1-Attack/41-Mining example expecting equipped=True is rejected (equip gate: Attack 1 < 40)",
any("equipped=True" in e and "equipped=False" in e for e in errs))
errs = mining_model_errors("bad3.yaml", {**good, "pickaxe_examples": [
{"account": "40-Attack/70-Mining", "attack": 40, "mining": 70,
"expected_tier": "rune-pickaxe", "expected_equipped": True,
"expected_slots": 28}]})
check("missing the 1-Attack/41-Mining example is rejected (owner-mandated)",
any("1-Attack/41-Mining" in e for e in errs))
# level-1 tie-break (governor review 10476, owner decision 2026-08-29):
# Mining 1 must select IRON (rank tie-break over bronze) and EQUIP it
# (attack gate 1) with 28 ore slots — the old bronze-canonical model is
# rejected on three independent axes
tie_bronze = {"account": "1-Attack/1-Mining", "attack": 1, "mining": 1,
"expected_tier": "bronze-pickaxe", "expected_equipped": True,
"expected_slots": 28}
errs = mining_model_errors("f18.yaml", {**good, "pickaxe_examples":
[x for x in good_examples if x["mining"] != 1] + [tie_bronze]})
check("Mining 1 expecting the bronze pickaxe is rejected (rank tie-break selects iron)",
any("'bronze-pickaxe'" in e and "selects 'iron-pickaxe'" in e for e in errs))
tie_carry = {"account": "1-Attack/1-Mining", "attack": 1, "mining": 1,
"expected_tier": "iron-pickaxe", "expected_equipped": False,
"expected_slots": 27}
errs = mining_model_errors("f19.yaml", {**good, "pickaxe_examples":
[x for x in good_examples if x["mining"] != 1] + [tie_carry]})
check("1-Attack/1-Mining expecting a CARRIED iron pickaxe is rejected (Attack 1 >= iron's attack gate 1 -> equipped)",
any("equipped=True" in e and "equipped=False" in e for e in errs) or
any("gives 28 ore slots" in e for e in errs))
tie_slots = {"account": "1-Attack/1-Mining", "attack": 1, "mining": 1,
"expected_tier": "iron-pickaxe", "expected_equipped": True,
"expected_slots": 27}
errs = mining_model_errors("f20.yaml", {**good, "pickaxe_examples":
[x for x in good_examples if x["mining"] != 1] + [tie_slots]})
check("1-Attack/1-Mining with 27 slots while equipped is rejected (equipped iron = 28 ore slots)",
any("28 equipped / 27 carried" in e for e in errs))
no_11 = {**good, "pickaxe_examples":
[x for x in good_examples if not (x["attack"] == 1 and x["mining"] == 1)]}
errs = mining_model_errors("f21.yaml", no_11)
check("missing the 1-Attack/1-Mining fixture is rejected (owner-mandated tie-break proof)",
any("1-Attack/1-Mining" in e for e in errs))
bad_tiers2 = {k: dict(v) for k, v in canon.items()}
del bad_tiers2["iron-pickaxe"]["rank"]
errs = mining_model_errors("f22.yaml", {**good, "pickaxe_tiers": bad_tiers2})
check("a pickaxe tier missing the rank field is rejected (tie-break must be explicit)",
any("iron-pickaxe" in e and "rank" in e for e in errs))
# mode contract
errs = mining_model_errors("bad4.yaml", {**good, "mining_modes": {
"fast-xp": {"inventory_policy": "bank",
"ore_policy": "iron", "sites": ["al-kharid-powermine"]},
"banked-income": {"inventory_policy": "bank", "targets": ["se-varrock-bank"],
"stop": "gp target"}}})
check("fast-xp mode with inventory_policy bank is rejected",
any("fast-xp.inventory_policy" in e for e in errs))
errs = mining_model_errors("bad5.yaml", {**good, "mining_modes": {
"fast-xp": {"inventory_policy": "drop", "ore_policy": "iron",
"sites": ["al-kharid-powermine"]},
"banked-income": {"inventory_policy": "bank",
"targets": ["runite-wilderness"], "stop": "gp target"}},
"mine_sites": _with_runite(sites)})
check("banked-income targeting a selectable:false site is rejected",
any("selectable: false" in e for e in errs))
errs = mining_model_errors("bad6.yaml", {**good, "mining_modes": {
"fast-xp": {"inventory_policy": "drop", "ore_policy": "iron",
"sites": ["nope-mine"]},
"banked-income": {"inventory_policy": "bank", "targets": ["se-varrock-bank"],
"stop": "gp target"}}})
check("mode referencing an undeclared site is rejected",
any("not declared in mine_sites" in e for e in errs))
# long-respawn / wilderness / hop executability
no_hop = {**good, "mine_sites": dict(sites, al_kharid_south_adamantite={
**sites["al-kharid-south-adamantite"], "world_hop": False})}
errs = mining_model_errors("bad7.yaml", no_hop)
check("selectable 240s-respawn site without world_hop is rejected",
any("world_hop: true" in e for e in errs))
hop_no_action = {**good, "actions": [{"id": "a1", "step": "mine-stuff",
"verb": "click_object", "on_fail": "retry"}]}
errs = mining_model_errors("bad8.yaml", hop_no_action)
check("world_hop site without a hop_world action is rejected",
any("hop_world action" in e for e in errs))
wild = {**good, "mine_sites": dict(sites, wildy={"ore": "Runite ore",
"ore_id": 451, "mining_level": 85, "xp": 125, "rock_count": 2,
"respawn_seconds": 720, "rolls_per_ore": 3.0, "modeled_ores_per_hour": 0,
"world_hop": False, "selectable": True, "inventory_policy": "bank",
"access": "Wilderness level 46 (PK risk)", "safety": "none",
"route": "wilderness"})}
errs = mining_model_errors("bad9.yaml", wild)
check("selectable Wilderness site is rejected",
any("Wilderness" in e and "selectable" in e for e in errs))
# rate ceiling
over = {**good, "mine_sites": dict(sites, al_kharid_south_adamantite={
**sites["al-kharid-south-adamantite"], "modeled_ores_per_hour": 839})}
errs = mining_model_errors("bad10.yaml", over)
check("adamantite 839 ores/hr exceeds the 2.0-roll cadence ceiling (~666) and is rejected",
any("cadence ceiling" in e for e in errs))
# headline xp consistency
errs = mining_model_errors("bad11.yaml", {**good, "economy": {"xp_per_hour": 32400}})
check("xp_per_hour 32,400 != modelled fast-xp rate (~28,105) is rejected",
any("xp_per_hour" in e for e in errs))
errs = mining_model_errors("ok2.yaml", {**good, "economy": {"xp_per_hour": 28000}})
check("xp_per_hour 28,000 within 2% of the modelled fast-xp rate 28,105 passes",
errs == [])
# below-15 fallback + mining_selections pins (governor bounce 5)
no_fallback = {**good, "mining_modes": {
"fast-xp": {"inventory_policy": "drop", "ore_policy": "iron 15+ only",
"sites": ["al-kharid-powermine"]},
"banked-income": {"inventory_policy": "bank", "targets": ["se-varrock-bank",
"al-kharid-south-adamantite"], "stop": "gp target"}}}
errs = mining_model_errors("f14.yaml", no_fallback)
check("min_level 1 with iron-only fast-xp sites (no copper/tin fallback) is rejected",
any("below-15" in e for e in errs))
wrongpin = {**good, "mining_selections": {"fast_xp": [
{"level": 1, "expected_site": "se-varrock-tin"},
{"level": 14, "expected_site": "se-varrock-copper"}]}}
errs = mining_model_errors("f15.yaml", wrongpin)
check("mining_selections level-1 expected site mismatching the recomputed selection is rejected",
any("selection rule picks" in e for e in errs))
no_pin14 = {**good, "mining_selections": {"fast_xp": [
{"level": 1, "expected_site": "se-varrock-copper"}]}}
errs = mining_model_errors("f16.yaml", no_pin14)
check("mining_selections missing the level-14 pin is rejected when min_level < 15",
any("pin level 14" in e for e in errs))
ge_no_open = {**good, "actions": [
{"id": "a1", "step": "buy-pick", "verb": "ge_buy", "on_fail": "retry"},
{"id": "a2", "step": "hop-world", "verb": "hop_world", "on_fail": "retry"}]}
errs = mining_model_errors("f17.yaml", ge_no_open)
check("ge_buy without ge_open / ge_offers_done / ge_collect is rejected",
any("ge_open" in e for e in errs))
return errors
MARGIN_TAGS = {"flipping", "merchanting"}
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 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 alch_worked_example_regression(rel, data, raw_text):
"""One-cast model worked-example regression (owner guidance 2026-08-28,
PR #8 repair, merged via origin/main during branch reconciliation):
entries with the
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. 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
# --- GE-flip lifecycle gates (owner guidance 2026-08-28, governor 10323 + 10390) ---
# Canonical per-slot sub-states: each GE slot is an INDEPENDENT asynchronous
# machine driven by a global scheduler pass; a pass never closes or advances
# a slot it did not touch. A COMPLETED offer leaves its assets in the GE
# collection box until ge_collect: fully filled buys and sells each get an
# explicit *-filled-awaiting-collect state (governor 10390 finding 2).
SLOT_SUB_STATES = [
"free", # no offer, nothing held
"buy-open", # resting buy offer (partial fills keep resting)
"buy-filled-awaiting-collect", # buy FULLY filled — units sit in the GE collection box until ge_collect
"buy-cancelled-awaiting-collect", # aborted buy: bought units + refunded CASH in the box
"ready-to-sell", # units in hand (ge_collect'ed), sell edge positive
"sell-open", # resting sell offer (partial sells tracked)
"sell-filled-awaiting-collect", # sell FULLY filled — proceeds sit in the box until ge_collect
"sell-cancelled-awaiting-collect", # aborted sell: proceeds + unsold units in the box
"ready-to-requote", # collected units/refunds; requote or abandon
"closed", # completed cycle: ledger reconciled + ARCHIVED; the physical slot is released to free via the closed -> free archival transition (governor 10471)
]
# Dispatch table: the pure action a slot in a given sub-state needs this pass.
SLOT_DISPATCH = {
"free": "quote",
"buy-open": None, # resting; fill/abort decided by policies
"buy-filled-awaiting-collect": "collect",
"buy-cancelled-awaiting-collect": "collect",
"ready-to-sell": "sell",
"sell-open": None, # resting; fill/abort decided by policies
"sell-filled-awaiting-collect": "collect",
"sell-cancelled-awaiting-collect": "collect",
"ready-to-requote": "requote",
"closed": "release", # archival release: closed -> free (governor 10471)
}
def scheduler_dispatch(sub_state):
"""The pending action for one slot in one scheduler pass, or None.
A None (resting) slot is left completely untouched — the pass must not
close, advance, or mutate it. This is the property the worked-example
regression proves. A 'closed' slot is NOT resting: its completed
cycle's ledger was archived at ge_collect time and the pass releases
the physical slot to free ('release' = the closed -> free archival
transition), so the slot can run another flip instead of stranding
forever (governor 10471). Resting states are buy-open / sell-open
only.
"""
return SLOT_DISPATCH.get(sub_state)
# Side-aware GE bucket names (governor 10390 finding 4).
# BUY: the player commits CASH; acquired items are `filled` ONLY — the
# unfilled/cancelled quantity is an unspent-CASH refund, never "refunded
# units". SELL: the player holds units; sold units yield proceeds.
BUY_UNIT_BUCKETS = ["filled", "still_open", "cancelled_unfilled"]
SELL_UNIT_BUCKETS = ["sold", "still_open", "returned_unsold"]
# Physically impossible co-occurrences per side (cancel is atomic: a live
# offer has no remainder once cancelled).
IMPOSSIBLE_UNIT_PAIRS = {
"buy": [("still_open", "cancelled_unfilled")],
"sell": [("still_open", "returned_unsold")],
}
def conservation_violations(record):
"""Side-aware per-slot GE conservation (governor 10390 finding 4).
record: {
'side': 'buy'|'sell', # REQUIRED — the ledger is
'quoted': {item: qty}, # side-aware
**unit buckets (side-specific): # {item: qty}
'cash_quoted': int, # BUY: cash committed
'spent'/'still_committed'/'refunded': int, # BUY cash partition
'proceeds_in_box'/'cash_collected': int, # SELL proceeds
'sub_state': str, # phase gate input
}
BUY invariants:
- unit partition: quoted == filled + still_open + cancelled_unfilled
per item;
- ACQUIRED ITEMS == filled ONLY: any sell-side unit bucket (sold /
returned_unsold) or a 'refunded' unit bucket on a buy record is a
violation — the unfilled/cancelled quantity is CASH (phantom
"refunded units" guard);
- cash partition: cash_quoted == spent + still_committed + refunded
(the unspent remainder of a buy is the cash refund);
- no item may be both still_open and cancelled_unfilled (cancel is
atomic).
SELL invariants:
- unit partition: quoted == sold + still_open + returned_unsold per
item;
- no item may be both still_open and returned_unsold (cancel is
atomic); sold + still_open IS legal (partial sell rest).
Shared: no negative quantities.
"""
out = []
side = record.get("side", "buy")
if side not in ("buy", "sell"):
out.append(f"record side {side!r} must be 'buy' or 'sell'")
return out
buckets = BUY_UNIT_BUCKETS if side == "buy" else SELL_UNIT_BUCKETS
foreign = {"filled", "still_open", "cancelled_unfilled", "sold",
"returned_unsold"} - set(buckets)
for b in sorted(foreign):
if record.get(b):
out.append(
f"{side} ledger uses {b} units — invalid for side {side}: "
f"buy-side unfilled/cancelled quantity is CASH "
f"(refunded), never {b} units; sell-side units partition "
f"into sold/still_open/returned_unsold")
if side == "buy":
# A unit map under a CASH field is the phantom-items guard: a buy's
# unfilled/cancelled quantity is an unspent-CASH refund, never
# "refunded units" (governor 10390 finding 4).
for c in ("cash_quoted", "spent", "still_committed", "refunded"):
if isinstance(record.get(c), dict):
out.append(
f"buy ledger carries {c} as units — the unfilled/"
f"cancelled quantity of a buy is CASH (refunded), "
f"never {c} units (phantom items)")
for b in buckets:
for item in record.get(b, {}):
q = record[b][item]
if q < 0:
out.append(f"item '{item}' bucket {b} is negative ({q})")
quoted = record.get("quoted", {})
for item, qq in quoted.items():
if qq < 0:
out.append(f"item '{item}' quoted is negative ({qq})")
total = sum(record.get(b, {}).get(item, 0) for b in buckets)
if total != qq:
out.append(
f"item '{item}' quoted {qq} != "
f"{' + '.join(buckets)} ({total})")
in_b = {b for b in buckets if item in record.get(b, {})}
for a, b in IMPOSSIBLE_UNIT_PAIRS[side]:
if (a in in_b and b in in_b
and record.get(a, {}).get(item)
and record.get(b, {}).get(item)):
out.append(
f"item '{item}' appears in both {a} and {b} — "
f"physically impossible (a cancelled offer has no "
f"live remainder)")
if side == "buy":
cash_fields = ["cash_quoted", "spent", "still_committed", "refunded"]
for c in cash_fields:
if isinstance(record.get(c), (int, float)) and record[c] < 0:
out.append(f"{c} is negative ({record[c]})")
cash_vals = {c: record.get(c) for c in cash_fields}
if all(isinstance(v, (int, float)) for v in cash_vals.values()):
if cash_vals["cash_quoted"] != (
cash_vals["spent"] + cash_vals["still_committed"]
+ cash_vals["refunded"]):
out.append(
f"buy cash quoted {cash_vals['cash_quoted']} != spent "
f"{cash_vals['spent']} + still_committed "
f"{cash_vals['still_committed']} + refunded "
f"{cash_vals['refunded']} — the unspent remainder of a "
f"buy is the CASH refund")
else:
for c in ("proceeds_in_box", "cash_collected"):
if record.get(c, 0) < 0:
out.append(f"{c} is negative ({record[c]})")
return out
def slot_phase_violations(record):
"""Phase gates (governor 10390 finding 4): assets may not be reused or
a slot closed before the GE collection box is withdrawn.
- 'buy-filled-awaiting-collect': filled must equal quoted (fully
filled) and NOTHING may have been sold yet — a full buy that sells
before ge_collect is a violation (owned units cannot leave the
collection box early).
- 'sell-filled-awaiting-collect': sold units require proceeds_in_box
> 0 — proceeds arrive in the box as units fill and only leave via
ge_collect.
- 'closed': nothing may remain uncollected — proceeds_in_box must be
0 and no cancelled/returned units may still be in the box (a full
sell may not close before proceeds collection).
"""
out = []
sub = record.get("sub_state")
side = record.get("side", "buy")
sold = sum((record.get("sold") or {}).values())
filled = sum((record.get("filled") or {}).values())
quoted = sum((record.get("quoted") or {}).values())
if side == "buy" and sub == "buy-filled-awaiting-collect":
if sold:
out.append(
"buy-filled-awaiting-collect slot has sold units — a full "
"buy cannot sell before ge_collect")
if filled != quoted:
out.append(
f"buy-filled-awaiting-collect slot is not fully filled "
f"(filled {filled} != quoted {quoted})")
if sub == "sell-filled-awaiting-collect" and sold and not record.get("proceeds_in_box"):
out.append(
"sell-filled-awaiting-collect slot has sold units but no "
"proceeds_in_box — proceeds arrive in the collection box on "
"fill and leave only via ge_collect")
if sub == "closed":
if record.get("proceeds_in_box"):
out.append(
"closed slot still has proceeds_in_box — a full sell may "
"not close before proceeds collection (ge_collect first)")
cancelled = sum((record.get("cancelled_unfilled") or {}).values())
returned = sum((record.get("returned_unsold") or {}).values())
if cancelled or returned:
out.append(
"closed slot still holds cancelled/returned units — "
"collect them before closing the slot")
return out
def buy_capacity_ok(limit_used, limit, qty_requested):
"""The 4h buy-limit ledger (governor 10323 finding 1): the limit resets
4h after the FIRST unit was bought; cancelling an offer does NOT renew
it, so a requote before the reset cannot buy beyond the remaining
capacity. Returns True when limit_used + qty_requested <= limit.
"""
return limit_used + qty_requested <= limit
# Deterministic ledger frames for ONE physical GE slot across TWO full
# flips (governor 10471): the fixture walks a complete buy/collect/
# sell/collect cycle, archives the first ledger at `closed`, releases the
# physical slot to `free`, and opens a SECOND buy in the SAME slot. Every
# frame must satisfy both conservation checks; the archived first-cycle
# record must survive rotation 2 untouched (no loss, duplication, or
# mutation). Prices: Rune arrowtips buy 223 / sell 244 / tax
# min(floor(244*0.02), 5M) = 4 / proceeds 2,440,000 (snapshot 2026-08-27,
# matches the worked example).
_CYCLE_ITEM = "Rune arrowtips"
_CYCLE_FRAMES = [
# (sub_state, ledger) — ledger is side-aware and conservation-clean.
("free", {"side": "buy", "quoted": {}, "filled": {}, "still_open": {},
"cancelled_unfilled": {}, "cash_quoted": 0, "spent": 0,
"still_committed": 0, "refunded": 0}),
("buy-open", {"side": "buy", "quoted": {_CYCLE_ITEM: 10_000},
"filled": {}, "still_open": {_CYCLE_ITEM: 10_000},
"cancelled_unfilled": {}, "cash_quoted": 2_230_000,
"spent": 0, "still_committed": 2_230_000, "refunded": 0}),
("buy-filled-awaiting-collect",
{"side": "buy", "quoted": {_CYCLE_ITEM: 10_000},
"filled": {_CYCLE_ITEM: 10_000}, "still_open": {},
"cancelled_unfilled": {}, "cash_quoted": 2_230_000,
"spent": 2_230_000, "still_committed": 0, "refunded": 0}),
("ready-to-sell",
{"side": "buy", "quoted": {_CYCLE_ITEM: 10_000},
"filled": {_CYCLE_ITEM: 10_000}, "still_open": {},
"cancelled_unfilled": {}, "cash_quoted": 2_230_000,
"spent": 2_230_000, "still_committed": 0, "refunded": 0}),
("sell-open", {"side": "sell", "quoted": {_CYCLE_ITEM: 10_000},
"sold": {}, "still_open": {_CYCLE_ITEM: 10_000},
"returned_unsold": {}, "proceeds_in_box": 0,
"cash_collected": 0}),
("sell-filled-awaiting-collect",
{"side": "sell", "quoted": {_CYCLE_ITEM: 10_000},
"sold": {_CYCLE_ITEM: 10_000}, "still_open": {},
"returned_unsold": {}, "proceeds_in_box": 2_440_000,
"cash_collected": 0}),
("closed", {"side": "sell", "quoted": {_CYCLE_ITEM: 10_000},
"sold": {_CYCLE_ITEM: 10_000}, "still_open": {},
"returned_unsold": {}, "proceeds_in_box": 0,
"cash_collected": 2_440_000}), # ARCHIVED here
("free", {"side": "buy", "quoted": {}, "filled": {}, "still_open": {},
"cancelled_unfilled": {}, "cash_quoted": 0, "spent": 0,
"still_committed": 0, "refunded": 0}), # released: physical slot reusable
# rotation 2: the SAME physical slot opens a second buy (ledger reset;
# the archived rotation-1 record above must remain untouched).
("buy-open", {"side": "buy", "quoted": {_CYCLE_ITEM: 10_000},
"filled": {}, "still_open": {_CYCLE_ITEM: 10_000},
"cancelled_unfilled": {}, "cash_quoted": 2_230_000,
"spent": 0, "still_committed": 2_230_000, "refunded": 0}),
]
# The canonical complete flip cycle: rotation 1 (buy/collect/sell/collect)
# plus the archival release and the start of rotation 2.
_CYCLE_PATH = [f[0] for f in _CYCLE_FRAMES]
# Index of the `closed` frame (archived ledger) and the frames around it.
_CLOSED_IDX = _CYCLE_PATH.index("closed")
def slot_reuse_violations(transitions):
"""Two-rotation slot-reuse regression (governor 10471).
Walks ``_CYCLE_PATH`` through the entry's ``slot_transitions`` and
validates every ledger frame against both conservation gates. A
``closed`` state with no outgoing transition (or one that does not
release to ``free``) strands the physical GE slot after one flip —
the exact defect this regression exists to forbid. Errors are
returned as a list of strings.
"""
out = []
for a, b in zip(_CYCLE_PATH, _CYCLE_PATH[1:]):
if b not in (transitions.get(a) or []):
out.append(
f"slot cycle: transition {a} -> {b} is missing — a completed "
f"flip must end by releasing the physical slot to free "
f"(archival release closed -> free), otherwise the slot "
f"performs one flip and then becomes permanently unusable")
for state, record in _CYCLE_FRAMES:
rec = dict(record, sub_state=state)
con = conservation_violations(rec)
if con:
out.append(f"slot cycle: conservation fails at '{state}': {con}")
ph = slot_phase_violations(rec)
if ph:
out.append(f"slot cycle: phase gate fails at '{state}': {ph}")
return out
def archived_ledger_preserved_after_second_buy():
"""The rotation-1 archived ledger survives rotation 2 intact.
``closed`` archives a deep copy of the completed cycle's ledger; when
the same physical slot is released to free and opens a second buy, the
archived record must be bit-for-bit unchanged (no loss, duplication,
or mutation) while the live ledger is a fresh empty record.
"""
archived = json.loads(json.dumps(_CYCLE_FRAMES[_CLOSED_IDX][1]))
archived_state = _CYCLE_FRAMES[_CLOSED_IDX][0]
live = {k: (dict(v) if isinstance(v, dict) else v)
for k, v in _CYCLE_FRAMES[-1][1].items()} # rotation-2 buy-open ledger
if archived != _CYCLE_FRAMES[_CLOSED_IDX][1]:
return ["slot cycle: archived rotation-1 ledger was MUTATED by rotation 2"]
if archived_state != "closed":
return [f"slot cycle: unexpected archived state {archived_state!r}"]
# The live ledger at rotation-2 buy-open must be a fresh record, not a
# copy of the archived one: quoted cash/units belong to rotation 2.
if live.get("cash_quoted") != 2_230_000 or live.get("proceeds_in_box"):
return ["slot cycle: rotation-2 live ledger is not a fresh record"]
return []
def ge_lifecycle_errors(rel, data):
"""GE-flip lifecycle gates (flipping/merchanting entries only):
- the 4h buy-limit ledger is distinct from offer expiry: the entry must
declare a limit_ledger + an independent quote_age_policy;
- the slot model is per-slot asynchronous: slot_state_machine must
declare exactly the canonical sub-states, valid transitions, a
scheduler_iteration, persisted_state, conservation and break_resume;
'closed' is NOT a sink: it must release the physical slot via the
archival transition closed -> free, and the committed two-rotation
reuse regression (slot_reuse_violations) must pass;
- rates: ceiling_gp_per_hour is the typed economic ceiling and
0 <= gp_per_hour_low <= gp_per_hour_high <= ceiling.
"""
errors = []
tags = {str(t).lower() for t in data.get("tags", [])}
if not tags & MARGIN_TAGS:
return errors
eco = data.get("economy") or {}
# --- typed ceiling vs realized bounds ---
ceil = eco.get("ceiling_gp_per_hour")
if not isinstance(ceil, (int, float)) or ceil <= 0:
errors.append(f"{rel}: flipping entries must declare "
f"economy.ceiling_gp_per_hour (the typed 4h buy-limit "
f"economic ceiling, distinct from realized rates)")
return errors
low, high = eco.get("gp_per_hour_low"), eco.get("gp_per_hour_high")
for name, val in (("gp_per_hour_low", low), ("gp_per_hour_high", high)):
if not isinstance(val, (int, float)) or val < 0:
errors.append(f"{rel}: {name} must be a non-negative realized "
f"scenario bound")
if isinstance(low, (int, float)) and isinstance(high, (int, float)) \
and isinstance(ceil, (int, float)):
if not (low <= high <= ceil):
errors.append(
f"{rel}: realized bounds must nest under the ceiling: "
f"0 <= gp_per_hour_low {low} <= gp_per_hour_high {high} "
f"<= ceiling_gp_per_hour {ceil}")
# --- per-slot asynchronous model ---
ssm = data.get("slot_state_machine") or {}
if not isinstance(ssm, dict):
errors.append(f"{rel}: flipping entries must declare a "
f"slot_state_machine block (per-slot asynchronous sub-states)")
return errors
declared = [str(s) for s in ssm.get("sub_states", [])]
if sorted(declared) != sorted(SLOT_SUB_STATES):
errors.append(
f"{rel}: slot_state_machine.sub_states must be exactly the canonical "
f"set: {', '.join(SLOT_SUB_STATES)} (got: {', '.join(declared) or 'none'})")
trans = ssm.get("slot_transitions") or {}
if not isinstance(trans, dict):
errors.append(f"{rel}: slot_state_machine.slot_transitions must be a mapping")
trans = {}
for from_s, tos in trans.items():
if from_s not in SLOT_SUB_STATES:
errors.append(f"{rel}: slot transition from unknown sub-state '{from_s}'")
for to_s in tos if isinstance(tos, list) else []:
if to_s not in SLOT_SUB_STATES:
errors.append(f"{rel}: slot transition '{from_s}' -> unknown "
f"sub-state '{to_s}'")
for state in SLOT_SUB_STATES:
if not trans.get(state):
errors.append(f"{rel}: slot sub-state '{state}' has no transitions "
f"— every state must be able to advance; a closed "
f"slot must release the physical slot via the "
f"archival closed -> free transition (governor 10471)")
if list(trans.get("closed") or []) != ["free"]:
errors.append(f"{rel}: 'closed' must have exactly the archival release "
f"['free'] — a closed slot with no outgoing transition "
f"performs one flip and then becomes permanently "
f"unusable (governor 10471)")
whens = [str(w) for w in ssm.get("slot_transition_whens", []) or []]
for w in whens:
arrow = w.split("->")
if len(arrow) != 2 or arrow[0].strip() not in SLOT_SUB_STATES \
or arrow[1].strip().split(":")[0].strip() not in SLOT_SUB_STATES:
errors.append(f"{rel}: slot_transition_whens entry '{w}' must be "
f"'from -> to: condition' with canonical sub-states")
for field in ("scheduler_iteration", "quote_age_policy", "persisted_state",
"limit_ledger", "conservation", "break_resume"):
if not str(ssm.get(field, "")).strip():
errors.append(f"{rel}: slot_state_machine.{field} must be declared")
# Two-rotation reuse regression (governor 10471): every flipping entry's
# encoded machine must complete buy/collect/sell/collect, release the
# physical slot to free, and support a second buy in the same slot.
errors += [f"{rel}: {e}" for e in slot_reuse_violations(trans)]
return errors
# --- Woodcutting structured gates (owner 10288 + governor 10398 shape) ---
# Canonical F2P axe table (OSRS wiki, verified 2026-08-28): the Woodcutting
# level is the TOOL-USE requirement (chopping), the Attack level is the
# EQUIP requirement. An account uses the highest axe whose Woodcutting level
# it meets, and equips it only when its Attack level is also met — otherwise
# it CARRIES that same axe (costing one inventory slot). Order is the tier
# tiebreak (iron beats bronze at equal requirements; gilded ~ rune).
AXE_TIERS = [
# (axe, woodcutting_level_to_use, attack_level_to_equip, members_only)
("bronze", 1, 1, False),
("iron", 1, 1, False),
("steel", 6, 5, False),
("black", 11, 10, False),
("mithril", 21, 20, False),
("adamant", 31, 30, False),
("rune", 41, 40, False),
("gilded", 41, 40, False),
("dragon", 61, 60, True),
]
AXE_NAMES = {row[0] for row in AXE_TIERS}
TIER_RANK = {"bronze": 0, "iron": 1, "steel": 2, "black": 3, "mithril": 4,
"adamant": 5, "rune": 6, "gilded": 6, "dragon": 7}
XP_PER_LOG = {"tree": 25, "oak": 37.5, "willow": 67.5, "maple": 100, "yew": 175}
TREE_MIN_LEVEL = {"tree": 1, "oak": 15, "willow": 30, "maple": 45, "yew": 60}
def best_axe(wc_level, attack_level):
"""Deterministic axe selection (owner guidance 2026-08-28 / governor 10323):
pick the highest tier usable at wc_level (tool-use requirement); equip it
only when attack_level meets that tier's equip requirement, else carry it.
Returns (axe, equipped_bool). Ties at the same Woodcutting requirement are
broken by TIER_RANK (iron beats bronze; rune beats the cosmetic gilded)."""
usable = [row for row in AXE_TIERS if row[3] is False and wc_level >= row[1]]
if not usable:
return None, False
best = max(usable, key=lambda row: (row[1], TIER_RANK[row[0]]))
return best[0], attack_level >= best[2]
def select_ladder_row(ladder, wc_level, quest_gate, combat_level):
"""Deterministic first match over a mode ladder (rows ordered min_level
DESC per the schema contract): the first row whose min_level the account
meets AND whose gate is satisfied.
Gates: none -> always; combat-15-plus -> combat_level >= 15;
combat-under-15 -> combat_level < 15 (below the Draynor dark-wizard
threshold); dragon-slayer-i -> quest_gate (Corsair Cove Resource Area).
Returns the matched row dict or None."""
for r in ladder:
if wc_level < int(r.get("min_level", 0)):
continue
gate = r.get("gate", "none")
if gate == "combat-15-plus" and not (combat_level >= 15):
continue
if gate == "combat-under-15" and not (combat_level < 15):
continue
if gate == "dragon-slayer-i" and not quest_gate:
continue
return r
return None
def row_rates(row, model, policy, capacity):
"""Per-SELECTED-ROW realized rate derivation (governor 10398 finding 4):
the account's ladder row carries its own chop cadence / bank-walk /
respawn duty / log price, so rates derive from the row actually chosen
(and the actual 27/28 capacity from axe_equipped) — never from a fixed
top tree or a hardcoded 28-log cycle.
Returns (logs_per_hour, xp_per_hour, gp_per_hour) with gp_per_hour = 0
for drop modes (realized GP; log value is reference-only)."""
chop = float(row.get("chop_seconds_per_log", 0))
walk = float(row.get("walk_seconds_each_way", 0))
respawn = float(row.get("respawn_duty", 0))
price = float(row.get("log_price", 0))
drop = float(model.get("drop_seconds_per_log") or 0)
deposit = float(model.get("deposit_seconds") or 0)
brk = float(model.get("break_duty") or 0)
tree = row.get("tree")
if tree not in XP_PER_LOG or chop <= 0 or respawn <= 0 or brk <= 0:
return None
if policy == "drop":
cycle = capacity * (chop + drop) + deposit # no bank legs
else:
cycle = capacity * chop + 2 * walk + deposit
if cycle <= 0:
return None
logs = (capacity / cycle) * 3600.0 * respawn * brk
xp = logs * XP_PER_LOG[tree]
gp = 0.0 if policy == "drop" else logs * price
return logs, xp, gp
def woodcutting_errors(rel, data):
"""Structured gates for woodcutting entries (owner guidance 10288 +
governor 10398; no phrase-search, exact computed checks):
- axe_tiers must match the canonical table exactly;
- axe_examples are recomputed and must match (incl. the 1-Attack /
41-Woodcutting rune-axe carry case);
- woodcutting_modes must be exactly [fast-xp, afk-xp, banked-income],
each ladder deterministic (rows ordered by min_level desc, first
match), with canonical per-tree level/gate constraints;
- mode `rates` is the canonical TOP-ROW reference (ladder[0], axe
equipped = 28 slots) and must equal that row's model chain within
tolerance, with xp/gp derived from the row's tree xp and log price;
- every LADDER ROW carries its own chop/walk/respawn/log_price and the
committed per-account woodcutting_fixtures rows must match the
selected-row derivation (27/28 capacity from axe_equipped) within
tolerance — governor 10398 finding 4;
- drop-policy modes realize ZERO gp (log value is a reference field,
never realized income) — governor 10398 finding 5;
- the state machine wires axe preparation BEFORE travel (a `prepare`
state with a bounded GE-buy chain leading to walk_to_trees), routes
an unobtainable axe to `failed` (never `complete`), and gates the
inventory-full drop transition to the drop policy mode only —
governor 10398 findings 1-3.
"""
errors = []
tags = {str(t).lower() for t in data.get("tags", [])}
if "woodcutting" not in tags:
return errors
# --- canonical axe table ---
tiers = data.get("axe_tiers") or []
canon = {(row[0], row[1], row[2], bool(row[3])) for row in AXE_TIERS}
declared = {(str(t.get("axe")), t.get("woodcutting_level"), t.get("attack_level"),
bool(t.get("members"))) for t in tiers}
if declared != canon:
errors.append(
f"{rel}: axe_tiers must match the canonical F2P table exactly "
f"(bronze 1/1, iron 1/1, steel 6/5, black 11/10, mithril 21/20, "
f"adamant 31/30, rune 41/40, gilded 41/40, dragon 61/60 members); "
f"got {sorted(declared)}")
return errors
# --- recompute the committed axe examples ---
for ex in data.get("axe_examples") or []:
name = ex.get("account", "?")
wc, atk = ex.get("woodcutting_level"), ex.get("attack_level")
axe, eq = best_axe(wc, atk)
if axe != ex.get("expected_axe") or eq != ex.get("expected_equipped"):
errors.append(
f"{rel}: axe_examples '{name}' (WC {wc}, Attack {atk}) — "
f"recomputed {axe}/equipped={eq}, entry declares "
f"{ex.get('expected_axe')}/equipped={ex.get('expected_equipped')}")
# --- modes ---
modes = data.get("woodcutting_modes") or []
if sorted(str(m.get("mode")) for m in modes) != ["afk-xp", "banked-income", "fast-xp"]:
errors.append(f"{rel}: woodcutting_modes must declare exactly "
f"fast-xp, afk-xp and banked-income")
return errors
for m in modes:
mode = m["mode"]
ladder = m.get("ladder") or []
if not ladder:
errors.append(f"{rel}: {mode} ladder is empty")
continue
levels = [int(r["min_level"]) for r in ladder]
if levels != sorted(levels, reverse=True):
errors.append(f"{rel}: {mode} ladder must be ordered by "
f"min_level DESC (deterministic first-match); got {levels}")
if any(re.search(r"\bor\b", str(r.get("spot", "")).lower())
or re.search(r"\bor\b", str(r.get("bank", "")).lower())
for r in ladder):
errors.append(f"{rel}: {mode} ladder rows must be unambiguous "
f"(no 'or' inside spot/bank fields)")
for r in ladder:
tree, min_level, gate = r["tree"], int(r["min_level"]), r.get("gate")
if min_level != TREE_MIN_LEVEL.get(tree):
errors.append(f"{rel}: {mode} {tree} must have min_level "
f"{TREE_MIN_LEVEL.get(tree)} (got {min_level})")
if tree == "maple" and gate != "dragon-slayer-i":
errors.append(f"{rel}: {mode} maple requires gate "
f"dragon-slayer-i (Corsair Cove Resource Area)")
if tree != "maple" and gate == "dragon-slayer-i":
errors.append(f"{rel}: {mode} gate dragon-slayer-i may only "
f"apply to the maple row (got {tree})")
# governor 10398 finding 4: every row must carry its own rate
# constants — rates derive from the SELECTED row, never from a
# fixed top-tree model.
for field in ("chop_seconds_per_log", "walk_seconds_each_way",
"respawn_duty", "log_price"):
if not isinstance(r.get(field), (int, float)):
errors.append(f"{rel}: {mode} {tree} row must carry "
f"numeric {field} (per-selected-row rates, "
f"governor 10398 finding 4)")
if mode == "fast-xp" and not any(r["tree"] == "willow" and int(r["min_level"]) == 30
for r in ladder):
errors.append(f"{rel}: fast-xp must include a willow row at "
f"min_level 30 (willow is the intended fast-XP target)")
if mode == "banked-income" and not any(r["tree"] == "yew" for r in ladder):
errors.append(f"{rel}: banked-income must include a yew row "
f"(yew from 60 when eligible)")
if mode == "afk-xp" and not any(r["tree"] == "maple" for r in ladder):
errors.append(f"{rel}: afk-xp must include a maple row (quest-gated "
f"AFK target); include quest-gated maple only when "
f"access is satisfied — with lower fallbacks below")
if not any(int(r["min_level"]) == 1 for r in ladder):
errors.append(f"{rel}: {mode} ladder must bottom out at a level-1 "
f"fallback (tree) — deterministic lower-level fallback")
# --- canonical TOP-ROW `rates` reference (ladder[0], 28 slots equipped) ---
model = m.get("model") or {}
policy = m.get("inventory_policy")
top = row_rates(ladder[0], model, policy, 28)
if top is None:
errors.append(f"{rel}: {mode} top-row rate chain does not compute "
f"(row constants/model numeric, cycle positive)")
continue
t_logs, t_xp, t_gp = top
rates = m.get("rates") or {}
logs = rates.get("logs_per_hour")
if not isinstance(logs, (int, float)) or abs(logs - t_logs) / t_logs > 0.01:
errors.append(
f"{rel}: {mode} logs_per_hour {logs} must equal the SELECTED-row "
f"chain (top row {ladder[0]['tree']}, 28 slots equipped: "
f"{t_logs:.1f}) within 1% — rates are per-selected-row, never "
f"a fixed 28-log top-tree hardcode (governor 10398 finding 4)")
if abs(rates.get("xp_per_hour", -1) - t_xp) / t_xp > 0.02:
errors.append(f"{rel}: {mode} xp_per_hour {rates.get('xp_per_hour')} "
f"must equal the row chain xp {t_xp:.0f} "
f"(logs x {XP_PER_LOG[ladder[0]['tree']]}) within 2%")
if policy == "drop":
# governor 10398 finding 5: dropped logs realize ZERO gp; the log
# sell price is reference/opportunity cost only.
if rates.get("gp_per_hour", -1) != 0:
errors.append(f"{rel}: {mode} inventory_policy drop must "
f"realize gp_per_hour 0 (dropped logs produce no "
f"income; log_price is reference only) — got "
f"{rates.get('gp_per_hour')}")
else:
price = rates.get("log_price")
if isinstance(logs, (int, float)) and isinstance(price, (int, float)):
gp_expect = t_logs * price
if abs(rates.get("gp_per_hour", -1) - gp_expect) / gp_expect > 0.03:
errors.append(f"{rel}: {mode} gp_per_hour "
f"{rates.get('gp_per_hour')} must equal "
f"logs_per_hour x declared log_price "
f"({gp_expect:.0f}) within 3%")
# --- per-account fixtures: selected-row + 27/28 capacity derivation ---
for fx in data.get("woodcutting_fixtures") or []:
name = fx.get("account", "?")
mode = fx.get("mode")
wc, atk = fx.get("woodcutting_level"), fx.get("attack_level")
qgate = bool(fx.get("quest_gate"))
combat = fx.get("combat_level", 0)
modes_by_name = {str(m.get("mode")): m for m in modes}
m = modes_by_name.get(str(mode))
if m is None:
errors.append(f"{rel}: fixture '{name}' references unknown "
f"mode {mode!r}")
continue
row = select_ladder_row(m.get("ladder") or [], wc, qgate, combat)
if row is None:
errors.append(f"{rel}: fixture '{name}' (WC {wc}, quest "
f"{qgate}, combat {combat}) matches no ladder row")
continue
if row["tree"] != fx.get("expected_tree"):
errors.append(f"{rel}: fixture '{name}' selected row "
f"{row['tree']}@{row['min_level']} but declares "
f"expected_tree {fx.get('expected_tree')}")
axe, eq = best_axe(wc, atk)
if axe != fx.get("expected_axe") or eq != fx.get("expected_equipped"):
errors.append(f"{rel}: fixture '{name}' — recomputed axe "
f"{axe}/equipped={eq}, declares "
f"{fx.get('expected_axe')}/equipped="
f"{fx.get('expected_equipped')}")
capacity = 28 if eq else 27 # governor 10398 finding 4
rr = row_rates(row, m.get("model") or {}, m.get("inventory_policy"), capacity)
if rr is None:
errors.append(f"{rel}: fixture '{name}' rate chain does not "
f"compute (capacity {capacity})")
continue
f_logs, f_xp, f_gp = rr
# fixture values are pinned to the recomputed chain within the same
# tolerances as the mode rates (logs 1% / xp 2% / gp 3%); drop-mode
# fixtures must realize zero gp.
expect_logs = fx.get("expected_logs_per_hour")
if not isinstance(expect_logs, (int, float)) or abs(expect_logs - f_logs) / f_logs > 0.01:
errors.append(f"{rel}: fixture '{name}' expected_logs_per_hour "
f"{expect_logs} must equal the selected-row chain "
f"({row['tree']}@{row['min_level']}, capacity "
f"{capacity}, respawn {row.get('respawn_duty')}, "
f"equipped={eq}) = {f_logs:.1f} within 1%")
expect_xp = fx.get("expected_xp_per_hour")
if not isinstance(expect_xp, (int, float)) or abs(expect_xp - f_xp) / f_xp > 0.02:
errors.append(f"{rel}: fixture '{name}' expected_xp_per_hour "
f"{expect_xp} != selected-row xp {f_xp:.0f} within 2%")
if m.get("inventory_policy") == "drop":
if fx.get("expected_gp_per_hour", -1) != 0:
errors.append(f"{rel}: fixture '{name}' is drop mode — "
f"realized expected_gp_per_hour must be 0")
else:
expect_gp = fx.get("expected_gp_per_hour")
if expect_gp is None or abs(expect_gp - f_gp) / f_gp > 0.03:
errors.append(f"{rel}: fixture '{name}' expected_gp_per_hour "
f"{expect_gp} != selected-row gp {f_gp:.0f} "
f"within 3%")
# --- state-machine wiring (governor 10398 findings 1-3) ---
sm = data.get("state_machine") or {}
states = set(sm.get("states", {}))
if "prepare" not in states:
errors.append(f"{rel}: state_machine must declare a `prepare` state "
f"(executable axe acquisition before walking to trees — "
f"governor 10398 finding 1)")
else:
tr = sm.get("transitions") or []
from_to = {(t.get("from"), t.get("to")) for t in tr}
prep_f = sum(1 for t in tr if t.get("from") == "prepare" and t.get("to") == "walk_to_trees")
if not prep_f:
errors.append(f"{rel}: `prepare` must transition to walk_to_trees "
f"when the axe is acquired (governor 10398 finding 1)")
if ("prepare", "failed") not in from_to:
errors.append(f"{rel}: `prepare` must transition to `failed` when "
f"no axe is obtainable after bounded recovery — a "
f"missing tool is a FAILED run, never `complete` "
f"(governor 10398 finding 2)")
# idle -> complete must not be the no-axe escape hatch
for t in tr:
if t.get("from") == "idle" and t.get("to") == "complete":
if re.search(r"no axe|no-axe|obtainable", str(t.get("when", "")), re.I):
errors.append(f"{rel}: idle -> complete must fire only on "
f"the session target, never on a missing axe "
f"(governor 10398 finding 2)")
# inventory-full drop transition gated to the drop policy mode only
drop_trans = [t for t in (sm.get("transitions") or [])
if t.get("from") == "chopping" and t.get("to") == "drop_logs"]
bank_trans = [t for t in (sm.get("transitions") or [])
if t.get("from") == "chopping" and t.get("to") == "walk_to_bank"]
if drop_trans and "afk-xp" not in str(drop_trans[0].get("when", "")):
errors.append(f"{rel}: chopping -> drop_logs must be gated explicitly "
f"to the afk-xp drop mode (inventory_policy drop) — "
f"governor 10398 finding 3")
if not drop_trans and bank_trans:
# drop mode exists somewhere in the modes; if the entry declares a
# drop policy it must also declare the drop transition
if any(str(m.get("inventory_policy")) == "drop" for m in modes):
errors.append(f"{rel}: a drop-policy mode exists but chopping has "
f"no drop_logs transition (finding 3)")
return errors
def run_woodcutting_selftest(verbose=False):
"""Regression fixtures for the woodcutting structured gates
(canonical axe table, deterministic per-mode ladders,
per-selected-row 27/28 dynamic rates, drop-mode realized GP zero,
prepare/failed wiring, drop-transition gating); returns 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}")
# --- woodcutting structured gates (owner 10288 / governor 10398) ---
check("axe: 41-Woodcutting/1-Attack account carries and USES a rune axe (does not pick a lower axe)",
best_axe(41, 1) == ("rune", False))
check("axe: 41-Woodcutting/40-Attack account equips the rune axe",
best_axe(41, 40) == ("rune", True))
check("axe: 30-Woodcutting account is below adamant (31) and uses mithril (equippable at 20 Attack)",
best_axe(30, 40) == ("mithril", True))
check("axe: 31-Woodcutting/5-Attack account uses adamant but cannot equip it (carries)",
best_axe(31, 5) == ("adamant", False))
check("axe: level-1 account uses an iron axe equipped from the start",
best_axe(1, 1) == ("iron", True))
check("axe: dragon (61/60, members) is never selected for an F2P woodcutting entry; maxed account still picks the top F2P axe (rune)",
best_axe(99, 99) == ("rune", True))
canon = {(row[0], row[1], row[2]) for row in AXE_TIERS} - {("dragon", 61, 60)}
want = {("bronze", 1, 1), ("iron", 1, 1), ("steel", 6, 5), ("black", 11, 10),
("mithril", 21, 20), ("adamant", 31, 30), ("rune", 41, 40), ("gilded", 41, 40)}
check("axe: canonical F2P axe table matches the wiki (WC to use / Attack to equip)",
canon == want)
def wc_fake_entry(**over):
base = {"tags": ["woodcutting"],
"axe_tiers": [{"axe": a, "woodcutting_level": w, "attack_level": k, "members": m}
for a, w, k, m in AXE_TIERS],
"axe_examples": [{"account": "41WC/1Atk", "woodcutting_level": 41, "attack_level": 1,
"expected_axe": "rune", "expected_equipped": False}],
"state_machine": {"states": {"idle": {"goal": "start", "next": ["arm_run"]},
"arm_run": {"goal": "pick", "next": ["prepare"]},
"prepare": {"goal": "acquire axe", "next": ["walk_to_trees", "failed"]},
"chopping": {"goal": "chop", "next": ["drop_logs", "walk_to_bank"]},
"drop_logs": {"goal": "drop", "next": ["chopping"]},
"walk_to_bank": {"goal": "bank", "next": ["chopping"]},
"failed": {"goal": "fail"}},
"initial": "idle", "terminal": ["failed"],
"transitions": [
{"from": "idle", "when": "session target reached", "to": "complete"},
{"from": "arm_run", "when": "row pinned", "to": "prepare"},
{"from": "prepare", "when": "axe acquired", "to": "walk_to_trees"},
{"from": "prepare", "when": "axe unobtainable", "to": "failed"},
{"from": "chopping", "when": "mode is afk-xp (inventory_policy drop) and inventory full", "to": "drop_logs"},
{"from": "chopping", "when": "mode is fast-xp or banked-income and inventory full", "to": "walk_to_bank"}]},
"woodcutting_fixtures": [
{"account": "f1", "mode": "fast-xp", "woodcutting_level": 50, "attack_level": 40,
"quest_gate": False, "combat_level": 50, "expected_tree": "willow",
"expected_axe": "rune", "expected_equipped": True,
"expected_logs_per_hour": 622, "expected_xp_per_hour": 41985,
"expected_gp_per_hour": 13062}],
"woodcutting_modes": [
{"mode": "fast-xp", "intent": "max xp per hour on the F2P ladder",
"inventory_policy": "bank",
"ladder": [
{"tree": "willow", "min_level": 30, "gate": "combat-15-plus",
"spot": "draynor willows", "bank": "draynor",
"chop_seconds_per_log": 4.0, "walk_seconds_each_way": 12,
"respawn_duty": 0.92, "log_price": 21},
{"tree": "willow", "min_level": 30, "gate": "combat-under-15",
"spot": "edgeville willow", "bank": "edgeville",
"chop_seconds_per_log": 4.0, "walk_seconds_each_way": 6,
"respawn_duty": 0.90, "log_price": 21},
{"tree": "oak", "min_level": 15, "gate": "none",
"spot": "draynor oaks", "bank": "draynor",
"chop_seconds_per_log": 3.5, "walk_seconds_each_way": 8,
"respawn_duty": 0.93, "log_price": 40},
{"tree": "tree", "min_level": 1, "gate": "none",
"spot": "lumbridge west", "bank": "lumbridge",
"chop_seconds_per_log": 3.0, "walk_seconds_each_way": 10,
"respawn_duty": 0.95, "log_price": 23}],
"model": {"drop_seconds_per_log": 0, "deposit_seconds": 4,
"break_duty": 0.94},
"rates": {"logs_per_hour": 622, "xp_per_hour": 41985,
"gp_per_hour": 13062, "log_price": 21,
"price_verified": "2026-08-28"}},
{"mode": "afk-xp", "intent": "low-interaction quest-gated progress",
"inventory_policy": "drop",
"ladder": [
{"tree": "maple", "min_level": 45, "gate": "dragon-slayer-i",
"spot": "corsair cove resource area", "bank": "none-drop",
"chop_seconds_per_log": 6.0, "walk_seconds_each_way": 0,
"respawn_duty": 0.80, "log_price": 12},
{"tree": "willow", "min_level": 30, "gate": "combat-15-plus",
"spot": "draynor willows", "bank": "none-drop",
"chop_seconds_per_log": 6.0, "walk_seconds_each_way": 0,
"respawn_duty": 0.92, "log_price": 21},
{"tree": "oak", "min_level": 15, "gate": "none",
"spot": "draynor oaks", "bank": "none-drop",
"chop_seconds_per_log": 5.5, "walk_seconds_each_way": 0,
"respawn_duty": 0.93, "log_price": 40},
{"tree": "tree", "min_level": 1, "gate": "none",
"spot": "lumbridge west", "bank": "none-drop",
"chop_seconds_per_log": 5.0, "walk_seconds_each_way": 0,
"respawn_duty": 0.95, "log_price": 23}],
"model": {"drop_seconds_per_log": 1.0, "deposit_seconds": 0,
"break_duty": 0.94},
"rates": {"logs_per_hour": 386, "xp_per_hour": 38600,
"gp_per_hour": 0, "log_price": 12,
"price_verified": "2026-08-28"}},
{"mode": "banked-income", "intent": "banked logs sold at the GE",
"inventory_policy": "bank",
"ladder": [
{"tree": "yew", "min_level": 60, "gate": "none",
"spot": "edgeville south yews", "bank": "edgeville",
"chop_seconds_per_log": 7.5, "walk_seconds_each_way": 8,
"respawn_duty": 0.52, "log_price": 111},
{"tree": "willow", "min_level": 30, "gate": "combat-15-plus",
"spot": "draynor willows", "bank": "draynor",
"chop_seconds_per_log": 4.0, "walk_seconds_each_way": 12,
"respawn_duty": 0.92, "log_price": 21},
{"tree": "oak", "min_level": 15, "gate": "none",
"spot": "draynor oaks", "bank": "draynor",
"chop_seconds_per_log": 3.5, "walk_seconds_each_way": 8,
"respawn_duty": 0.93, "log_price": 40},
{"tree": "tree", "min_level": 1, "gate": "none",
"spot": "lumbridge west", "bank": "lumbridge",
"chop_seconds_per_log": 3.0, "walk_seconds_each_way": 10,
"respawn_duty": 0.95, "log_price": 23}],
"model": {"drop_seconds_per_log": 0, "deposit_seconds": 4,
"break_duty": 0.94},
"rates": {"logs_per_hour": 214, "xp_per_hour": 37450,
"gp_per_hour": 23754, "log_price": 111,
"price_verified": "2026-08-28"}}]}
base.update(over)
return base
check("modes: a compliant 3-mode entry (per-row rates, drop-GP-zero, prepare wiring) passes every structured gate",
woodcutting_errors("selftest/wc-good", wc_fake_entry()) == [])
bad = wc_fake_entry(woodcutting_modes=wc_fake_entry()["woodcutting_modes"][1:])
check("modes: missing a mode (only 2 declared) is rejected",
any("exactly" in e for e in woodcutting_errors("selftest/wc-missing", bad)))
bad = wc_fake_entry(axe_examples=[{"account": "41WC/1Atk", "woodcutting_level": 41, "attack_level": 1,
"expected_axe": "adamant", "expected_equipped": True}])
check("modes: a 41-WC/1-Attack example declaring a lower/equipped axe is rejected",
any("recomputed rune" in e for e in woodcutting_errors("selftest/wc-badex", bad)))
bad = wc_fake_entry(woodcutting_modes=[dict(m, ladder=[dict(r, min_level=30 if r["tree"] == "maple" else r["min_level"]) for r in m["ladder"]])
for m in wc_fake_entry()["woodcutting_modes"]])
check("modes: maple below level 45 is rejected",
any("maple must have min_level 45" in e for e in woodcutting_errors("selftest/wc-maple45", bad)))
bad = wc_fake_entry(woodcutting_modes=[dict(m, ladder=[dict(r, gate="none") for r in m["ladder"]])
for m in wc_fake_entry()["woodcutting_modes"]])
check("modes: maple without the dragon-slayer-i gate is rejected",
any("gate dragon-slayer-i" in e for e in woodcutting_errors("selftest/wc-maplegate", bad)))
bad = wc_fake_entry(woodcutting_modes=[dict(m, ladder=[dict(r) for r in m["ladder"][::-1]])
for m in wc_fake_entry()["woodcutting_modes"]])
check("modes: a ladder not ordered by min_level DESC is rejected (nondeterministic)",
any("ordered by min_level DESC" in e for e in woodcutting_errors("selftest/wc-order", bad)))
badm = [dict(m, rates=dict(m["rates"], logs_per_hour=9999))
for m in wc_fake_entry()["woodcutting_modes"]]
check("modes: a rate far above the model chain (9999/hr) is rejected",
any("SELECTED-row" in e for e in woodcutting_errors("selftest/wc-rate", wc_fake_entry(woodcutting_modes=badm))))
bad = wc_fake_entry(axe_tiers=[{"axe": a, "woodcutting_level": w, "attack_level": k, "members": m}
for a, w, k, m in AXE_TIERS if a != "rune"])
check("modes: a non-canonical axe table (rune missing) is rejected",
any("canonical F2P table" in e for e in woodcutting_errors("selftest/wc-axes", bad)))
badm = wc_fake_entry()["woodcutting_modes"]
badm[0] = dict(badm[0], rates=dict(badm[0]["rates"], xp_per_hour=40000))
check("modes: xp_per_hour not equal to logs x per-log xp is rejected",
any("xp_per_hour" in e for e in woodcutting_errors("selftest/wc-xp", wc_fake_entry(woodcutting_modes=badm))))
# governor 10398 finding 4: a fixture whose expected logs/XP do not match
# the selected-row chain with the 27/28 capacity is rejected.
bad = wc_fake_entry(woodcutting_fixtures=[
{"account": "carried-rune-27", "mode": "afk-xp", "woodcutting_level": 50, "attack_level": 1,
"quest_gate": True, "combat_level": 50, "expected_tree": "maple",
"expected_axe": "rune", "expected_equipped": False,
"expected_logs_per_hour": 9999, "expected_xp_per_hour": 999999,
"expected_gp_per_hour": 0}])
check("modes: a fixture whose declared rates do not match the selected-row 27-slot chain is rejected (governor 10398 finding 4)",
any("selected-row chain" in e for e in woodcutting_errors("selftest/wc-fxrate", bad)))
# governor 10398 finding 5: a drop mode declaring non-zero realized GP.
badm = wc_fake_entry()["woodcutting_modes"]
badm[1] = dict(badm[1], rates=dict(badm[1]["rates"], gp_per_hour=4632))
check("modes: drop mode with non-zero realized gp_per_hour is rejected (dropped logs realize 0 gp — governor 10398 finding 5)",
any("realize gp_per_hour 0" in e for e in woodcutting_errors("selftest/wc-dropgp", wc_fake_entry(woodcutting_modes=badm))))
# governor 10398 finding 2: idle -> complete must not be the no-axe hatch.
bad = wc_fake_entry(state_machine=dict(
wc_fake_entry()["state_machine"],
transitions=[{"from": "idle", "when": "no axe obtainable", "to": "complete"}]))
check("modes: idle -> complete on a missing axe is rejected (unobtainable axe must go to failed — governor 10398 finding 2)",
any("never on a missing axe" in e for e in woodcutting_errors("selftest/wc-idlecomplete", bad)))
# governor 10398 finding 3: the drop transition must be gated to afk-xp.
bad = wc_fake_entry(state_machine=dict(
wc_fake_entry()["state_machine"],
transitions=[{"from": "chopping", "when": "inventory full", "to": "drop_logs"}]))
check("modes: an ungated chopping -> drop_logs transition is rejected (drop route exclusive to the afk-xp drop mode — governor 10398 finding 3)",
any("gated explicitly" in e for e in woodcutting_errors("selftest/wc-dropgate", bad)))
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 run_flip_selftest(verbose=False):
"""Regression fixtures for the GE floor-tax + GE-flip lifecycle gates
(merged from PR #7 / main reconciliation); returns 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}")
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),
]
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'}")
# --- per-slot scheduler dispatch: the worked example as a regression ---
# Governor 10323 finding 3 + owner: reproduce the simultaneous
# partial-buy / full-buy-fill / no-fill-buy case and prove no
# unrelated slot is closed or advanced. Under governor 10390 finding 2
# a FULL buy fill goes to buy-filled-awaiting-collect (assets sit in
# the GE collection box) — the pass dispatches ge_collect, it never
# lets the slot sell directly.
slot_states = {"arrowtips": "buy-open", # partial fill 4,000/10,000 — resting
"iron": "buy-filled-awaiting-collect", # full fill — ge_collect this pass
"cosmic": "buy-open"} # no fill, edge positive — resting
dispatched = {name: scheduler_dispatch(s) for name, s in slot_states.items()}
ok = dispatched == {"arrowtips": None, "iron": "collect", "cosmic": None}
check("worked example: partial-buy + full-buy-fill + no-fill slots dispatch only the iron slot to ge_collect; unrelated slots stay untouched",
ok)
check("resting slots (buy-open) are never closed or advanced by a pass",
dispatched.get("arrowtips") is None and dispatched.get("cosmic") is None)
check("a fully filled buy must pass through buy-filled-awaiting-collect (ge_collect) before any sell",
scheduler_dispatch("buy-filled-awaiting-collect") == "collect"
and scheduler_dispatch("sell-filled-awaiting-collect") == "collect"
and scheduler_dispatch("ready-to-sell") == "sell")
# --- side-aware conservation invariants (finding 4) ---
# BUY balanced snapshot mirroring the worked example t1: 10,000 Rune
# arrowtips quoted @ 223; 4,000 filled (units in the GE collection
# box), 6,000 still open on the resting offer; cash 2,230,000 =
# spent 892,000 + still_committed 1,338,000 + refunded 0.
buy_partial = {
"side": "buy",
"quoted": {"Rune arrowtips": 10_000},
"filled": {"Rune arrowtips": 4_000},
"still_open": {"Rune arrowtips": 6_000},
"cancelled_unfilled": {},
"cash_quoted": 2_230_000, "spent": 892_000,
"still_committed": 1_338_000, "refunded": 0,
}
check("conservation: BUY side — partial fill splits units into filled + still_open and cash into spent + still_committed (partition sums hold, zero violations)",
conservation_violations(buy_partial) == [])
buy_cancelled = dict(buy_partial, still_open={},
cancelled_unfilled={"Rune arrowtips": 6_000},
still_committed=0, refunded=1_338_000)
check("conservation: BUY side — a cancelled partial buy carries filled units + refunded CASH (still_open -> cancelled_unfilled + refunded), zero violations",
conservation_violations(buy_cancelled) == [])
phantom_rec = dict(buy_cancelled, refunded={"Rune arrowtips": 6_000})
check("conservation: BUY side — 'refunded UNITS' on a buy is a phantom (unfilled/cancelled quantity is CASH): violation",
any("CASH" in v or "invalid for side" in v
for v in conservation_violations(phantom_rec)))
sold_on_buy = dict(buy_partial, sold={"Rune arrowtips": 4_000})
check("conservation: BUY side — a full buy that sells before ge_collect (sold units on a buy ledger) is a violation",
any("invalid for side" in v for v in conservation_violations(sold_on_buy)))
buy_cash_bad = dict(buy_partial, still_committed=1_380_000)
check("conservation: BUY side — cash partition mismatch (spent + still_committed + refunded != cash_quoted) is a violation",
any("cash quoted" in v for v in conservation_violations(buy_cash_bad)))
# SELL balanced: 10,000 sold @ 157, proceeds 1,570,000 in the box.
sell_full = {
"side": "sell",
"quoted": {"Iron bar": 10_000},
"sold": {"Iron bar": 10_000},
"still_open": {}, "returned_unsold": {},
"proceeds_in_box": 1_570_000, "cash_collected": 0,
}
check("conservation: SELL side — full sell partitions sold + still_open + returned_unsold, proceeds in the box (zero violations)",
conservation_violations(sell_full) == [])
sell_partial = dict(sell_full, sold={"Iron bar": 4_000},
still_open={"Iron bar": 6_000},
proceeds_in_box=628_000)
check("conservation: SELL side — a partial sell legitimately spans sold + still_open (zero violations)",
conservation_violations(sell_partial) == [])
sell_dup = dict(sell_full, sold={"Iron bar": 6_000},
still_open={"Iron bar": 3_000},
returned_unsold={"Iron bar": 1_000})
check("conservation: SELL side — units both in a live offer AND returned (impossible pair) is a violation",
any("both still_open and returned_unsold" in v
for v in conservation_violations(sell_dup)))
over_rec = dict(buy_partial, filled={"Rune arrowtips": 7_000})
check("conservation: over-reported buckets break the quoted partition (sum 13,000 != quoted 10,000)",
any("quoted 10000 !=" in v for v in conservation_violations(over_rec)))
neg_rec = dict(sell_full, sold={"Iron bar": -1})
check("conservation: a negative bucket is a violation",
any("negative" in v for v in conservation_violations(neg_rec)))
# --- phase gates (finding 4: assets cannot be reused/closed pre-collect) ---
check("phase gate: a full buy that sells before ge_collect (buy-filled-awaiting-collect with sold units) is rejected",
any("cannot sell before ge_collect" in v
for v in slot_phase_violations(dict(buy_partial, sub_state="buy-filled-awaiting-collect",
filled={"Rune arrowtips": 10_000}, still_open={},
sold={"Rune arrowtips": 10_000}))))
check("phase gate: a full sell that closes before proceeds collection (closed with proceeds_in_box > 0) is rejected",
any("before proceeds collection" in v
for v in slot_phase_violations(dict(sell_full, sub_state="closed"))))
check("phase gate: a sell-filled-awaiting-collect slot with sold units but no proceeds_in_box is rejected",
any("no proceeds_in_box" in v
for v in slot_phase_violations(dict(sell_full, sub_state="sell-filled-awaiting-collect",
proceeds_in_box=0))))
check("phase gate: a closed slot that has been fully collected (no proceeds_in_box, no cancelled/returned units) passes",
slot_phase_violations(dict(sell_full, sub_state="closed",
proceeds_in_box=0, cash_collected=1_570_000)) == [])
# --- 4h buy-limit ledger vs offer expiry (finding 1) ---
check("buy limit: a requote within the remaining capacity is allowed (4,000 used of 10,000 -> 6,000)",
buy_capacity_ok(limit_used=4000, limit=10000, qty_requested=6000))
check("buy limit: a requote beyond the remaining capacity is blocked (6,001 > 6,000 remaining)",
not buy_capacity_ok(limit_used=4000, limit=10000, qty_requested=6001))
check("buy limit: a cancelled offer does NOT renew the limit (10,000 used of 10,000 -> any requote blocked before reset)",
not buy_capacity_ok(limit_used=10000, limit=10000, qty_requested=1))
# --- structured lifecycle gates (findings 1-5 + governor 10390) ---
margin_tags = ["flipping"]
ssm_ok = {
"sub_states": SLOT_SUB_STATES,
"slot_transitions": {
"free": ["buy-open"],
"buy-open": ["buy-filled-awaiting-collect",
"buy-cancelled-awaiting-collect"],
"buy-filled-awaiting-collect": ["ready-to-sell", "free"],
"buy-cancelled-awaiting-collect": ["ready-to-sell", "ready-to-requote", "free"],
"ready-to-sell": ["sell-open"],
"sell-open": ["sell-filled-awaiting-collect",
"sell-cancelled-awaiting-collect"],
"sell-filled-awaiting-collect": ["closed"],
"sell-cancelled-awaiting-collect": ["ready-to-requote", "closed"],
"ready-to-requote": ["buy-open", "sell-open", "free"],
"closed": ["free"], # archival release: physical slot reusable (governor 10471)
},
"slot_transition_whens": [
"free -> buy-open: quote placed",
"buy-open -> buy-filled-awaiting-collect: offer FULLY filled — units sit in the GE collection box until ge_collect",
"buy-open -> buy-cancelled-awaiting-collect: abort on stale edge (margin no longer positive) or strategy reallocation — never on the 4h limit reset",
"buy-filled-awaiting-collect -> ready-to-sell: ge_collect withdrawn units; sell edge positive",
"buy-filled-awaiting-collect -> free: ge_collect withdrawn; margin gone, slot released",
"buy-cancelled-awaiting-collect -> ready-to-sell: collected, sell edge positive",
"buy-cancelled-awaiting-collect -> ready-to-requote: collected; margin positive and capacity remains",
"buy-cancelled-awaiting-collect -> free: collected, full cash refund, slot freed",
"ready-to-sell -> sell-open: sell offer placed",
"sell-open -> sell-filled-awaiting-collect: sell FULLY filled — proceeds sit in the box until ge_collect",
"sell-open -> sell-cancelled-awaiting-collect: abort on stale edge or strategy reallocation",
"sell-filled-awaiting-collect -> closed: proceeds collected and ledger reconciled + ARCHIVED",
"sell-cancelled-awaiting-collect -> ready-to-requote: collected, requote",
"sell-cancelled-awaiting-collect -> closed: collected, ledger archived, slot closed (exit taken)",
"closed -> free: archival release — completed cycle's ledger stays archived; the physical GE slot is reusable for the next flip",
"ready-to-requote -> buy-open: requoted buy",
"ready-to-requote -> sell-open: requoted sell",
"ready-to-requote -> free: abandoned",
],
"scheduler_iteration": "one pass over all slots; dispatch pending actions only",
"quote_age_policy": "offers rest while the scanned margin stays positive; aborted only on a stale edge or strategy reallocation — the 4h limit reset never cancels an offer",
"persisted_state": "per-slot record: sub-state + side-aware quoted/filled/sold/collected buckets",
"limit_ledger": "limit_used + first_fill_ts + limit_resets_at, independent of offer expiry; cancellation never renews it",
"conservation": "side-aware: buy quoted units = filled + still_open + cancelled_unfilled and acquired items = filled only; sell quoted = sold + still_open + returned_unsold",
"break_resume": "breaks between passes; state persists in the per-slot record",
}
flip_ok = {"tags": margin_tags, "economy": {
"income_per_cycle": 244, "cost_per_cycle": 223, "net_per_cycle": 17,
"ceiling_gp_per_hour": 149500, "gp_per_hour_low": 102600,
"gp_per_hour_high": 138800}, "slot_state_machine": ssm_ok}
check("flip lifecycle: ceiling + nested realized bounds + canonical 10-state slot machine passes",
ge_lifecycle_errors("ok.yaml", flip_ok) == [])
errs = ge_lifecycle_errors("bad.yaml", {"tags": margin_tags, "economy": {
"income_per_cycle": 244, "cost_per_cycle": 223, "net_per_cycle": 17,
"ceiling_gp_per_hour": 149500, "gp_per_hour_low": 160000,
"gp_per_hour_high": 138800}})
check("flip lifecycle: realized low above the ceiling is rejected (bounds must nest)",
any("nest" in e for e in errs))
errs = ge_lifecycle_errors("bad2.yaml", {"tags": margin_tags, "economy": {
"income_per_cycle": 244, "cost_per_cycle": 223, "net_per_cycle": 17,
"gp_per_hour_low": 0, "gp_per_hour_high": 138800}})
check("flip lifecycle: missing typed ceiling field is rejected",
any("ceiling_gp_per_hour" in e for e in errs))
errs = ge_lifecycle_errors("bad3.yaml", dict(flip_ok, slot_state_machine=dict(
ssm_ok, sub_states=["free", "buy-open"])))
check("flip lifecycle: non-canonical sub-state set is rejected",
any("canonical" in e for e in errs))
errs = ge_lifecycle_errors("bad4.yaml", dict(flip_ok, slot_state_machine=dict(
ssm_ok, slot_transitions={"free": ["buy-open"],
"buy-open": ["ready-to-sell",
"buy-cancelled-awaiting-collect"]},
slot_transition_whens=ssm_ok["slot_transition_whens"][:4])))
check("flip lifecycle: a sub-state without transitions (ready-to-sell missing) is rejected",
any("no transitions" in e for e in errs))
errs = ge_lifecycle_errors("bad5.yaml", dict(flip_ok, slot_state_machine=dict(
ssm_ok, slot_transitions=dict(ssm_ok["slot_transitions"], closed=[]))))
check("flip lifecycle: a 'closed' sink (no archival release) is rejected — the physical slot would perform one flip and strand",
any("archival release" in e for e in errs))
# --- slot reuse through rotations (governor 10471) ---
check("slot cycle: the archival release closed -> free is a dispatched scheduler action",
scheduler_dispatch("closed") == "release")
check("slot reuse: one full buy/collect/sell/collect lifecycle completes and the SAME physical slot opens a second buy (two-rotation walk passes)",
slot_reuse_violations(ssm_ok["slot_transitions"]) == [])
check("slot reuse: the rotation-1 archived ledger survives rotation 2 intact (no loss, duplication, or mutation)",
archived_ledger_preserved_after_second_buy() == [])
sink_trans = dict(ssm_ok["slot_transitions"], closed=[])
check("slot reuse: a machine whose closed has no outgoing transition FAILS the two-rotation regression (one flip, then permanently unusable)",
any("permanently unusable" in v
for v in slot_reuse_violations(sink_trans)))
closed_no_free = dict(ssm_ok["slot_transitions"], closed=["ready-to-sell"])
check("slot reuse: a closed that does not release to free FAILS the two-rotation regression",
any("closed -> free" in v
for v in slot_reuse_violations(closed_no_free)))
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")
env = dict(os.environ, CODEX_SKIP_REGRESSION_SUITES="1")
proc = subprocess.run(
[sys.executable, str(Path(__file__).resolve()), "--stale-days", "9999"],
capture_output=True, text=True, cwd=str(ROOT), timeout=120, env=env)
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)
errors += run_flip_selftest(verbose=True)
errors += run_mining_selftest(verbose=True)
errors += run_woodcutting_selftest(verbose=True)
print(f"RESULT: selftest, {len(errors)} errors, 0 warnings")
return 1 if errors else 0
# The fixes themselves are gated: all regression suites run on every
# invocation (alchemy/citation, mining model, GE flip lifecycle,
# woodcutting),
# not just --selftest. A nested validator spawned by the
# _end_to_end_bad_mapping fixture must not re-run the suites (that
# would recurse); the fixture's child scans the tree only.
if not os.environ.get("CODEX_SKIP_REGRESSION_SUITES"):
errors += run_selftest(verbose=False)
errors += run_flip_selftest(verbose=False)
errors += run_mining_selftest(verbose=False)
errors += run_woodcutting_selftest(verbose=False)
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")
# 2b. structured mining-model gates (pickaxe gates + run-mode contract)
errors += mining_model_errors(rel, data)
# 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)
# 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)
# GE-flip lifecycle gates (per-slot async model, 4h limit ledger vs
# offer expiry, typed ceiling vs realized scenario bounds).
errors += ge_lifecycle_errors(rel, data)
# one-cast worked-example + loss-reserve/fill-price regressions
# (merged from low-alch-members PR #8)
errors += alch_worked_example_regression(rel, data, path.read_text())
errors += alch_accounting_regression(rel, data, path.read_text())
# Woodcutting structured gates (owner 10288 / governor 10398
# shape: canonical axe table, deterministic per-mode ladders,
# per-selected-row dynamic rates with 27/28 capacity, drop-mode
# realized GP zero, prepare/failed wiring).
errors += woodcutting_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())
Reference in New Issue View Git Blame Copy Permalink