#!/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) - contract template catalogs: content/contracts/*_contract_templates.json rows vs content/schemas/contract-template.schema.json (NEO-144); completionRewardBundle via quest-reward-bundle.schema.json """ 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" CONTRACT_TEMPLATE_SCHEMA = REPO_ROOT / "content/schemas/contract-template.schema.json" CONTRACT_SEED_SCHEMA = REPO_ROOT / "content/schemas/contract-seed.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" CONTRACTS_DIR = REPO_ROOT / "content/contracts" # 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): Slice 1 onboarding quest ids (historical subset). # Active CI roster gate is PROTOTYPE_E7M3_QUEST_IDS (five quests, NEO-133). # Keep in sync with E7.M1 freeze table; server ExpectedQuestIds expanded for E7M3. 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} ], }, } # Epic 7 Slice 4 prototype lock (NEO-144): one contract template + band-1 economy caps. # Keep in sync with E7.M4 freeze table and future NEO-145 server loader. PROTOTYPE_E7M4_TEMPLATE_IDS = frozenset({"prototype_contract_clear_combat_pocket"}) PROTOTYPE_E7M4_TEMPLATE_FREEZE: dict[str, dict] = { "prototype_contract_clear_combat_pocket": { "displayName": "Clear Combat Pocket (Repeat)", "zoneDifficultyBand": 1, "objectiveKind": "encounter_clear", "encounterTemplateId": "prototype_combat_pocket", "minFactionStanding": { "factionId": "prototype_faction_grid_operators", "minStanding": 0, }, "completionRewardBundle": { "itemGrants": [{"itemId": "scrap_metal_bulk", "quantity": 5}], "skillXpGrants": [{"skillId": "salvage", "amount": 15}], }, }, } PROTOTYPE_E7M4_BAND_CAPS: dict[int, dict[str, int]] = { 1: { "maxItemQuantity": 10, "maxSkillXpAmount": 25, "maxReputationAmount": 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 _contract_template_validator() -> Draft202012Validator: """Build a validator that resolves contract-template $ref to bundle and gate schemas.""" template_schema = json.loads(CONTRACT_TEMPLATE_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( [ (template_schema["$id"], Resource.from_contents(template_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(template_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 _validate_contract_catalogs( *, contract_files: list[Path], contract_validator: Draft202012Validator, ) -> tuple[int, dict[str, str], dict[str, dict]]: """Validate contract template 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 contract_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 contract_templates = data.get("contractTemplates") if not isinstance(contract_templates, list): print(f"error: {rel}: expected top-level 'contractTemplates' array", file=sys.stderr) errors += 1 continue for i, row in enumerate(contract_templates): if not isinstance(row, dict): print(f"error: {rel}: contractTemplates[{i}] must be an object", file=sys.stderr) errors += 1 continue row_schema_errors = 0 for err in sorted(contract_validator.iter_errors(row), key=lambda e: e.path): loc = ".".join(str(p) for p in err.path) or "(root)" print(f"error: {rel} contractTemplates[{i}] {loc}: {err.message}", file=sys.stderr) row_schema_errors += 1 errors += 1 tid = row.get("id") if isinstance(tid, str) and row_schema_errors == 0: prev = seen_ids.get(tid) if prev: print(f"error: duplicate contract template id {tid!r} in {prev} and {rel}", file=sys.stderr) errors += 1 else: seen_ids[tid] = rel id_to_row[tid] = row 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_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_standing_band_gate(id_to_row: dict[str, dict]) -> str | None: """Return a human-readable error if faction standing band ordering fails, else None.""" for fid, row in id_to_row.items(): if not isinstance(row, dict): continue min_standing = row.get("minStanding") max_standing = row.get("maxStanding") neutral_standing = row.get("neutralStanding") if not all(isinstance(value, int) for value in (min_standing, max_standing, neutral_standing)): continue if min_standing > neutral_standing or neutral_standing > max_standing: return ( f"error: faction {fid!r} requires minStanding <= neutralStanding <= maxStanding " f"(got min={min_standing!r}, neutral={neutral_standing!r}, max={max_standing!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_e7m4_template_roster_gate(seen_ids: dict[str, str]) -> str | None: """Return a human-readable error if E7M4 template roster fails, else None.""" ids = frozenset(seen_ids.keys()) if ids != PROTOTYPE_E7M4_TEMPLATE_IDS: return ( "error: prototype E7M4 expects exactly contract template ids " f"{sorted(PROTOTYPE_E7M4_TEMPLATE_IDS)!r}, got {sorted(ids)!r}" ) return None def _prototype_e7m4_template_freeze_gate(id_to_row: dict[str, dict]) -> str | None: """Return a human-readable error if template rows diverge from E7M4 freeze table, else None.""" for tid, expected in PROTOTYPE_E7M4_TEMPLATE_FREEZE.items(): row = id_to_row.get(tid) if not isinstance(row, dict): return f"error: missing contract template {tid!r}" for field in ( "displayName", "zoneDifficultyBand", "objectiveKind", "encounterTemplateId", ): if row.get(field) != expected[field]: return ( f"error: contract template {tid!r} {field} must be " f"{expected[field]!r}, got {row.get(field)!r}" ) min_standing = row.get("minFactionStanding") if min_standing != expected["minFactionStanding"]: return ( f"error: contract template {tid!r} minFactionStanding must be " f"{expected['minFactionStanding']!r}, got {min_standing!r}" ) bundle = row.get("completionRewardBundle") if not isinstance(bundle, dict): return f"error: contract template {tid!r} must include completionRewardBundle object" actual_bundle = _normalize_completion_bundle(bundle) expected_bundle = _normalize_completion_bundle(expected["completionRewardBundle"]) if actual_bundle != expected_bundle: return ( f"error: contract template {tid!r} completionRewardBundle must match E7M4 freeze table " f"(expected {expected_bundle!r}, got {actual_bundle!r})" ) return None def _prototype_e7m4_cross_ref_gate( id_to_row: dict[str, dict], encounter_seen_ids: dict[str, str], faction_seen_ids: dict[str, str], skill_id_to_allowed_kinds: dict[str, list[str]], ) -> str | None: """Return a human-readable error if contract template refs fail cross-checks, else None.""" for tid in sorted(PROTOTYPE_E7M4_TEMPLATE_IDS): row = id_to_row.get(tid) if not isinstance(row, dict): continue encounter_id = row.get("encounterTemplateId") if isinstance(encounter_id, str) and encounter_id not in encounter_seen_ids: return ( f"error: contract template {tid!r} encounterTemplateId {encounter_id!r} " "is not in the loaded encounter catalog" ) min_standing = row.get("minFactionStanding") if isinstance(min_standing, dict): faction_id = min_standing.get("factionId") if isinstance(faction_id, str) and faction_id not in faction_seen_ids: return ( f"error: contract template {tid!r} minFactionStanding.factionId " f"{faction_id!r} is not in the frozen prototype faction catalog" ) 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: contract template {tid!r} completionRewardBundle.itemGrants[{gi}]: " f"itemId {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: contract template {tid!r} completionRewardBundle.skillXpGrants[{gi}]: " f"skillId {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: contract template {tid!r} completionRewardBundle.skillXpGrants[{gi}]: " f"skillId {skill_id!r} must allow sourceKind " f"{PROTOTYPE_E7M2_MISSION_REWARD_SOURCE_KIND!r} in allowedXpSourceKinds " f"(got {allowed_kinds!r})" ) 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 faction_seen_ids: return ( f"error: contract template {tid!r} completionRewardBundle.reputationGrants[{gi}]: " f"factionId {faction_id!r} is not in the frozen prototype faction catalog" ) return None def _prototype_e7m4_band_cap_gate(id_to_row: dict[str, dict]) -> str | None: """Return a human-readable error if a template bundle exceeds band caps, else None.""" for tid in sorted(PROTOTYPE_E7M4_TEMPLATE_IDS): row = id_to_row.get(tid) if not isinstance(row, dict): continue band = row.get("zoneDifficultyBand") if not isinstance(band, int): continue caps = PROTOTYPE_E7M4_BAND_CAPS.get(band) if caps is None: return ( f"error: contract template {tid!r} zoneDifficultyBand {band} has no " "prototype economy cap policy" ) 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 quantity = grant.get("quantity") if isinstance(quantity, int) and quantity > caps["maxItemQuantity"]: return ( f"error: contract template {tid!r} completionRewardBundle.itemGrants[{gi}] " f"quantity {quantity} exceeds band {band} cap {caps['maxItemQuantity']}" ) 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 amount = grant.get("amount") if isinstance(amount, int) and amount > caps["maxSkillXpAmount"]: return ( f"error: contract template {tid!r} completionRewardBundle.skillXpGrants[{gi}] " f"amount {amount} exceeds band {band} cap {caps['maxSkillXpAmount']}" ) reputation_grants = bundle.get("reputationGrants") if isinstance(reputation_grants, list): for gi, grant in enumerate(reputation_grants): if not isinstance(grant, dict): continue amount = grant.get("amount") if isinstance(amount, int) and abs(amount) > caps["maxReputationAmount"]: return ( f"error: contract template {tid!r} completionRewardBundle.reputationGrants[{gi}] " f"amount {amount} exceeds band {band} rep cap {caps['maxReputationAmount']}" ) 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() contract_validator = _contract_template_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 if not CONTRACTS_DIR.is_dir(): print(f"error: missing directory {CONTRACTS_DIR}", file=sys.stderr) return 1 contract_files = sorted(CONTRACTS_DIR.glob("*_contract_templates.json")) if not contract_files: print(f"error: no *_contract_templates.json files under {CONTRACTS_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 e7m3_faction_band_err = _prototype_e7m3_faction_standing_band_gate(faction_id_to_row) if e7m3_faction_band_err: print(e7m3_faction_band_err, file=sys.stderr) return 1 contract_errors, contract_seen_ids, contract_id_to_row = _validate_contract_catalogs( contract_files=contract_files, contract_validator=contract_validator, ) if contract_errors: print(f"content validation failed with {contract_errors} error(s)", file=sys.stderr) return 1 e7m4_template_roster_err = _prototype_e7m4_template_roster_gate(contract_seen_ids) if e7m4_template_roster_err: print(e7m4_template_roster_err, file=sys.stderr) return 1 e7m4_template_freeze_err = _prototype_e7m4_template_freeze_gate(contract_id_to_row) if e7m4_template_freeze_err: print(e7m4_template_freeze_err, file=sys.stderr) return 1 e7m4_cross_ref_err = _prototype_e7m4_cross_ref_gate( contract_id_to_row, encounter_seen_ids, faction_seen_ids, skill_id_to_allowed_kinds, ) if e7m4_cross_ref_err: print(e7m4_cross_ref_err, file=sys.stderr) return 1 e7m4_band_cap_err = _prototype_e7m4_band_cap_gate(contract_id_to_row) if e7m4_band_cap_err: print(e7m4_band_cap_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(contract_files)} contract template 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(contract_seen_ids)} unique contract template 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())