don
/
neon-sprawl
mirror of https://github.com/ViPro-Technologies/neon-sprawl
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
neon-sprawl/scripts/validate_content.py

2383 lines
99 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Validate content catalogs against JSON Schema (CI + local).
Validates:
- skill catalogs: content/skills/*_skills.json rows vs content/schemas/skill-def.schema.json
- level curves: content/skills/*_level_curve.json vs content/schemas/level-curve.schema.json
- mastery catalogs: content/mastery/*_mastery.json vs content/schemas/mastery-catalog.schema.json
(post-schema: tierIndex unique and sequential 1..N per track, aligned with server boot — NEO-46)
- item catalogs: content/items/*_items.json rows vs content/schemas/item-def.schema.json (NEO-50)
- resource node catalogs: content/resource-nodes/*_resource_nodes.json vs resource-node-def.schema.json (NEO-57)
- resource yield catalogs: content/resource-nodes/*_resource_yields.json vs resource-yield-row.schema.json (NEO-57)
- recipe catalogs: content/recipes/*_recipes.json rows vs content/schemas/recipe-def.schema.json (NEO-65)
- ability catalogs: content/abilities/*_abilities.json rows vs content/schemas/ability-def.schema.json (NEO-76)
- npc behavior catalogs: content/npc-behaviors/*_npc_behaviors.json rows vs content/schemas/npc-behavior-def.schema.json (NEO-87)
- reward table catalogs: content/reward-tables/*_reward_tables.json rows vs content/schemas/reward-table.schema.json (NEO-100)
- encounter catalogs: content/encounters/*_encounters.json rows vs content/schemas/encounter-def.schema.json (NEO-100)
- quest catalogs: content/quests/*_quests.json rows vs content/schemas/quest-def.schema.json (NEO-112);
optional completionRewardBundle via quest-reward-bundle.schema.json (NEO-124);
optional factionGateRules + reputationGrants (NEO-133)
- faction catalogs: content/factions/*_factions.json rows vs content/schemas/faction-def.schema.json (NEO-133)
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
from jsonschema import Draft202012Validator
from referencing import Registry, Resource
REPO_ROOT = Path(__file__).resolve().parent.parent
SKILL_SCHEMA = REPO_ROOT / "content/schemas/skill-def.schema.json"
LEVEL_CURVE_SCHEMA = REPO_ROOT / "content/schemas/level-curve.schema.json"
MASTERY_SCHEMA = REPO_ROOT / "content/schemas/mastery-catalog.schema.json"
ITEM_SCHEMA = REPO_ROOT / "content/schemas/item-def.schema.json"
RESOURCE_NODE_SCHEMA = REPO_ROOT / "content/schemas/resource-node-def.schema.json"
RESOURCE_YIELD_ROW_SCHEMA = REPO_ROOT / "content/schemas/resource-yield-row.schema.json"
RECIPE_IO_ROW_SCHEMA = REPO_ROOT / "content/schemas/recipe-io-row.schema.json"
RECIPE_SCHEMA = REPO_ROOT / "content/schemas/recipe-def.schema.json"
ABILITY_SCHEMA = REPO_ROOT / "content/schemas/ability-def.schema.json"
NPC_BEHAVIOR_SCHEMA = REPO_ROOT / "content/schemas/npc-behavior-def.schema.json"
REWARD_GRANT_ROW_SCHEMA = REPO_ROOT / "content/schemas/reward-grant-row.schema.json"
REWARD_TABLE_SCHEMA = REPO_ROOT / "content/schemas/reward-table.schema.json"
ENCOUNTER_DEF_SCHEMA = REPO_ROOT / "content/schemas/encounter-def.schema.json"
QUEST_OBJECTIVE_DEF_SCHEMA = REPO_ROOT / "content/schemas/quest-objective-def.schema.json"
QUEST_STEP_DEF_SCHEMA = REPO_ROOT / "content/schemas/quest-step-def.schema.json"
QUEST_DEF_SCHEMA = REPO_ROOT / "content/schemas/quest-def.schema.json"
QUEST_REWARD_BUNDLE_SCHEMA = REPO_ROOT / "content/schemas/quest-reward-bundle.schema.json"
QUEST_SKILL_XP_GRANT_SCHEMA = REPO_ROOT / "content/schemas/quest-skill-xp-grant.schema.json"
FACTION_DEF_SCHEMA = REPO_ROOT / "content/schemas/faction-def.schema.json"
FACTION_GATE_RULE_SCHEMA = REPO_ROOT / "content/schemas/faction-gate-rule.schema.json"
REPUTATION_GRANT_ROW_SCHEMA = REPO_ROOT / "content/schemas/reputation-grant-row.schema.json"
SKILLS_DIR = REPO_ROOT / "content/skills"
MASTERY_DIR = REPO_ROOT / "content/mastery"
ITEMS_DIR = REPO_ROOT / "content/items"
RESOURCE_NODES_DIR = REPO_ROOT / "content/resource-nodes"
RECIPES_DIR = REPO_ROOT / "content/recipes"
ABILITIES_DIR = REPO_ROOT / "content/abilities"
NPC_BEHAVIORS_DIR = REPO_ROOT / "content/npc-behaviors"
REWARD_TABLES_DIR = REPO_ROOT / "content/reward-tables"
ENCOUNTERS_DIR = REPO_ROOT / "content/encounters"
QUESTS_DIR = REPO_ROOT / "content/quests"
FACTIONS_DIR = REPO_ROOT / "content/factions"
# Slice 1 prototype lock (NEO-33): exact ids + category coverage after schema passes.
PROTOTYPE_SLICE1_SKILL_IDS = frozenset({"salvage", "refine", "intrusion"})
# Slice 1 prototype lock (NEO-50): exact item ids + prototypeRole coverage after schema passes.
PROTOTYPE_SLICE1_ITEM_IDS = frozenset(
{
"scrap_metal_bulk",
"refined_plate_stock",
"field_stim_mk0",
"survey_drone_kit",
"contract_handoff_token",
"prototype_armor_shell",
}
)
PROTOTYPE_SLICE1_ITEM_ROLES = frozenset(
{"material", "intermediate", "consumable", "utility", "quest_token", "equip_stub"}
)
# Slice 4 prototype lock (NEO-45): exactly one salvage track across all mastery files.
PROTOTYPE_SLICE4_SALVAGE_SKILL_ID = "salvage"
# Slice 2 prototype lock (NEO-57): exact nodeDefIds + gatherLens coverage after schema passes.
# Keep in sync with server loader when E3M1-02 lands.
PROTOTYPE_SLICE2_NODE_IDS = frozenset(
{
"prototype_resource_node_alpha",
"prototype_subsurface_vein_beta",
"prototype_bio_mat_gamma",
"prototype_urban_bulk_delta",
}
)
PROTOTYPE_SLICE2_GATHER_LENSES = frozenset(
{"consumer_salvage", "subsurface", "bio", "urban_bulk"}
)
PROTOTYPE_SLICE2_YIELD_ITEM_IDS = frozenset({"scrap_metal_bulk"})
# Slice 3 prototype lock (NEO-65): exact recipe ids + recipeKind coverage after schema passes.
# Keep in sync with E3.M2 freeze table and future NEO-66 server loader.
PROTOTYPE_SLICE3_RECIPE_IDS = frozenset(
{
"refine_scrap_standard",
"refine_scrap_efficient",
"make_field_stim_mk0",
"make_field_stim_batch",
"make_prototype_armor",
"make_armor_quick",
"make_survey_drone_kit",
"make_contract_token",
}
)
PROTOTYPE_SLICE3_REFINE_SKILL_ID = "refine"
# Epic 5 Slice 1 prototype lock (NEO-76): exact ability ids after schema passes.
# Keep in sync with E5.M1 freeze table and future NEO-77 server loader.
PROTOTYPE_E5M1_ABILITY_IDS = frozenset(
{
"prototype_pulse",
"prototype_guard",
"prototype_dash",
"prototype_burst",
}
)
# Epic 5 Slice 2 NPC attack abilities (NEO-98): exact ids after schema passes.
PROTOTYPE_E5M2_NPC_ATTACK_ABILITY_IDS = frozenset(
{
"prototype_npc_melee_strike",
"prototype_npc_ranged_shot",
"prototype_npc_elite_slam",
}
)
PROTOTYPE_ABILITY_CATALOG_IDS = PROTOTYPE_E5M1_ABILITY_IDS | PROTOTYPE_E5M2_NPC_ATTACK_ABILITY_IDS
# Epic 5 Slice 2 prototype lock (NEO-87): exact npc behavior ids after schema passes.
# Keep in sync with E5.M2 freeze table and future NEO-88 server loader.
PROTOTYPE_E5M2_NPC_BEHAVIOR_IDS = frozenset(
{
"prototype_melee_pressure",
"prototype_ranged_control",
"prototype_elite_mini_boss",
}
)
# Epic 5 Slice 2 NPC instance ids (server PrototypeNpcRegistry); CI cross-ref only (NEO-100).
PROTOTYPE_E5M2_NPC_INSTANCE_IDS = frozenset(
{
"prototype_npc_melee",
"prototype_npc_ranged",
"prototype_npc_elite",
}
)
# Epic 5 Slice 3 prototype lock (NEO-100): exact encounter + reward table ids after schema passes.
PROTOTYPE_E5M3_ENCOUNTER_IDS = frozenset({"prototype_combat_pocket"})
PROTOTYPE_E5M3_REWARD_TABLE_IDS = frozenset({"prototype_combat_pocket_clear"})
# Frozen fixed grants for prototype_combat_pocket_clear (keep in sync with E5.M3 freeze table).
PROTOTYPE_E5M3_REWARD_GRANTS: dict[str, int] = {
"scrap_metal_bulk": 10,
"contract_handoff_token": 1,
}
# Epic 7 Slice 1 prototype lock (NEO-112): exact quest ids after schema passes.
# Keep in sync with E7.M1 freeze table and future NEO-113 server loader.
PROTOTYPE_E7M1_QUEST_IDS = frozenset(
{
"prototype_quest_gather_intro",
"prototype_quest_refine_intro",
"prototype_quest_combat_intro",
"prototype_quest_operator_chain",
}
)
PROTOTYPE_E7M1_CHAIN_QUEST_ID = "prototype_quest_operator_chain"
PROTOTYPE_E7M1_CHAIN_TERMINAL_ITEM_ID = "contract_handoff_token"
# Epic 7 Slice 2 prototype lock (NEO-124): completionRewardBundle on four frozen quests.
# Keep in sync with E7.M2 freeze table and future NEO-125 server loader.
PROTOTYPE_E7M2_MISSION_REWARD_SOURCE_KIND = "mission_reward"
PROTOTYPE_E7M2_COMPLETION_BUNDLES: dict[str, dict] = {
"prototype_quest_gather_intro": {
"itemGrants": [],
"skillXpGrants": [{"skillId": "salvage", "amount": 25}],
},
"prototype_quest_refine_intro": {
"itemGrants": [],
"skillXpGrants": [{"skillId": "refine", "amount": 25}],
},
"prototype_quest_combat_intro": {
"itemGrants": [],
"skillXpGrants": [{"skillId": "salvage", "amount": 25}],
},
"prototype_quest_operator_chain": {
"itemGrants": [{"itemId": "survey_drone_kit", "quantity": 1}],
"skillXpGrants": [{"skillId": "salvage", "amount": 50}],
},
}
# Epic 7 Slice 3 prototype lock (NEO-133): five-quest roster + faction catalog.
# Keep in sync with E7.M3 freeze table and future NEO-134 server loader.
PROTOTYPE_E7M3_FACTION_IDS = frozenset(
{
"prototype_faction_grid_operators",
"prototype_faction_rust_collective",
}
)
PROTOTYPE_E7M3_FACTION_FREEZE: dict[str, dict] = {
"prototype_faction_grid_operators": {
"displayName": "Grid Operators",
"minStanding": -100,
"maxStanding": 100,
"neutralStanding": 0,
},
"prototype_faction_rust_collective": {
"displayName": "Rust Collective",
"minStanding": -100,
"maxStanding": 100,
"neutralStanding": 0,
},
}
PROTOTYPE_E7M3_QUEST_IDS = frozenset(
{
"prototype_quest_gather_intro",
"prototype_quest_refine_intro",
"prototype_quest_combat_intro",
"prototype_quest_operator_chain",
"prototype_quest_grid_contract",
}
)
PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID = "prototype_quest_grid_contract"
PROTOTYPE_E7M3_GRID_CONTRACT_PREREQ_QUEST_ID = "prototype_quest_operator_chain"
PROTOTYPE_E7M3_GRID_CONTRACT_GATE_FACTION_ID = "prototype_faction_grid_operators"
PROTOTYPE_E7M3_GRID_CONTRACT_MIN_STANDING = 15
PROTOTYPE_E7M3_GRID_CONTRACT_OBJECTIVE_ITEM_ID = "survey_drone_kit"
PROTOTYPE_E7M3_COMPLETION_BUNDLES: dict[str, dict] = {
"prototype_quest_gather_intro": {
"itemGrants": [],
"skillXpGrants": [{"skillId": "salvage", "amount": 25}],
"reputationGrants": [],
},
"prototype_quest_refine_intro": {
"itemGrants": [],
"skillXpGrants": [{"skillId": "refine", "amount": 25}],
"reputationGrants": [],
},
"prototype_quest_combat_intro": {
"itemGrants": [],
"skillXpGrants": [{"skillId": "salvage", "amount": 25}],
"reputationGrants": [],
},
"prototype_quest_operator_chain": {
"itemGrants": [{"itemId": "survey_drone_kit", "quantity": 1}],
"skillXpGrants": [{"skillId": "salvage", "amount": 50}],
"reputationGrants": [
{"factionId": "prototype_faction_grid_operators", "amount": 15}
],
},
"prototype_quest_grid_contract": {
"itemGrants": [],
"skillXpGrants": [],
"reputationGrants": [
{"factionId": "prototype_faction_rust_collective", "amount": 10}
],
},
}
def _normalize_completion_bundle(bundle: dict) -> dict:
"""Normalize grant rows for stable freeze-table comparison."""
item_grants = bundle.get("itemGrants")
skill_xp_grants = bundle.get("skillXpGrants")
return {
"itemGrants": sorted(
[
{"itemId": grant["itemId"], "quantity": grant["quantity"]}
for grant in (item_grants if isinstance(item_grants, list) else [])
if isinstance(grant, dict)
and isinstance(grant.get("itemId"), str)
and isinstance(grant.get("quantity"), int)
],
key=lambda grant: (grant["itemId"], grant["quantity"]),
),
"skillXpGrants": sorted(
[
{"skillId": grant["skillId"], "amount": grant["amount"]}
for grant in (skill_xp_grants if isinstance(skill_xp_grants, list) else [])
if isinstance(grant, dict)
and isinstance(grant.get("skillId"), str)
and isinstance(grant.get("amount"), int)
],
key=lambda grant: (grant["skillId"], grant["amount"]),
),
}
def _normalize_e7m3_completion_bundle(bundle: dict) -> dict:
"""Normalize item, skill XP, and reputation grant rows for E7M3 freeze-table comparison."""
normalized = _normalize_completion_bundle(bundle)
reputation_grants = bundle.get("reputationGrants")
normalized["reputationGrants"] = sorted(
[
{"factionId": grant["factionId"], "amount": grant["amount"]}
for grant in (reputation_grants if isinstance(reputation_grants, list) else [])
if isinstance(grant, dict)
and isinstance(grant.get("factionId"), str)
and isinstance(grant.get("amount"), int)
],
key=lambda grant: (grant["factionId"], grant["amount"]),
)
return normalized
def _prototype_slice1_gate(seen_ids: dict[str, str], id_to_category: dict[str, str]) -> str | None:
"""Return a human-readable error if Slice 1 contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_SLICE1_SKILL_IDS:
return (
"error: prototype Slice 1 expects exactly skill ids "
f"{sorted(PROTOTYPE_SLICE1_SKILL_IDS)!r}, got {sorted(ids)!r}"
)
cats = set(id_to_category.values())
if "gather" not in cats or "tech" not in cats:
return (
"error: prototype Slice 1 requires at least one gather and one tech skill; "
f"categories seen: {sorted(cats)!r}"
)
if "process" not in cats and "make" not in cats:
return (
"error: prototype Slice 1 requires at least one process or make skill; "
f"categories seen: {sorted(cats)!r}"
)
return None
def _tier_index_gate(rel: str, track_index: int, tier_index_values: list[int]) -> str | None:
"""Keep in sync with MasteryCatalogLoader.TryGetTierIndexGateError (NEO-46)."""
if not tier_index_values:
return None
seen: set[int] = set()
for value in tier_index_values:
if value in seen:
return f"error: {rel}: tracks[{track_index}] duplicate tierIndex {value}"
seen.add(value)
expected = set(range(1, len(tier_index_values) + 1))
if seen != expected:
sorted_vals = ", ".join(str(v) for v in sorted(seen))
n = len(tier_index_values)
return (
f"error: {rel}: tracks[{track_index}] tierIndex values must be unique and "
f"sequential 1..{n}, got [{sorted_vals}]"
)
return None
def _validate_mastery_catalogs(
*,
mastery_files: list[Path],
mastery_validator: Draft202012Validator,
skill_ids: frozenset[str],
) -> tuple[int, list[str]]:
"""Validate mastery JSON files. Returns (error_count, track_skill_ids across all files)."""
errors = 0
all_track_skill_ids: list[str] = []
global_perk_ids: dict[str, str] = {}
# Each perkId may appear in at most one branch reference catalog-wide (v1 mutually exclusive trees).
global_referenced_perk_ids: dict[str, str] = {}
for path in mastery_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_errors = 0
for err in sorted(mastery_validator.iter_errors(data), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} {loc}: {err.message}", file=sys.stderr)
schema_errors += 1
errors += 1
if schema_errors > 0:
continue
perks = data.get("perks")
tracks = data.get("tracks")
if not isinstance(perks, dict) or not isinstance(tracks, list):
print(f"error: {rel}: expected top-level 'perks' object and 'tracks' array", file=sys.stderr)
errors += 1
continue
referenced_perk_ids_in_file: set[str] = set()
for perk_key, perk in perks.items():
if not isinstance(perk, dict):
print(f"error: {rel}: perks[{perk_key!r}] must be an object", file=sys.stderr)
errors += 1
continue
pid = perk.get("id")
if not isinstance(pid, str):
continue
if perk_key != pid:
print(
f"error: {rel}: perks map key {perk_key!r} must match PerkDef.id {pid!r}",
file=sys.stderr,
)
errors += 1
prev = global_perk_ids.get(pid)
if prev:
print(f"error: duplicate perk id {pid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
global_perk_ids[pid] = rel
for ti, track in enumerate(tracks):
if not isinstance(track, dict):
print(f"error: {rel}: tracks[{ti}] must be an object", file=sys.stderr)
errors += 1
continue
skill_id = track.get("skillId")
if isinstance(skill_id, str):
all_track_skill_ids.append(skill_id)
if skill_id not in skill_ids:
print(
f"error: {rel}: tracks[{ti}].skillId {skill_id!r} is not a known SkillDef id "
f"(known: {sorted(skill_ids)!r})",
file=sys.stderr,
)
errors += 1
tiers = track.get("tiers")
if not isinstance(tiers, list):
continue
prev_required_level = 0
prev_tier_branch_ids: frozenset[str] | None = None
tier_index_values: list[int] = []
for tier_i, tier in enumerate(tiers):
if not isinstance(tier, dict):
continue
tier_index = tier.get("tierIndex")
if isinstance(tier_index, int):
tier_index_values.append(tier_index)
required_level = tier.get("requiredLevel")
if isinstance(required_level, int) and required_level <= prev_required_level:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}].requiredLevel must be strictly "
f"greater than previous tier ({prev_required_level})",
file=sys.stderr,
)
errors += 1
if isinstance(required_level, int):
prev_required_level = required_level
branches = tier.get("branches")
if not isinstance(branches, list):
continue
tier_branch_ids: list[str] = []
for bi, branch in enumerate(branches):
if not isinstance(branch, dict):
continue
branch_id = branch.get("branchId")
if isinstance(branch_id, str):
if branch_id in tier_branch_ids:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}] duplicate branchId "
f"{branch_id!r}",
file=sys.stderr,
)
errors += 1
tier_branch_ids.append(branch_id)
perk_ids = branch.get("perkIds")
if not isinstance(perk_ids, list):
continue
for perk_id in perk_ids:
if not isinstance(perk_id, str):
continue
if perk_id not in perks:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}].branches[{bi}] "
f"references unknown perkId {perk_id!r}",
file=sys.stderr,
)
errors += 1
prev_ref = global_referenced_perk_ids.get(perk_id)
if prev_ref:
print(
f"error: duplicate perk id reference {perk_id!r} in {prev_ref} and {rel}",
file=sys.stderr,
)
errors += 1
else:
global_referenced_perk_ids[perk_id] = (
f"{rel} tracks[{ti}].tiers[{tier_i}].branches[{bi}]"
)
referenced_perk_ids_in_file.add(perk_id)
tier_branch_set = frozenset(tier_branch_ids)
if prev_tier_branch_ids is not None and tier_branch_set != prev_tier_branch_ids:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}] branchId set "
f"{sorted(tier_branch_set)!r} must match previous tier "
f"{sorted(prev_tier_branch_ids)!r}",
file=sys.stderr,
)
errors += 1
if tier_branch_ids:
prev_tier_branch_ids = tier_branch_set
tier_index_err = _tier_index_gate(rel, ti, tier_index_values)
if tier_index_err:
print(tier_index_err, file=sys.stderr)
errors += 1
for perk_key in perks:
if isinstance(perk_key, str) and perk_key not in referenced_perk_ids_in_file:
print(
f"error: {rel}: perks[{perk_key!r}] is not referenced by any branch perkIds",
file=sys.stderr,
)
errors += 1
return errors, all_track_skill_ids
def _prototype_slice1_item_gate(
seen_ids: dict[str, str],
id_to_role: dict[str, str],
id_to_slot_kind: dict[str, str],
id_to_stack_max: dict[str, int],
) -> str | None:
"""Return a human-readable error if Slice 1 item contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_SLICE1_ITEM_IDS:
return (
"error: prototype Slice 1 expects exactly item ids "
f"{sorted(PROTOTYPE_SLICE1_ITEM_IDS)!r}, got {sorted(ids)!r}"
)
roles = set(id_to_role.values())
if roles != PROTOTYPE_SLICE1_ITEM_ROLES:
return (
"error: prototype Slice 1 requires exactly one row per prototypeRole "
f"{sorted(PROTOTYPE_SLICE1_ITEM_ROLES)!r}, roles seen: {sorted(roles)!r}"
)
equip_stub_id = next((iid for iid, role in id_to_role.items() if role == "equip_stub"), None)
if equip_stub_id is not None:
if id_to_slot_kind.get(equip_stub_id) != "equipment":
return (
f"error: {equip_stub_id!r} (equip_stub) must use inventorySlotKind 'equipment', "
f"got {id_to_slot_kind.get(equip_stub_id)!r}"
)
if id_to_stack_max.get(equip_stub_id) != 1:
return (
f"error: {equip_stub_id!r} (equip_stub) must use stackMax 1, "
f"got {id_to_stack_max.get(equip_stub_id)!r}"
)
return None
def _validate_item_catalogs(
*,
item_files: list[Path],
item_validator: Draft202012Validator,
) -> tuple[int, dict[str, str], dict[str, str], dict[str, str], dict[str, int]]:
"""Validate item JSON files. Returns (error_count, seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_role: dict[str, str] = {}
id_to_slot_kind: dict[str, str] = {}
id_to_stack_max: dict[str, int] = {}
for path in item_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
items = data.get("items")
if not isinstance(items, list):
print(f"error: {rel}: expected top-level 'items' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(items):
if not isinstance(row, dict):
print(f"error: {rel}: items[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(item_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} items[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
iid = row.get("id")
if isinstance(iid, str) and row_schema_errors == 0:
prev = seen_ids.get(iid)
if prev:
print(f"error: duplicate item id {iid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[iid] = rel
role = row.get("prototypeRole")
if isinstance(role, str):
id_to_role[iid] = role
slot_kind = row.get("inventorySlotKind")
if isinstance(slot_kind, str):
id_to_slot_kind[iid] = slot_kind
stack_max = row.get("stackMax")
if isinstance(stack_max, int):
id_to_stack_max[iid] = stack_max
return errors, seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max
def _validate_ability_catalogs(
*,
ability_files: list[Path],
ability_validator: Draft202012Validator,
) -> tuple[int, dict[str, str], dict[str, dict]]:
"""Validate ability JSON files. Returns (error_count, seen_ids, id_to_row)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_row: dict[str, dict] = {}
for path in ability_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
abilities = data.get("abilities")
if not isinstance(abilities, list):
print(f"error: {rel}: expected top-level 'abilities' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(abilities):
if not isinstance(row, dict):
print(f"error: {rel}: abilities[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(ability_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} abilities[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
aid = row.get("id")
if isinstance(aid, str) and row_schema_errors == 0:
prev = seen_ids.get(aid)
if prev:
print(f"error: duplicate ability id {aid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[aid] = rel
id_to_row[aid] = row
return errors, seen_ids, id_to_row
def _prototype_ability_catalog_gate(seen_ids: dict[str, str]) -> str | None:
"""Return a human-readable error if prototype ability catalog contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_ABILITY_CATALOG_IDS:
return (
"error: prototype ability catalog expects exactly ability ids "
f"{sorted(PROTOTYPE_ABILITY_CATALOG_IDS)!r}, got {sorted(ids)!r}"
)
return None
def _validate_npc_behavior_catalogs(
*,
npc_behavior_files: list[Path],
npc_behavior_validator: Draft202012Validator,
) -> tuple[int, dict[str, str], dict[str, dict]]:
"""Validate NPC behavior JSON files. Returns (error_count, seen_ids, id_to_row)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_row: dict[str, dict] = {}
for path in npc_behavior_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
npc_behaviors = data.get("npcBehaviors")
if not isinstance(npc_behaviors, list):
print(f"error: {rel}: expected top-level 'npcBehaviors' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(npc_behaviors):
if not isinstance(row, dict):
print(f"error: {rel}: npcBehaviors[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(npc_behavior_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} npcBehaviors[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
bid = row.get("id")
if isinstance(bid, str) and row_schema_errors == 0:
prev = seen_ids.get(bid)
if prev:
print(f"error: duplicate npc behavior id {bid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[bid] = rel
id_to_row[bid] = row
return errors, seen_ids, id_to_row
def _prototype_e5m2_npc_behavior_gate(seen_ids: dict[str, str]) -> str | None:
"""Return a human-readable error if E5M2 npc behavior contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_E5M2_NPC_BEHAVIOR_IDS:
return (
"error: prototype E5M2 expects exactly npc behavior ids "
f"{sorted(PROTOTYPE_E5M2_NPC_BEHAVIOR_IDS)!r}, got {sorted(ids)!r}"
)
return None
def _prototype_e5m2_npc_behavior_numeric_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if leash/aggro invariants fail, else None."""
for bid, row in id_to_row.items():
aggro = row.get("aggroRadius")
leash = row.get("leashRadius")
if isinstance(aggro, (int, float)) and isinstance(leash, (int, float)):
if not leash > aggro:
return (
f"error: npc behavior {bid!r}: leashRadius {leash} must be > aggroRadius {aggro}"
)
return None
def _prototype_e5m2_npc_behavior_attack_ability_gate(
id_to_row: dict[str, dict],
ability_id_to_row: dict[str, dict],
) -> str | None:
"""Return a human-readable error when attackAbilityId is missing or attackDamage mismatches."""
for bid, row in id_to_row.items():
aid = row.get("attackAbilityId")
if not isinstance(aid, str):
continue
ability = ability_id_to_row.get(aid)
if ability is None:
return (
f"error: npc behavior {bid!r}: attackAbilityId {aid!r} missing from ability catalog"
)
attack_damage = row.get("attackDamage")
base_damage = ability.get("baseDamage")
if isinstance(attack_damage, int) and isinstance(base_damage, int) and attack_damage != base_damage:
return (
f"error: npc behavior {bid!r}: attackDamage {attack_damage} must match "
f"ability {aid!r} baseDamage {base_damage}"
)
return None
def _reward_table_def_validator() -> Draft202012Validator:
"""Build a validator that resolves reward-table $ref to reward-grant-row schema."""
grant_schema = json.loads(REWARD_GRANT_ROW_SCHEMA.read_text(encoding="utf-8"))
table_schema = json.loads(REWARD_TABLE_SCHEMA.read_text(encoding="utf-8"))
registry = Registry().with_resources(
[
(table_schema["$id"], Resource.from_contents(table_schema)),
(grant_schema["$id"], Resource.from_contents(grant_schema)),
]
)
return Draft202012Validator(table_schema, registry=registry)
def _validate_reward_table_catalogs(
*,
reward_table_files: list[Path],
reward_table_validator: Draft202012Validator,
known_item_ids: frozenset[str],
) -> tuple[int, dict[str, str], dict[str, dict]]:
"""Validate reward table JSON files. Returns (error_count, seen_ids, id_to_row)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_row: dict[str, dict] = {}
for path in reward_table_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
reward_tables = data.get("rewardTables")
if not isinstance(reward_tables, list):
print(f"error: {rel}: expected top-level 'rewardTables' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(reward_tables):
if not isinstance(row, dict):
print(f"error: {rel}: rewardTables[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(reward_table_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} rewardTables[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
rid = row.get("id")
if isinstance(rid, str) and row_schema_errors == 0:
prev = seen_ids.get(rid)
if prev:
print(f"error: duplicate reward table id {rid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[rid] = rel
id_to_row[rid] = row
fixed_grants = row.get("fixedGrants")
if isinstance(fixed_grants, list) and row_schema_errors == 0:
seen_grant_item_ids: set[str] = set()
for j, grant in enumerate(fixed_grants):
if not isinstance(grant, dict):
continue
item_id = grant.get("itemId")
if isinstance(item_id, str):
if item_id in seen_grant_item_ids:
print(
f"error: {rel} rewardTables[{i}].fixedGrants[{j}]: "
f"duplicate itemId {item_id!r}",
file=sys.stderr,
)
errors += 1
else:
seen_grant_item_ids.add(item_id)
if item_id not in known_item_ids:
print(
f"error: {rel} rewardTables[{i}].fixedGrants[{j}]: itemId {item_id!r} "
"is not in item catalogs",
file=sys.stderr,
)
errors += 1
return errors, seen_ids, id_to_row
def _prototype_e5m3_reward_table_gate(seen_ids: dict[str, str]) -> str | None:
"""Return a human-readable error if E5M3 reward table contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_E5M3_REWARD_TABLE_IDS:
return (
"error: prototype E5M3 expects exactly reward table ids "
f"{sorted(PROTOTYPE_E5M3_REWARD_TABLE_IDS)!r}, got {sorted(ids)!r}"
)
return None
def _prototype_e5m3_reward_grant_content_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if frozen grant quantities fail, else None."""
row = id_to_row.get("prototype_combat_pocket_clear")
if row is None:
return None
fixed_grants = row.get("fixedGrants")
if not isinstance(fixed_grants, list):
return None
grants: dict[str, int] = {}
for grant in fixed_grants:
if not isinstance(grant, dict):
continue
item_id = grant.get("itemId")
quantity = grant.get("quantity")
if isinstance(item_id, str) and isinstance(quantity, int):
grants[item_id] = quantity
if grants != PROTOTYPE_E5M3_REWARD_GRANTS:
return (
"error: prototype_combat_pocket_clear fixedGrants must match "
f"{PROTOTYPE_E5M3_REWARD_GRANTS!r}, got {grants!r}"
)
return None
def _validate_encounter_catalogs(
*,
encounter_files: list[Path],
encounter_validator: Draft202012Validator,
known_reward_table_ids: frozenset[str],
) -> tuple[int, dict[str, str], dict[str, dict]]:
"""Validate encounter JSON files. Returns (error_count, seen_ids, id_to_row)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_row: dict[str, dict] = {}
for path in encounter_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
encounters = data.get("encounters")
if not isinstance(encounters, list):
print(f"error: {rel}: expected top-level 'encounters' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(encounters):
if not isinstance(row, dict):
print(f"error: {rel}: encounters[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(encounter_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} encounters[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
eid = row.get("id")
if isinstance(eid, str) and row_schema_errors == 0:
prev = seen_ids.get(eid)
if prev:
print(f"error: duplicate encounter id {eid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[eid] = rel
id_to_row[eid] = row
reward_table_id = row.get("rewardTableId")
if isinstance(reward_table_id, str) and reward_table_id not in known_reward_table_ids:
print(
f"error: {rel} encounters[{i}]: rewardTableId {reward_table_id!r} "
"is not in reward table catalogs",
file=sys.stderr,
)
errors += 1
return errors, seen_ids, id_to_row
def _prototype_e5m3_encounter_gate(seen_ids: dict[str, str]) -> str | None:
"""Return a human-readable error if E5M3 encounter contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_E5M3_ENCOUNTER_IDS:
return (
"error: prototype E5M3 expects exactly encounter ids "
f"{sorted(PROTOTYPE_E5M3_ENCOUNTER_IDS)!r}, got {sorted(ids)!r}"
)
return None
def _prototype_e5m3_encounter_cross_ref_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if NPC instance or reward table cross-refs fail, else None."""
for eid, row in id_to_row.items():
npc_ids = row.get("requiredNpcInstanceIds")
if isinstance(npc_ids, list):
npc_set = frozenset(str(x) for x in npc_ids if isinstance(x, str))
if npc_set != PROTOTYPE_E5M2_NPC_INSTANCE_IDS:
return (
f"error: encounter {eid!r}: requiredNpcInstanceIds must be exactly "
f"{sorted(PROTOTYPE_E5M2_NPC_INSTANCE_IDS)!r}, got {sorted(npc_set)!r}"
)
return None
def _quest_def_validator() -> Draft202012Validator:
"""Build a validator that resolves quest-def $ref to step, objective, bundle, and gate schemas."""
objective_schema = json.loads(QUEST_OBJECTIVE_DEF_SCHEMA.read_text(encoding="utf-8"))
step_schema = json.loads(QUEST_STEP_DEF_SCHEMA.read_text(encoding="utf-8"))
def_schema = json.loads(QUEST_DEF_SCHEMA.read_text(encoding="utf-8"))
bundle_schema = json.loads(QUEST_REWARD_BUNDLE_SCHEMA.read_text(encoding="utf-8"))
skill_xp_grant_schema = json.loads(QUEST_SKILL_XP_GRANT_SCHEMA.read_text(encoding="utf-8"))
grant_row_schema = json.loads(REWARD_GRANT_ROW_SCHEMA.read_text(encoding="utf-8"))
faction_gate_rule_schema = json.loads(FACTION_GATE_RULE_SCHEMA.read_text(encoding="utf-8"))
reputation_grant_row_schema = json.loads(REPUTATION_GRANT_ROW_SCHEMA.read_text(encoding="utf-8"))
registry = Registry().with_resources(
[
(def_schema["$id"], Resource.from_contents(def_schema)),
(step_schema["$id"], Resource.from_contents(step_schema)),
(objective_schema["$id"], Resource.from_contents(objective_schema)),
(bundle_schema["$id"], Resource.from_contents(bundle_schema)),
(skill_xp_grant_schema["$id"], Resource.from_contents(skill_xp_grant_schema)),
(grant_row_schema["$id"], Resource.from_contents(grant_row_schema)),
(faction_gate_rule_schema["$id"], Resource.from_contents(faction_gate_rule_schema)),
(reputation_grant_row_schema["$id"], Resource.from_contents(reputation_grant_row_schema)),
]
)
return Draft202012Validator(def_schema, registry=registry)
def _validate_faction_catalogs(
*,
faction_files: list[Path],
faction_validator: Draft202012Validator,
) -> tuple[int, dict[str, str], dict[str, dict]]:
"""Validate faction JSON files. Returns (error_count, seen_ids, id_to_row)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_row: dict[str, dict] = {}
for path in faction_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
factions = data.get("factions")
if not isinstance(factions, list):
print(f"error: {rel}: expected top-level 'factions' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(factions):
if not isinstance(row, dict):
print(f"error: {rel}: factions[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(faction_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} factions[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
fid = row.get("id")
if isinstance(fid, str) and row_schema_errors == 0:
prev = seen_ids.get(fid)
if prev:
print(f"error: duplicate faction id {fid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[fid] = rel
id_to_row[fid] = row
return errors, seen_ids, id_to_row
def _validate_quest_catalogs(
*,
quest_files: list[Path],
quest_validator: Draft202012Validator,
) -> tuple[int, dict[str, str], dict[str, dict]]:
"""Validate quest JSON files. Returns (error_count, seen_ids, id_to_row)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_row: dict[str, dict] = {}
for path in quest_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
quests = data.get("quests")
if not isinstance(quests, list):
print(f"error: {rel}: expected top-level 'quests' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(quests):
if not isinstance(row, dict):
print(f"error: {rel}: quests[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(quest_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} quests[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
qid = row.get("id")
if isinstance(qid, str) and row_schema_errors == 0:
prev = seen_ids.get(qid)
if prev:
print(f"error: duplicate quest id {qid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[qid] = rel
id_to_row[qid] = row
steps = row.get("steps")
if isinstance(steps, list):
step_ids_seen: set[str] = set()
obj_ids_seen: set[str] = set()
for si, step in enumerate(steps):
if not isinstance(step, dict):
continue
sid = step.get("id")
if isinstance(sid, str):
if sid in step_ids_seen:
print(
f"error: {rel} quests[{i}] steps[{si}]: duplicate step id {sid!r}",
file=sys.stderr,
)
errors += 1
else:
step_ids_seen.add(sid)
objectives = step.get("objectives")
if isinstance(objectives, list):
for oi, obj in enumerate(objectives):
if not isinstance(obj, dict):
continue
oid = obj.get("id")
if isinstance(oid, str):
if oid in obj_ids_seen:
print(
f"error: {rel} quests[{i}] steps[{si}] objectives[{oi}]: "
f"duplicate objective id {oid!r} within quest",
file=sys.stderr,
)
errors += 1
else:
obj_ids_seen.add(oid)
return errors, seen_ids, id_to_row
def _prototype_e7m3_quest_roster_gate(seen_ids: dict[str, str]) -> str | None:
"""Return a human-readable error if E7M3 five-quest roster fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_E7M3_QUEST_IDS:
return (
"error: prototype E7M3 expects exactly quest ids "
f"{sorted(PROTOTYPE_E7M3_QUEST_IDS)!r}, got {sorted(ids)!r}"
)
return None
def _prototype_e7m1_quest_gate(seen_ids: dict[str, str]) -> str | None:
"""Return a human-readable error if E7M1 quest contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_E7M1_QUEST_IDS:
return (
"error: prototype E7M1 expects exactly quest ids "
f"{sorted(PROTOTYPE_E7M1_QUEST_IDS)!r}, got {sorted(ids)!r}"
)
return None
def _prototype_e7m1_quest_prerequisite_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if prerequisite ids are unknown or cyclic, else None."""
known = frozenset(id_to_row.keys())
for qid, row in id_to_row.items():
prereqs = row.get("prerequisiteQuestIds")
if not isinstance(prereqs, list):
continue
for pid in prereqs:
if isinstance(pid, str) and pid not in known:
return (
f"error: quest {qid!r}: prerequisiteQuestIds references unknown quest id {pid!r}"
)
visiting: set[str] = set()
visited: set[str] = set()
def dfs(node: str) -> str | None:
if node in visiting:
return f"error: cyclic prerequisiteQuestIds detected involving quest {node!r}"
if node in visited:
return None
visiting.add(node)
row = id_to_row.get(node, {})
prereqs = row.get("prerequisiteQuestIds", [])
if isinstance(prereqs, list):
for pid in prereqs:
if isinstance(pid, str):
err = dfs(pid)
if err:
return err
visiting.remove(node)
visited.add(node)
return None
for qid in known:
err = dfs(qid)
if err:
return err
return None
def _prototype_e7m1_quest_cross_ref_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if objective catalog cross-refs fail, else None."""
for qid, row in id_to_row.items():
steps = row.get("steps")
if not isinstance(steps, list):
continue
for si, step in enumerate(steps):
if not isinstance(step, dict):
continue
objectives = step.get("objectives")
if not isinstance(objectives, list):
continue
for oi, obj in enumerate(objectives):
if not isinstance(obj, dict):
continue
kind = obj.get("kind")
item_id = obj.get("itemId")
if isinstance(item_id, str) and kind in ("gather_item", "inventory_has_item"):
if item_id not in PROTOTYPE_SLICE1_ITEM_IDS:
return (
f"error: quest {qid!r} steps[{si}] objectives[{oi}]: itemId {item_id!r} "
"is not in the frozen prototype item catalog"
)
recipe_id = obj.get("recipeId")
if isinstance(recipe_id, str) and kind == "craft_recipe":
if recipe_id not in PROTOTYPE_SLICE3_RECIPE_IDS:
return (
f"error: quest {qid!r} steps[{si}] objectives[{oi}]: recipeId {recipe_id!r} "
"is not in the frozen prototype recipe catalog"
)
encounter_id = obj.get("encounterId")
if isinstance(encounter_id, str) and kind == "encounter_complete":
if encounter_id not in PROTOTYPE_E5M3_ENCOUNTER_IDS:
return (
f"error: quest {qid!r} steps[{si}] objectives[{oi}]: encounterId "
f"{encounter_id!r} is not in the frozen prototype encounter catalog"
)
return None
def _prototype_e7m1_chain_terminal_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if operator chain terminal step fails, else None."""
chain = id_to_row.get(PROTOTYPE_E7M1_CHAIN_QUEST_ID)
if not isinstance(chain, dict):
return f"error: missing chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r}"
steps = chain.get("steps")
if not isinstance(steps, list) or len(steps) == 0:
return f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} must have at least one step"
last_step = steps[-1]
if not isinstance(last_step, dict):
return f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal step must be an object"
objectives = last_step.get("objectives")
if not isinstance(objectives, list) or len(objectives) != 1:
return (
f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal step must have "
"exactly one objective"
)
obj = objectives[0]
if not isinstance(obj, dict):
return (
f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal objective must be an object"
)
if obj.get("kind") != "inventory_has_item":
return (
f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal objective kind must be "
"'inventory_has_item'"
)
if obj.get("itemId") != PROTOTYPE_E7M1_CHAIN_TERMINAL_ITEM_ID:
return (
f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal itemId must be "
f"{PROTOTYPE_E7M1_CHAIN_TERMINAL_ITEM_ID!r}, got {obj.get('itemId')!r}"
)
if obj.get("quantity") != 1:
return (
f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal quantity must be 1, "
f"got {obj.get('quantity')!r}"
)
return None
def _prototype_e7m2_completion_bundle_presence_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if a frozen quest lacks completionRewardBundle, else None."""
for qid in sorted(PROTOTYPE_E7M3_QUEST_IDS):
row = id_to_row.get(qid)
if not isinstance(row, dict):
return f"error: missing quest {qid!r}"
bundle = row.get("completionRewardBundle")
if not isinstance(bundle, dict):
return f"error: quest {qid!r} must include completionRewardBundle object"
return None
def _prototype_e7m2_completion_bundle_freeze_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if bundle contents diverge from E7M2 freeze table, else None."""
for qid, expected in PROTOTYPE_E7M2_COMPLETION_BUNDLES.items():
row = id_to_row.get(qid)
if not isinstance(row, dict):
return f"error: missing quest {qid!r}"
bundle = row.get("completionRewardBundle")
if not isinstance(bundle, dict):
return f"error: quest {qid!r} must include completionRewardBundle object"
actual = _normalize_completion_bundle(bundle)
expected_normalized = _normalize_completion_bundle(expected)
if actual != expected_normalized:
return (
f"error: quest {qid!r} completionRewardBundle must match E7M2 freeze table "
f"(expected {expected_normalized!r}, got {actual!r})"
)
return None
def _prototype_e7m2_completion_bundle_cross_ref_gate(
id_to_row: dict[str, dict],
skill_id_to_allowed_kinds: dict[str, list[str]],
) -> str | None:
"""Return a human-readable error if bundle refs fail cross-checks, else None."""
for qid in sorted(PROTOTYPE_E7M3_QUEST_IDS):
row = id_to_row.get(qid)
if not isinstance(row, dict):
continue
bundle = row.get("completionRewardBundle")
if not isinstance(bundle, dict):
continue
item_grants = bundle.get("itemGrants")
if isinstance(item_grants, list):
for gi, grant in enumerate(item_grants):
if not isinstance(grant, dict):
continue
item_id = grant.get("itemId")
if isinstance(item_id, str) and item_id not in PROTOTYPE_SLICE1_ITEM_IDS:
return (
f"error: quest {qid!r} completionRewardBundle.itemGrants[{gi}]: itemId "
f"{item_id!r} is not in the frozen prototype item catalog"
)
skill_xp_grants = bundle.get("skillXpGrants")
if isinstance(skill_xp_grants, list):
for gi, grant in enumerate(skill_xp_grants):
if not isinstance(grant, dict):
continue
skill_id = grant.get("skillId")
if isinstance(skill_id, str) and skill_id not in PROTOTYPE_SLICE1_SKILL_IDS:
return (
f"error: quest {qid!r} completionRewardBundle.skillXpGrants[{gi}]: skillId "
f"{skill_id!r} is not in the frozen prototype skill catalog"
)
allowed_kinds = skill_id_to_allowed_kinds.get(skill_id, [])
if PROTOTYPE_E7M2_MISSION_REWARD_SOURCE_KIND not in allowed_kinds:
return (
f"error: quest {qid!r} completionRewardBundle.skillXpGrants[{gi}]: skillId "
f"{skill_id!r} must allow sourceKind "
f"{PROTOTYPE_E7M2_MISSION_REWARD_SOURCE_KIND!r} in allowedXpSourceKinds "
f"(got {allowed_kinds!r})"
)
return None
def _prototype_e7m3_faction_roster_gate(faction_seen_ids: dict[str, str]) -> str | None:
"""Return a human-readable error if E7M3 faction roster fails, else None."""
ids = frozenset(faction_seen_ids.keys())
if ids != PROTOTYPE_E7M3_FACTION_IDS:
return (
"error: prototype E7M3 expects exactly faction ids "
f"{sorted(PROTOTYPE_E7M3_FACTION_IDS)!r}, got {sorted(ids)!r}"
)
return None
def _prototype_e7m3_faction_freeze_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if faction rows diverge from E7M3 freeze table, else None."""
for fid, expected in PROTOTYPE_E7M3_FACTION_FREEZE.items():
row = id_to_row.get(fid)
if not isinstance(row, dict):
return f"error: missing faction {fid!r}"
for key, expected_value in expected.items():
if row.get(key) != expected_value:
return (
f"error: faction {fid!r} must match E7M3 freeze table "
f"(expected {key}={expected_value!r}, got {row.get(key)!r})"
)
return None
def _prototype_e7m3_faction_cross_ref_gate(
id_to_row: dict[str, dict],
faction_seen_ids: dict[str, str],
) -> str | None:
"""Return a human-readable error if quest faction refs fail, else None."""
known_factions = frozenset(faction_seen_ids.keys())
for qid, row in id_to_row.items():
gate_rules = row.get("factionGateRules")
if isinstance(gate_rules, list):
for gi, rule in enumerate(gate_rules):
if not isinstance(rule, dict):
continue
faction_id = rule.get("factionId")
if isinstance(faction_id, str) and faction_id not in known_factions:
return (
f"error: quest {qid!r} factionGateRules[{gi}]: factionId {faction_id!r} "
"is not in the frozen prototype faction catalog"
)
bundle = row.get("completionRewardBundle")
if isinstance(bundle, dict):
reputation_grants = bundle.get("reputationGrants")
if isinstance(reputation_grants, list):
for gi, grant in enumerate(reputation_grants):
if not isinstance(grant, dict):
continue
faction_id = grant.get("factionId")
if isinstance(faction_id, str) and faction_id not in known_factions:
return (
f"error: quest {qid!r} completionRewardBundle.reputationGrants[{gi}]: "
f"factionId {faction_id!r} is not in the frozen prototype faction catalog"
)
return None
def _prototype_e7m3_completion_bundle_freeze_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if bundle contents diverge from E7M3 freeze table, else None."""
for qid, expected in PROTOTYPE_E7M3_COMPLETION_BUNDLES.items():
row = id_to_row.get(qid)
if not isinstance(row, dict):
return f"error: missing quest {qid!r}"
bundle = row.get("completionRewardBundle")
if not isinstance(bundle, dict):
return f"error: quest {qid!r} must include completionRewardBundle object"
actual = _normalize_e7m3_completion_bundle(bundle)
expected_normalized = _normalize_e7m3_completion_bundle(expected)
if actual != expected_normalized:
return (
f"error: quest {qid!r} completionRewardBundle must match E7M3 freeze table "
f"(expected {expected_normalized!r}, got {actual!r})"
)
return None
def _prototype_e7m3_grid_contract_shape_gate(id_to_row: dict[str, dict]) -> str | None:
"""Return a human-readable error if grid contract quest shape fails, else None."""
row = id_to_row.get(PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID)
if not isinstance(row, dict):
return f"error: missing quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r}"
prereqs = row.get("prerequisiteQuestIds")
if prereqs != [PROTOTYPE_E7M3_GRID_CONTRACT_PREREQ_QUEST_ID]:
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} prerequisiteQuestIds must be "
f"{[PROTOTYPE_E7M3_GRID_CONTRACT_PREREQ_QUEST_ID]!r}, got {prereqs!r}"
)
gate_rules = row.get("factionGateRules")
expected_gate = [
{
"factionId": PROTOTYPE_E7M3_GRID_CONTRACT_GATE_FACTION_ID,
"minStanding": PROTOTYPE_E7M3_GRID_CONTRACT_MIN_STANDING,
}
]
if gate_rules != expected_gate:
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} factionGateRules must be "
f"{expected_gate!r}, got {gate_rules!r}"
)
steps = row.get("steps")
if not isinstance(steps, list) or len(steps) != 1:
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} must have exactly one step"
)
step = steps[0]
if not isinstance(step, dict):
return f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} step must be an object"
objectives = step.get("objectives")
if not isinstance(objectives, list) or len(objectives) != 1:
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} must have exactly one objective"
)
obj = objectives[0]
if not isinstance(obj, dict):
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} objective must be an object"
)
if obj.get("kind") != "inventory_has_item":
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} objective kind must be "
"'inventory_has_item'"
)
if obj.get("itemId") != PROTOTYPE_E7M3_GRID_CONTRACT_OBJECTIVE_ITEM_ID:
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} objective itemId must be "
f"{PROTOTYPE_E7M3_GRID_CONTRACT_OBJECTIVE_ITEM_ID!r}, got {obj.get('itemId')!r}"
)
if obj.get("quantity") != 1:
return (
f"error: quest {PROTOTYPE_E7M3_GRID_CONTRACT_QUEST_ID!r} objective quantity must be 1, "
f"got {obj.get('quantity')!r}"
)
return None
def _prototype_slice4_gate(track_skill_ids: list[str]) -> str | None:
"""Return a human-readable error if Slice 4 contract fails, else None."""
if len(track_skill_ids) != 1:
return (
"error: prototype Slice 4 expects exactly one MasteryTrack across all mastery files, "
f"got {len(track_skill_ids)} track(s) with skillIds {track_skill_ids!r}"
)
if track_skill_ids[0] != PROTOTYPE_SLICE4_SALVAGE_SKILL_ID:
return (
"error: prototype Slice 4 expects the sole track skillId "
f"{PROTOTYPE_SLICE4_SALVAGE_SKILL_ID!r}, got {track_skill_ids[0]!r}"
)
return None
def _prototype_slice2_resource_node_gate(
seen_ids: dict[str, str],
id_to_lens: dict[str, str],
) -> str | None:
"""Return a human-readable error if Slice 2 resource-node contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_SLICE2_NODE_IDS:
return (
"error: prototype Slice 2 expects exactly nodeDefIds "
f"{sorted(PROTOTYPE_SLICE2_NODE_IDS)!r}, got {sorted(ids)!r}"
)
lenses = set(id_to_lens.values())
if lenses != PROTOTYPE_SLICE2_GATHER_LENSES:
return (
"error: prototype Slice 2 requires exactly one row per gatherLens "
f"{sorted(PROTOTYPE_SLICE2_GATHER_LENSES)!r}, lenses seen: {sorted(lenses)!r}"
)
return None
def _prototype_slice2_yield_gate(
yield_node_ids: frozenset[str],
yield_item_ids: frozenset[str],
) -> str | None:
"""Return a human-readable error if Slice 2 yield contract fails, else None."""
if yield_node_ids != PROTOTYPE_SLICE2_NODE_IDS:
return (
"error: prototype Slice 2 expects exactly one yield row per nodeDefId "
f"{sorted(PROTOTYPE_SLICE2_NODE_IDS)!r}, yield nodeDefIds: {sorted(yield_node_ids)!r}"
)
if yield_item_ids != PROTOTYPE_SLICE2_YIELD_ITEM_IDS:
return (
"error: prototype Slice 2 expects yield itemIds "
f"{sorted(PROTOTYPE_SLICE2_YIELD_ITEM_IDS)!r} only, got {sorted(yield_item_ids)!r}"
)
return None
def _validate_resource_node_catalogs(
*,
node_files: list[Path],
node_validator: Draft202012Validator,
) -> tuple[int, dict[str, str], dict[str, str]]:
"""Validate resource node JSON files. Returns (error_count, seen_ids, id_to_lens)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_lens: dict[str, str] = {}
for path in node_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
nodes = data.get("nodes")
if not isinstance(nodes, list):
print(f"error: {rel}: expected top-level 'nodes' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(nodes):
if not isinstance(row, dict):
print(f"error: {rel}: nodes[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(node_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} nodes[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
nid = row.get("nodeDefId")
if isinstance(nid, str) and row_schema_errors == 0:
prev = seen_ids.get(nid)
if prev:
print(
f"error: duplicate nodeDefId {nid!r} in {prev} and {rel}",
file=sys.stderr,
)
errors += 1
else:
seen_ids[nid] = rel
gather_lens = row.get("gatherLens")
if isinstance(gather_lens, str):
id_to_lens[nid] = gather_lens
return errors, seen_ids, id_to_lens
def _validate_resource_yield_catalogs(
*,
yield_files: list[Path],
yield_validator: Draft202012Validator,
known_node_ids: frozenset[str],
known_item_ids: frozenset[str],
) -> tuple[int, frozenset[str], frozenset[str]]:
"""Validate yield JSON files. Returns (error_count, yield_node_ids, yield_item_ids)."""
errors = 0
seen_node_ids: dict[str, str] = {}
yield_item_ids: set[str] = set()
for path in yield_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
yields = data.get("yields")
if not isinstance(yields, list):
print(f"error: {rel}: expected top-level 'yields' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(yields):
if not isinstance(row, dict):
print(f"error: {rel}: yields[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(yield_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} yields[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
nid = row.get("nodeDefId")
item_id = row.get("itemId")
if isinstance(nid, str) and row_schema_errors == 0:
prev = seen_node_ids.get(nid)
if prev:
print(
f"error: duplicate yield nodeDefId {nid!r} in {prev} and {rel}",
file=sys.stderr,
)
errors += 1
else:
seen_node_ids[nid] = rel
if nid not in known_node_ids:
print(
f"error: {rel} yields[{i}]: nodeDefId {nid!r} is not defined in resource node catalogs",
file=sys.stderr,
)
errors += 1
if isinstance(item_id, str) and row_schema_errors == 0:
yield_item_ids.add(item_id)
if item_id not in known_item_ids:
print(
f"error: {rel} yields[{i}]: itemId {item_id!r} is not in item catalogs",
file=sys.stderr,
)
errors += 1
return errors, frozenset(seen_node_ids.keys()), frozenset(yield_item_ids)
def _recipe_def_validator() -> Draft202012Validator:
"""Build a validator that resolves recipe-def $ref to recipe-io-row schema."""
io_schema = json.loads(RECIPE_IO_ROW_SCHEMA.read_text(encoding="utf-8"))
def_schema = json.loads(RECIPE_SCHEMA.read_text(encoding="utf-8"))
registry = Registry().with_resources(
[
(def_schema["$id"], Resource.from_contents(def_schema)),
(io_schema["$id"], Resource.from_contents(io_schema)),
]
)
return Draft202012Validator(def_schema, registry=registry)
def _prototype_slice3_recipe_gate(
seen_ids: dict[str, str],
id_to_kind: dict[str, str],
id_to_skill: dict[str, str],
) -> str | None:
"""Return a human-readable error if Slice 3 recipe contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_SLICE3_RECIPE_IDS:
return (
"error: prototype Slice 3 expects exactly recipe ids "
f"{sorted(PROTOTYPE_SLICE3_RECIPE_IDS)!r}, got {sorted(ids)!r}"
)
kinds = set(id_to_kind.values())
if "process" not in kinds or "make" not in kinds:
return (
"error: prototype Slice 3 requires at least one process and one make recipe; "
f"recipeKind values seen: {sorted(kinds)!r}"
)
for rid, skill_id in id_to_skill.items():
if skill_id != PROTOTYPE_SLICE3_REFINE_SKILL_ID:
return (
f"error: {rid!r} requiredSkillId must be {PROTOTYPE_SLICE3_REFINE_SKILL_ID!r}, "
f"got {skill_id!r}"
)
return None
def _validate_recipe_catalogs(
*,
recipe_files: list[Path],
recipe_validator: Draft202012Validator,
known_item_ids: frozenset[str],
known_skill_ids: frozenset[str],
) -> tuple[int, dict[str, str], dict[str, str], dict[str, str]]:
"""Validate recipe JSON files. Returns (error_count, seen_ids, id_to_kind, id_to_skill)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_kind: dict[str, str] = {}
id_to_skill: dict[str, str] = {}
for path in recipe_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
recipes = data.get("recipes")
if not isinstance(recipes, list):
print(f"error: {rel}: expected top-level 'recipes' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(recipes):
if not isinstance(row, dict):
print(f"error: {rel}: recipes[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(recipe_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} recipes[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
rid = row.get("id")
if isinstance(rid, str) and row_schema_errors == 0:
prev = seen_ids.get(rid)
if prev:
print(f"error: duplicate recipe id {rid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[rid] = rel
recipe_kind = row.get("recipeKind")
if isinstance(recipe_kind, str):
id_to_kind[rid] = recipe_kind
skill_id = row.get("requiredSkillId")
if isinstance(skill_id, str):
id_to_skill[rid] = skill_id
if skill_id not in known_skill_ids:
print(
f"error: {rel} recipes[{i}]: requiredSkillId {skill_id!r} "
f"is not a known SkillDef id (known: {sorted(known_skill_ids)!r})",
file=sys.stderr,
)
errors += 1
for io_key in ("inputs", "outputs"):
io_rows = row.get(io_key)
if not isinstance(io_rows, list):
continue
for j, io_row in enumerate(io_rows):
if not isinstance(io_row, dict):
continue
item_id = io_row.get("itemId")
if isinstance(item_id, str) and item_id not in known_item_ids:
print(
f"error: {rel} recipes[{i}].{io_key}[{j}]: itemId {item_id!r} "
"is not in item catalogs",
file=sys.stderr,
)
errors += 1
return errors, seen_ids, id_to_kind, id_to_skill
def main() -> int:
if not SKILL_SCHEMA.is_file():
print(f"error: missing schema {SKILL_SCHEMA}", file=sys.stderr)
return 1
if not LEVEL_CURVE_SCHEMA.is_file():
print(f"error: missing schema {LEVEL_CURVE_SCHEMA}", file=sys.stderr)
return 1
if not MASTERY_SCHEMA.is_file():
print(f"error: missing schema {MASTERY_SCHEMA}", file=sys.stderr)
return 1
if not ITEM_SCHEMA.is_file():
print(f"error: missing schema {ITEM_SCHEMA}", file=sys.stderr)
return 1
if not RESOURCE_NODE_SCHEMA.is_file():
print(f"error: missing schema {RESOURCE_NODE_SCHEMA}", file=sys.stderr)
return 1
if not RESOURCE_YIELD_ROW_SCHEMA.is_file():
print(f"error: missing schema {RESOURCE_YIELD_ROW_SCHEMA}", file=sys.stderr)
return 1
if not RECIPE_IO_ROW_SCHEMA.is_file():
print(f"error: missing schema {RECIPE_IO_ROW_SCHEMA}", file=sys.stderr)
return 1
if not RECIPE_SCHEMA.is_file():
print(f"error: missing schema {RECIPE_SCHEMA}", file=sys.stderr)
return 1
if not ABILITY_SCHEMA.is_file():
print(f"error: missing schema {ABILITY_SCHEMA}", file=sys.stderr)
return 1
if not NPC_BEHAVIOR_SCHEMA.is_file():
print(f"error: missing schema {NPC_BEHAVIOR_SCHEMA}", file=sys.stderr)
return 1
if not REWARD_GRANT_ROW_SCHEMA.is_file():
print(f"error: missing schema {REWARD_GRANT_ROW_SCHEMA}", file=sys.stderr)
return 1
if not REWARD_TABLE_SCHEMA.is_file():
print(f"error: missing schema {REWARD_TABLE_SCHEMA}", file=sys.stderr)
return 1
if not ENCOUNTER_DEF_SCHEMA.is_file():
print(f"error: missing schema {ENCOUNTER_DEF_SCHEMA}", file=sys.stderr)
return 1
if not QUEST_OBJECTIVE_DEF_SCHEMA.is_file():
print(f"error: missing schema {QUEST_OBJECTIVE_DEF_SCHEMA}", file=sys.stderr)
return 1
if not QUEST_STEP_DEF_SCHEMA.is_file():
print(f"error: missing schema {QUEST_STEP_DEF_SCHEMA}", file=sys.stderr)
return 1
if not QUEST_DEF_SCHEMA.is_file():
print(f"error: missing schema {QUEST_DEF_SCHEMA}", file=sys.stderr)
return 1
if not QUEST_REWARD_BUNDLE_SCHEMA.is_file():
print(f"error: missing schema {QUEST_REWARD_BUNDLE_SCHEMA}", file=sys.stderr)
return 1
if not QUEST_SKILL_XP_GRANT_SCHEMA.is_file():
print(f"error: missing schema {QUEST_SKILL_XP_GRANT_SCHEMA}", file=sys.stderr)
return 1
if not FACTION_DEF_SCHEMA.is_file():
print(f"error: missing schema {FACTION_DEF_SCHEMA}", file=sys.stderr)
return 1
if not FACTION_GATE_RULE_SCHEMA.is_file():
print(f"error: missing schema {FACTION_GATE_RULE_SCHEMA}", file=sys.stderr)
return 1
if not REPUTATION_GRANT_ROW_SCHEMA.is_file():
print(f"error: missing schema {REPUTATION_GRANT_ROW_SCHEMA}", file=sys.stderr)
return 1
skill_schema = json.loads(SKILL_SCHEMA.read_text(encoding="utf-8"))
skill_validator = Draft202012Validator(skill_schema)
level_curve_schema = json.loads(LEVEL_CURVE_SCHEMA.read_text(encoding="utf-8"))
level_curve_validator = Draft202012Validator(level_curve_schema)
mastery_schema = json.loads(MASTERY_SCHEMA.read_text(encoding="utf-8"))
mastery_validator = Draft202012Validator(mastery_schema)
item_schema = json.loads(ITEM_SCHEMA.read_text(encoding="utf-8"))
item_validator = Draft202012Validator(item_schema)
resource_node_schema = json.loads(RESOURCE_NODE_SCHEMA.read_text(encoding="utf-8"))
resource_node_validator = Draft202012Validator(resource_node_schema)
resource_yield_schema = json.loads(RESOURCE_YIELD_ROW_SCHEMA.read_text(encoding="utf-8"))
resource_yield_validator = Draft202012Validator(resource_yield_schema)
recipe_validator = _recipe_def_validator()
ability_schema = json.loads(ABILITY_SCHEMA.read_text(encoding="utf-8"))
ability_validator = Draft202012Validator(ability_schema)
npc_behavior_schema = json.loads(NPC_BEHAVIOR_SCHEMA.read_text(encoding="utf-8"))
npc_behavior_validator = Draft202012Validator(npc_behavior_schema)
reward_table_validator = _reward_table_def_validator()
encounter_schema = json.loads(ENCOUNTER_DEF_SCHEMA.read_text(encoding="utf-8"))
encounter_validator = Draft202012Validator(encounter_schema)
quest_validator = _quest_def_validator()
faction_schema = json.loads(FACTION_DEF_SCHEMA.read_text(encoding="utf-8"))
faction_validator = Draft202012Validator(faction_schema)
if not SKILLS_DIR.is_dir():
print(f"error: missing directory {SKILLS_DIR}", file=sys.stderr)
return 1
skill_files = sorted(SKILLS_DIR.glob("*_skills.json"))
if not skill_files:
print(f"error: no *_skills.json files under {SKILLS_DIR}", file=sys.stderr)
return 1
curve_files = sorted(SKILLS_DIR.glob("*_level_curve.json"))
if not curve_files:
print(f"error: no *_level_curve.json files under {SKILLS_DIR}", file=sys.stderr)
return 1
if not MASTERY_DIR.is_dir():
print(f"error: missing directory {MASTERY_DIR}", file=sys.stderr)
return 1
mastery_files = sorted(MASTERY_DIR.glob("*_mastery.json"))
if not mastery_files:
print(f"error: no *_mastery.json files under {MASTERY_DIR}", file=sys.stderr)
return 1
if not ITEMS_DIR.is_dir():
print(f"error: missing directory {ITEMS_DIR}", file=sys.stderr)
return 1
item_files = sorted(ITEMS_DIR.glob("*_items.json"))
if not item_files:
print(f"error: no *_items.json files under {ITEMS_DIR}", file=sys.stderr)
return 1
if not RESOURCE_NODES_DIR.is_dir():
print(f"error: missing directory {RESOURCE_NODES_DIR}", file=sys.stderr)
return 1
node_files = sorted(RESOURCE_NODES_DIR.glob("*_resource_nodes.json"))
if not node_files:
print(f"error: no *_resource_nodes.json files under {RESOURCE_NODES_DIR}", file=sys.stderr)
return 1
yield_files = sorted(RESOURCE_NODES_DIR.glob("*_resource_yields.json"))
if not yield_files:
print(f"error: no *_resource_yields.json files under {RESOURCE_NODES_DIR}", file=sys.stderr)
return 1
if not RECIPES_DIR.is_dir():
print(f"error: missing directory {RECIPES_DIR}", file=sys.stderr)
return 1
recipe_files = sorted(RECIPES_DIR.glob("*_recipes.json"))
if not recipe_files:
print(f"error: no *_recipes.json files under {RECIPES_DIR}", file=sys.stderr)
return 1
if not ABILITIES_DIR.is_dir():
print(f"error: missing directory {ABILITIES_DIR}", file=sys.stderr)
return 1
ability_files = sorted(ABILITIES_DIR.glob("*_abilities.json"))
if not ability_files:
print(f"error: no *_abilities.json files under {ABILITIES_DIR}", file=sys.stderr)
return 1
if not NPC_BEHAVIORS_DIR.is_dir():
print(f"error: missing directory {NPC_BEHAVIORS_DIR}", file=sys.stderr)
return 1
npc_behavior_files = sorted(NPC_BEHAVIORS_DIR.glob("*_npc_behaviors.json"))
if not npc_behavior_files:
print(f"error: no *_npc_behaviors.json files under {NPC_BEHAVIORS_DIR}", file=sys.stderr)
return 1
if not REWARD_TABLES_DIR.is_dir():
print(f"error: missing directory {REWARD_TABLES_DIR}", file=sys.stderr)
return 1
reward_table_files = sorted(REWARD_TABLES_DIR.glob("*_reward_tables.json"))
if not reward_table_files:
print(f"error: no *_reward_tables.json files under {REWARD_TABLES_DIR}", file=sys.stderr)
return 1
if not ENCOUNTERS_DIR.is_dir():
print(f"error: missing directory {ENCOUNTERS_DIR}", file=sys.stderr)
return 1
encounter_files = sorted(ENCOUNTERS_DIR.glob("*_encounters.json"))
if not encounter_files:
print(f"error: no *_encounters.json files under {ENCOUNTERS_DIR}", file=sys.stderr)
return 1
if not QUESTS_DIR.is_dir():
print(f"error: missing directory {QUESTS_DIR}", file=sys.stderr)
return 1
quest_files = sorted(QUESTS_DIR.glob("*_quests.json"))
if not quest_files:
print(f"error: no *_quests.json files under {QUESTS_DIR}", file=sys.stderr)
return 1
if not FACTIONS_DIR.is_dir():
print(f"error: missing directory {FACTIONS_DIR}", file=sys.stderr)
return 1
faction_files = sorted(FACTIONS_DIR.glob("*_factions.json"))
if not faction_files:
print(f"error: no *_factions.json files under {FACTIONS_DIR}", file=sys.stderr)
return 1
seen_ids: dict[str, str] = {}
id_to_category: dict[str, str] = {}
skill_id_to_allowed_kinds: dict[str, list[str]] = {}
errors = 0
for path in skill_files:
data = json.loads(path.read_text(encoding="utf-8"))
skills = data.get("skills")
if not isinstance(skills, list):
print(f"error: {path.relative_to(REPO_ROOT)}: expected top-level 'skills' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(skills):
if not isinstance(row, dict):
print(f"error: {path.relative_to(REPO_ROOT)}: skills[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(skill_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(
f"error: {path.relative_to(REPO_ROOT)} skills[{i}] {loc}: {err.message}",
file=sys.stderr,
)
row_schema_errors += 1
errors += 1
sid = row.get("id")
# Reserve seen_ids (and duplicate detection) only for schema-clean rows so
# fixing a broken row does not surface a spurious duplicate vs an earlier invalid row.
if isinstance(sid, str) and row_schema_errors == 0:
prev = seen_ids.get(sid)
if prev:
print(
f"error: duplicate skill id {sid!r} in {prev} and {path.relative_to(REPO_ROOT)}",
file=sys.stderr,
)
errors += 1
else:
seen_ids[sid] = str(path.relative_to(REPO_ROOT))
cat = row.get("category")
if isinstance(cat, str):
id_to_category[sid] = cat
allowed_kinds = row.get("allowedXpSourceKinds")
if isinstance(allowed_kinds, list):
skill_id_to_allowed_kinds[sid] = [
kind for kind in allowed_kinds if isinstance(kind, str)
]
for path in curve_files:
data = json.loads(path.read_text(encoding="utf-8"))
row_errors = 0
for err in sorted(level_curve_validator.iter_errors(data), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(
f"error: {path.relative_to(REPO_ROOT)} {loc}: {err.message}",
file=sys.stderr,
)
row_errors += 1
errors += 1
if row_errors > 0:
continue
levels = data.get("levels")
if not isinstance(levels, list):
print(f"error: {path.relative_to(REPO_ROOT)}: expected top-level 'levels' array", file=sys.stderr)
errors += 1
continue
if len(levels) == 0:
print(f"error: {path.relative_to(REPO_ROOT)}: levels must contain at least one row", file=sys.stderr)
errors += 1
continue
first = levels[0]
if not isinstance(first, dict) or first.get("level") != 1 or first.get("requiredXp") != 0:
print(
f"error: {path.relative_to(REPO_ROOT)}: first row must be level=1 and requiredXp=0",
file=sys.stderr,
)
errors += 1
prev_level = 0
prev_xp = -1
for i, row in enumerate(levels):
if not isinstance(row, dict):
print(f"error: {path.relative_to(REPO_ROOT)}: levels[{i}] must be an object", file=sys.stderr)
errors += 1
continue
level = row.get("level")
required_xp = row.get("requiredXp")
if not isinstance(level, int) or not isinstance(required_xp, int):
print(
f"error: {path.relative_to(REPO_ROOT)}: levels[{i}] level/requiredXp must be integers",
file=sys.stderr,
)
errors += 1
continue
if level <= prev_level:
print(
f"error: {path.relative_to(REPO_ROOT)}: levels[{i}].level must be strictly increasing",
file=sys.stderr,
)
errors += 1
if required_xp <= prev_xp:
print(
f"error: {path.relative_to(REPO_ROOT)}: levels[{i}].requiredXp must be strictly increasing",
file=sys.stderr,
)
errors += 1
prev_level = level
prev_xp = required_xp
if errors:
print(f"content validation failed with {errors} error(s)", file=sys.stderr)
return 1
slice1_err = _prototype_slice1_gate(seen_ids, id_to_category)
if slice1_err:
print(slice1_err, file=sys.stderr)
return 1
mastery_errors, track_skill_ids = _validate_mastery_catalogs(
mastery_files=mastery_files,
mastery_validator=mastery_validator,
skill_ids=frozenset(seen_ids.keys()),
)
if mastery_errors:
print(f"content validation failed with {mastery_errors} error(s)", file=sys.stderr)
return 1
slice4_err = _prototype_slice4_gate(track_skill_ids)
if slice4_err:
print(slice4_err, file=sys.stderr)
return 1
item_errors, item_seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max = _validate_item_catalogs(
item_files=item_files,
item_validator=item_validator,
)
if item_errors:
print(f"content validation failed with {item_errors} error(s)", file=sys.stderr)
return 1
slice1_item_err = _prototype_slice1_item_gate(
item_seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max
)
if slice1_item_err:
print(slice1_item_err, file=sys.stderr)
return 1
node_errors, node_seen_ids, id_to_lens = _validate_resource_node_catalogs(
node_files=node_files,
node_validator=resource_node_validator,
)
if node_errors:
print(f"content validation failed with {node_errors} error(s)", file=sys.stderr)
return 1
slice2_node_err = _prototype_slice2_resource_node_gate(node_seen_ids, id_to_lens)
if slice2_node_err:
print(slice2_node_err, file=sys.stderr)
return 1
yield_errors, yield_node_ids, yield_item_ids = _validate_resource_yield_catalogs(
yield_files=yield_files,
yield_validator=resource_yield_validator,
known_node_ids=frozenset(node_seen_ids.keys()),
known_item_ids=frozenset(item_seen_ids.keys()),
)
if yield_errors:
print(f"content validation failed with {yield_errors} error(s)", file=sys.stderr)
return 1
slice2_yield_err = _prototype_slice2_yield_gate(yield_node_ids, yield_item_ids)
if slice2_yield_err:
print(slice2_yield_err, file=sys.stderr)
return 1
recipe_errors, recipe_seen_ids, id_to_kind, id_to_skill = _validate_recipe_catalogs(
recipe_files=recipe_files,
recipe_validator=recipe_validator,
known_item_ids=frozenset(item_seen_ids.keys()),
known_skill_ids=frozenset(seen_ids.keys()),
)
if recipe_errors:
print(f"content validation failed with {recipe_errors} error(s)", file=sys.stderr)
return 1
slice3_recipe_err = _prototype_slice3_recipe_gate(recipe_seen_ids, id_to_kind, id_to_skill)
if slice3_recipe_err:
print(slice3_recipe_err, file=sys.stderr)
return 1
ability_errors, ability_seen_ids, ability_id_to_row = _validate_ability_catalogs(
ability_files=ability_files,
ability_validator=ability_validator,
)
if ability_errors:
print(f"content validation failed with {ability_errors} error(s)", file=sys.stderr)
return 1
ability_catalog_err = _prototype_ability_catalog_gate(ability_seen_ids)
if ability_catalog_err:
print(ability_catalog_err, file=sys.stderr)
return 1
npc_behavior_errors, npc_behavior_seen_ids, npc_behavior_id_to_row = _validate_npc_behavior_catalogs(
npc_behavior_files=npc_behavior_files,
npc_behavior_validator=npc_behavior_validator,
)
if npc_behavior_errors:
print(f"content validation failed with {npc_behavior_errors} error(s)", file=sys.stderr)
return 1
e5m2_npc_behavior_err = _prototype_e5m2_npc_behavior_gate(npc_behavior_seen_ids)
if e5m2_npc_behavior_err:
print(e5m2_npc_behavior_err, file=sys.stderr)
return 1
e5m2_npc_behavior_numeric_err = _prototype_e5m2_npc_behavior_numeric_gate(npc_behavior_id_to_row)
if e5m2_npc_behavior_numeric_err:
print(e5m2_npc_behavior_numeric_err, file=sys.stderr)
return 1
e5m2_attack_ability_err = _prototype_e5m2_npc_behavior_attack_ability_gate(
npc_behavior_id_to_row,
ability_id_to_row,
)
if e5m2_attack_ability_err:
print(e5m2_attack_ability_err, file=sys.stderr)
return 1
reward_table_errors, reward_table_seen_ids, reward_table_id_to_row = _validate_reward_table_catalogs(
reward_table_files=reward_table_files,
reward_table_validator=reward_table_validator,
known_item_ids=frozenset(item_seen_ids.keys()),
)
if reward_table_errors:
print(f"content validation failed with {reward_table_errors} error(s)", file=sys.stderr)
return 1
e5m3_reward_table_err = _prototype_e5m3_reward_table_gate(reward_table_seen_ids)
if e5m3_reward_table_err:
print(e5m3_reward_table_err, file=sys.stderr)
return 1
e5m3_reward_grant_err = _prototype_e5m3_reward_grant_content_gate(reward_table_id_to_row)
if e5m3_reward_grant_err:
print(e5m3_reward_grant_err, file=sys.stderr)
return 1
encounter_errors, encounter_seen_ids, encounter_id_to_row = _validate_encounter_catalogs(
encounter_files=encounter_files,
encounter_validator=encounter_validator,
known_reward_table_ids=frozenset(reward_table_seen_ids.keys()),
)
if encounter_errors:
print(f"content validation failed with {encounter_errors} error(s)", file=sys.stderr)
return 1
e5m3_encounter_err = _prototype_e5m3_encounter_gate(encounter_seen_ids)
if e5m3_encounter_err:
print(e5m3_encounter_err, file=sys.stderr)
return 1
e5m3_encounter_cross_ref_err = _prototype_e5m3_encounter_cross_ref_gate(encounter_id_to_row)
if e5m3_encounter_cross_ref_err:
print(e5m3_encounter_cross_ref_err, file=sys.stderr)
return 1
faction_errors, faction_seen_ids, faction_id_to_row = _validate_faction_catalogs(
faction_files=faction_files,
faction_validator=faction_validator,
)
if faction_errors:
print(f"content validation failed with {faction_errors} error(s)", file=sys.stderr)
return 1
e7m3_faction_roster_err = _prototype_e7m3_faction_roster_gate(faction_seen_ids)
if e7m3_faction_roster_err:
print(e7m3_faction_roster_err, file=sys.stderr)
return 1
e7m3_faction_freeze_err = _prototype_e7m3_faction_freeze_gate(faction_id_to_row)
if e7m3_faction_freeze_err:
print(e7m3_faction_freeze_err, file=sys.stderr)
return 1
quest_errors, quest_seen_ids, quest_id_to_row = _validate_quest_catalogs(
quest_files=quest_files,
quest_validator=quest_validator,
)
if quest_errors:
print(f"content validation failed with {quest_errors} error(s)", file=sys.stderr)
return 1
e7m3_quest_roster_err = _prototype_e7m3_quest_roster_gate(quest_seen_ids)
if e7m3_quest_roster_err:
print(e7m3_quest_roster_err, file=sys.stderr)
return 1
e7m1_prereq_err = _prototype_e7m1_quest_prerequisite_gate(quest_id_to_row)
if e7m1_prereq_err:
print(e7m1_prereq_err, file=sys.stderr)
return 1
e7m1_cross_ref_err = _prototype_e7m1_quest_cross_ref_gate(quest_id_to_row)
if e7m1_cross_ref_err:
print(e7m1_cross_ref_err, file=sys.stderr)
return 1
e7m1_chain_terminal_err = _prototype_e7m1_chain_terminal_gate(quest_id_to_row)
if e7m1_chain_terminal_err:
print(e7m1_chain_terminal_err, file=sys.stderr)
return 1
e7m2_bundle_presence_err = _prototype_e7m2_completion_bundle_presence_gate(quest_id_to_row)
if e7m2_bundle_presence_err:
print(e7m2_bundle_presence_err, file=sys.stderr)
return 1
e7m2_bundle_freeze_err = _prototype_e7m2_completion_bundle_freeze_gate(quest_id_to_row)
if e7m2_bundle_freeze_err:
print(e7m2_bundle_freeze_err, file=sys.stderr)
return 1
e7m2_bundle_cross_ref_err = _prototype_e7m2_completion_bundle_cross_ref_gate(
quest_id_to_row,
skill_id_to_allowed_kinds,
)
if e7m2_bundle_cross_ref_err:
print(e7m2_bundle_cross_ref_err, file=sys.stderr)
return 1
e7m3_faction_cross_ref_err = _prototype_e7m3_faction_cross_ref_gate(
quest_id_to_row,
faction_seen_ids,
)
if e7m3_faction_cross_ref_err:
print(e7m3_faction_cross_ref_err, file=sys.stderr)
return 1
e7m3_bundle_freeze_err = _prototype_e7m3_completion_bundle_freeze_gate(quest_id_to_row)
if e7m3_bundle_freeze_err:
print(e7m3_bundle_freeze_err, file=sys.stderr)
return 1
e7m3_grid_contract_err = _prototype_e7m3_grid_contract_shape_gate(quest_id_to_row)
if e7m3_grid_contract_err:
print(e7m3_grid_contract_err, file=sys.stderr)
return 1
print(
"content OK: "
f"{len(skill_files)} skill catalog file(s), "
f"{len(curve_files)} level curve file(s), "
f"{len(mastery_files)} mastery catalog file(s), "
f"{len(item_files)} item catalog file(s), "
f"{len(node_files)} resource node catalog file(s), "
f"{len(yield_files)} resource yield catalog file(s), "
f"{len(recipe_files)} recipe catalog file(s), "
f"{len(ability_files)} ability catalog file(s), "
f"{len(npc_behavior_files)} npc behavior catalog file(s), "
f"{len(reward_table_files)} reward table catalog file(s), "
f"{len(encounter_files)} encounter catalog file(s), "
f"{len(faction_files)} faction catalog file(s), "
f"{len(quest_files)} quest catalog file(s), "
f"{len(seen_ids)} unique skill id(s), "
f"{len(item_seen_ids)} unique item id(s), "
f"{len(node_seen_ids)} unique nodeDefId(s), "
f"{len(recipe_seen_ids)} unique recipe id(s), "
f"{len(ability_seen_ids)} unique ability id(s), "
f"{len(npc_behavior_seen_ids)} unique npc behavior id(s), "
f"{len(reward_table_seen_ids)} unique reward table id(s), "
f"{len(encounter_seen_ids)} unique encounter id(s), "
f"{len(faction_seen_ids)} unique faction id(s), "
f"{len(quest_seen_ids)} unique quest id(s), "
f"{len(track_skill_ids)} mastery track(s)"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink