pbwraith/osrs-codex
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
02b366c7ef
osrs-codex/tools/validate.py
Claude 02b366c7ef flip-candidates: per-slot async sub-state lifecycle (governor 10323 bounce-fix 5) 
- slot_state_machine: canonical 8 sub-states (free/buy-open/
  buy-cancelled-awaiting-collect/ready-to-sell/sell-open/
  sell-cancelled-awaiting-collect/ready-to-requote/closed) + global
  scheduler pass that never advances unrelated slots
- ge_cancel moves filled items/refunds/unsold units/coins to the GE
  collection box; ge_collect + ledger reconciliation before reuse
- 4h buy-limit ledger (limit_used/first_fill_ts/limit_resets_at) distinct
  from offer expiry; cancellation never renews the limit
- per-slot conservation invariants w/ committed validator fixtures
- economy.ceiling_gp_per_hour typed ceiling, gp_per_hour_low/high as
  documented realized scenario bounds (no uncited multipliers)
- capped/floored 2% tax + 4 fixtures preserved (owner decision)
- action-grammar: ge_cancel/ge_collect rows corrected
2026-08-28 04:01:20 -05:00

578 lines
28 KiB
Python
Raw 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)
- 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 — cancelling never renews the limit; every slot is an
independent asynchronous sub-state machine (canonical set:
free / buy-open / buy-cancelled-awaiting-collect / ready-to-sell /
sell-open / sell-cancelled-awaiting-collect / ready-to-requote /
closed) driven by a global scheduler pass that never touches
unrelated slots; ge_cancel moves filled items/refunds/proceeds/
unsold units to the GE collection box for a separate ge_collect;
per-slot conservation (quoted == still_offered + collected +
refunded + sold; no negative/duplicated quantities) is provable;
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) ---
# 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.
SLOT_SUB_STATES = [
"free", # no offer, nothing held
"buy-open", # resting buy offer (partial fills stay attached)
"buy-cancelled-awaiting-collect", # aborted buy: assets sit in the GE collection box
"ready-to-sell", # units in hand, sell edge positive
"sell-open", # resting sell offer (partial sells tracked)
"sell-cancelled-awaiting-collect", # aborted sell: proceeds + unsold units in the box
"ready-to-requote", # collected units/refunds; requote or abandon
"closed", # slot finished, ledger reconciled, nothing held
]
# 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-cancelled-awaiting-collect": "collect",
"ready-to-sell": "sell",
"sell-open": None, # resting; fill/abort decided by policies
"sell-cancelled-awaiting-collect": "collect",
"ready-to-requote": "requote",
"closed": None,
}
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.
"""
return SLOT_DISPATCH.get(sub_state)
def conservation_violations(record):
"""Per-slot conservation invariants (governor 10323 finding 4).
record: dict with item buckets (still_offered, collected_items,
refunded, sold — each a {item: qty} map), quoted ({item: qty}), and cash
buckets (cash_quoted, still_offered_cash, collected_cash,
refunded_cash). Invariants:
- every quantity >= 0;
- per item: quoted == still_offered + collected_items + refunded + sold
(the buckets are a partition of the quote's units — no losses, no
duplicated, no negative quantities);
- physically impossible co-occurrences: units cannot sit in a live
offer AND already be sold or cancelled-refunded (cancel is atomic),
so an item may not appear in both {still_offered, sold} or both
{still_offered, refunded} — partial fills legitimately span
{still_offered, collected_items} until the offer is closed;
- cash: cash_quoted == still_offered_cash + collected_cash +
refunded_cash.
Returns a list of violation strings (empty = conserved).
"""
out = []
buckets = ["still_offered", "collected_items", "refunded", "sold"]
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} != still_offered + collected + "
f"refunded + sold ({total})")
# Physically impossible co-occurrences (the buckets are a
# partition of the quote's units; cancel is atomic and units in
# a live offer cannot already be sold/cancelled):
# still_offered + sold — units cannot be both in a live
# offer and already sold
# still_offered + refunded — a cancelled offer has no live
# remainder left in the offer
in_b = {b for b in buckets if item in record.get(b, {})}
for pair in (("still_offered", "sold"), ("still_offered", "refunded")):
if pair[0] in in_b and pair[1] in in_b:
out.append(
f"item '{item}' appears in multiple buckets "
f"({pair[0]} + {pair[1]} — physically impossible: "
f"units cannot be in a live offer and a terminal "
f"bucket at the same time)")
for c in ("cash_quoted", "still_offered_cash", "collected_cash", "refunded_cash"):
if record.get(c, 0) < 0:
out.append(f"{c} is negative ({record[c]})")
if record.get("cash_quoted", 0) != (
record.get("still_offered_cash", 0)
+ record.get("collected_cash", 0)
+ record.get("refunded_cash", 0)):
out.append(
f"cash quoted {record.get('cash_quoted')} != still_offered "
f"{record.get('still_offered_cash')} + collected "
f"{record.get('collected_cash')} + refunded {record.get('refunded_cash')}")
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
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;
- 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 state != "closed" and not trans.get(state):
errors.append(f"{rel}: slot sub-state '{state}' has no transitions "
f"(closed is the only sink)")
if trans.get("closed"):
errors.append(f"{rel}: 'closed' is a sink — it must have no outgoing transitions")
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")
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-sell-collect / no-fill-buy case and prove no
# unrelated slot is closed or advanced.
slot_states = {"arrowtips": "buy-open", # partial fill 4,000/10,000 — resting
"iron": "ready-to-sell", # full fill — sell this pass
"cosmic": "buy-open"} # no fill, within quote age — resting
dispatched = {name: scheduler_dispatch(s) for name, s in slot_states.items()}
ok = dispatched == {"arrowtips": None, "iron": "sell", "cosmic": None}
check("worked example: partial-buy + full-sell + no-fill slots dispatch only the iron slot to sell; 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)
# --- conservation invariants (finding 4) ---
# Balanced snapshot mirroring the worked example: 10,000 Rune arrowtips
# quoted, 4,000 filled (units in the player's possession), 6,000 still
# resting in the live offer; plus an independent cash quote (2,230 GP)
# fully committed in live buy offers.
clean_rec = {
"quoted": {"Rune arrowtips": 10},
"still_offered": {"Rune arrowtips": 6},
"collected_items": {"Rune arrowtips": 4},
"refunded": {},
"sold": {},
"cash_quoted": 2230, "still_offered_cash": 2230,
"collected_cash": 0, "refunded_cash": 0,
}
check("conservation: a balanced per-slot ledger (partial fill split across still_offered + collected, partition sums to quoted) has zero violations",
conservation_violations(clean_rec) == [])
dup_rec = dict(clean_rec, quoted={"Rune arrowtips": 10},
still_offered={"Rune arrowtips": 4},
sold={"Rune arrowtips": 6})
check("conservation: units both in a live offer AND already sold (impossible pair) is a violation",
any("multiple buckets" in v for v in conservation_violations(dup_rec)))
over_rec = dict(clean_rec, collected_items={"Rune arrowtips": 7})
check("conservation: over-reported buckets break the quoted partition (sum 13 != quoted 10)",
any("quoted 10 !=" in v for v in conservation_violations(over_rec)))
neg_rec = dict(clean_rec, sold={"Rune arrowtips": -1})
check("conservation: a negative bucket is a violation",
any("negative" in v for v in conservation_violations(neg_rec)))
cash_rec = dict(clean_rec, collected_cash=900)
check("conservation: cash quoted != still_offered + collected + refunded is a violation",
any("cash quoted" in v for v in conservation_violations(cash_rec)))
# --- 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) ---
margin_tags = ["flipping"]
ssm_ok = {
"sub_states": SLOT_SUB_STATES,
"slot_transitions": {
"free": ["buy-open"],
"buy-open": ["ready-to-sell", "buy-cancelled-awaiting-collect"],
"buy-cancelled-awaiting-collect": ["ready-to-sell", "free"],
"ready-to-sell": ["sell-open"],
"sell-open": ["closed", "sell-cancelled-awaiting-collect"],
"sell-cancelled-awaiting-collect": ["ready-to-requote", "closed"],
"ready-to-requote": ["buy-open", "sell-open", "free"],
"closed": [],
},
"slot_transition_whens": [
"free -> buy-open: quote placed",
"buy-open -> ready-to-sell: offer fully filled",
"buy-open -> buy-cancelled-awaiting-collect: abort (stale edge / quote age)",
"buy-cancelled-awaiting-collect -> ready-to-sell: collected, sell edge positive",
"buy-cancelled-awaiting-collect -> free: collected, full refund, slot freed",
"ready-to-sell -> sell-open: sell offer placed",
"sell-open -> closed: fully filled and proceeds collected",
"sell-open -> sell-cancelled-awaiting-collect: abort",
"sell-cancelled-awaiting-collect -> ready-to-requote: collected, requote",
"sell-cancelled-awaiting-collect -> closed: collected, slot closed",
"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": "resting offers re-checked every 600 s; stale edges requoted",
"persisted_state": "per-slot record: sub-state + quoted/filled/sold/collected",
"limit_ledger": "limit_used + first_fill_ts + limit_resets_at, independent of offer expiry",
"conservation": "quoted == still_offered + collected + refunded + sold",
"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 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))
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())
Reference in New Issue View Git Blame Copy Permalink