#!/usr/bin/env python3 """OSRS Codex validator — the gate for every codex/**.yaml entry. Runs: 1. JSON-Schema validation (schemas/entry.schema.json) 2. Cross-field consistency checks the schema can't express: - state_machine `next`/transition targets must be declared states - `initial` must be a declared state - duplicate action ids / step names - status:verified entries with `verified:` older than --stale-days => WARNING (entry should be re-verified, not dropped) - `best_practice` lines whose source starts with "researched" => WARNING (citation TODO — file a follow-up issue) - GE-margin entries (tags include flipping and/or merchanting — the unambiguous margin signals): the 2% GE convenience fee is rounded DOWN to whole coins per item and capped at 5,000,000 GP per item, so net_per_cycle MUST equal income - min(floor(0.02*income), 5_000_000) - cost. Fractional-tax arithmetic (0.98*sell - buy) and uncapped tax (6M on a 300M sale) are errors. NOTE: the generic `ge` tag alone does NOT gate this rule — a training/processing/procurement entry may legitimately carry `ge` while its income/cost model has nothing to do with flip tax. - 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. - 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. 3. Global id uniqueness across the whole codex tree. Exit codes: 0 = clean (warnings ok), 1 = validation errors. Usage: python3 tools/validate.py [--stale-days 30] [--selftest] --selftest: run the GE floor-tax regression fixtures only (no tree scan) """ import datetime import json import math import re import sys from pathlib import Path import jsonschema import yaml ROOT = Path(__file__).resolve().parent.parent SCHEMA = json.loads((ROOT / "schemas" / "entry.schema.json").read_text()) CODEX = ROOT / "codex" STALE_DAYS = 30 def load_entries(): out = [] for p in CODEX.rglob("*.yaml"): if p.name.startswith("_"): # dataset files (e.g. _f2p-alch-catalog.yaml) continue out.append(p) return sorted(out) MARGIN_TAGS = {"flipping", "merchanting"} def ge_floor_tax_error(rel, data): """GE floor-tax gate for a single entry. Applies ONLY to the unambiguous margin tags (flipping/merchanting). The generic `ge` tag is not proof that income_per_cycle is a taxed GE resale: a training/processing/procurement entry may carry `ge` while modeling unrelated income/cost, so it must not be gated by this rule. Returns an error message, or None when the entry is not gated or its floored-tax arithmetic is correct. Wiki: "The 2% tax rounds down to the nearest whole number, so items sold for below 50 coins have a tax of less than one coin" — per-item net MUST use the floored tax, never 0.98*sell - buy. The fee is also capped at 5,000,000 GP per item (per-item cap, documented in the flip-candidates best_practice/risk), so tax = min(floor(0.02*sell), 5_000_000). Owner decision 2026-08-28: one global formula for all entries (option A), no per-entry tax-model discriminator. """ tags = {str(t).lower() for t in data.get("tags", [])} if not tags & MARGIN_TAGS: return None eco = data.get("economy") or {} if not all(k in eco for k in ("income_per_cycle", "cost_per_cycle", "net_per_cycle")): return None sell, buy = eco["income_per_cycle"], eco["cost_per_cycle"] tax = min(math.floor(sell * 0.02), 5_000_000) expected = sell - tax - buy if eco["net_per_cycle"] != expected: return ( f"{rel}: net_per_cycle {eco['net_per_cycle']} != capped floored-tax " f"model {expected} (sell {sell} - min(floor({sell}*0.02), 5_000_000)" f"={tax} - buy {buy}); " f"GE tax rounds down per item and is capped at 5,000,000 GP per item " f"— fractional 0.98*sell - buy and uncapped tax are invalid") return None # --- 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 def run_selftest(verbose=False): """Regression fixtures for the GE floor-tax + lifecycle gates; 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 main(): global STALE_DAYS if "--stale-days" in sys.argv: STALE_DAYS = int(sys.argv[sys.argv.index("--stale-days") + 1]) errors, warnings = [], [] if "--selftest" in sys.argv: errors += run_selftest(verbose=True) print(f"RESULT: selftest, {len(errors)} errors, 0 warnings") return 1 if errors else 0 # The fix itself is gated: regression fixtures run on every invocation. errors += run_selftest(verbose=False) entries = load_entries() if not entries: print("ERROR: no codex entries found under codex/") return 1 ids = {} today = datetime.date.today() for path in entries: rel = str(path.relative_to(ROOT)) try: data = yaml.safe_load(path.read_text()) except yaml.YAMLError as e: errors.append(f"{rel}: YAML parse error: {e}") continue # 1. schema v = jsonschema.Draft7Validator(SCHEMA) for err in sorted(v.iter_errors(data), key=lambda e: list(e.path)): loc = "/".join(str(p) for p in err.path) or "(root)" errors.append(f"{rel} [{loc}]: {err.message}") if errors and errors[-1].startswith(rel): # schema already broke; cross-checks on a broken doc are noise continue eid = data.get("id", rel) if eid in ids: errors.append(f"{rel}: duplicate entry id '{eid}' (also in {ids[eid]})") ids[eid] = rel # 2. state machine consistency sm = data.get("state_machine", {}) states = sm.get("states", {}) declared = set(states) terminals = set(sm.get("terminal", [])) for sname, sdef in states.items(): if sname not in terminals and not sdef.get("next"): errors.append(f"{rel}: non-terminal state '{sname}' has no 'next'") for nxt in sdef.get("next", []): if nxt not in declared: errors.append(f"{rel}: state '{sname}' next->'{nxt}' not declared") if sm.get("initial") not in declared: errors.append(f"{rel}: initial state '{sm.get('initial')}' not declared") for t in sm.get("transitions", []): if t["from"] not in declared: errors.append(f"{rel}: transition from '{t['from']}' not declared") if t["to"] not in declared: errors.append(f"{rel}: transition to '{t['to']}' not declared") for term in sm.get("terminal", []): if term not in declared: errors.append(f"{rel}: terminal '{term}' not declared") # action id/step uniqueness a_ids = [a["id"] for a in data["actions"]] a_steps = [a["step"] for a in data["actions"]] if len(a_ids) != len(set(a_ids)): errors.append(f"{rel}: duplicate action ids") if len(a_steps) != len(set(a_steps)): errors.append(f"{rel}: duplicate action step names") # GE-margin floor-tax arithmetic (see ge_floor_tax_error): scoped to # the unambiguous margin tags flipping/merchanting. The generic `ge` # tag alone must not gate this rule — alchemy/processing/procurement # entries can carry `ge` with income unrelated to a taxed GE resale. err = ge_floor_tax_error(rel, data) if err: errors.append(err) # 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) # 3. staleness / research TODOs if data.get("status") == "verified": try: vdate = datetime.date.fromisoformat(data["verified"]) age = (today - vdate).days if age > STALE_DAYS: warnings.append( f"{rel}: status=verified but facts are {age}d old " f"(> {STALE_DAYS}d) — re-verify or drop to status:stale") except ValueError: pass for line in data.get("best_practice", {}).get("equipment", []) + \ data.get("best_practice", {}).get("techniques", []): if str(line.get("source", "")).startswith("researched"): warnings.append( f"{rel}: best_practice '{line.get('item') or line.get('name')}' " f"carries a 'researched' source — citation TODO, file follow-up") # report for w in warnings: print(f"WARN {w}") for e in errors: print(f"ERROR {e}") print(f"RESULT: {len(entries)} entries, {len(errors)} errors, {len(warnings)} warnings") return 1 if errors else 0 if __name__ == "__main__": sys.exit(main())