#!/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) """ from __future__ import annotations import json import sys from pathlib import Path from jsonschema import Draft202012Validator 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" SKILLS_DIR = REPO_ROOT / "content/skills" MASTERY_DIR = REPO_ROOT / "content/mastery" ITEMS_DIR = REPO_ROOT / "content/items" # 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" 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 _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 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 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) 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 seen_ids: dict[str, str] = {} id_to_category: dict[str, 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 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 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(seen_ids)} unique skill id(s), " f"{len(item_seen_ids)} unique item id(s), " f"{len(track_skill_ids)} mastery track(s)" ) return 0 if __name__ == "__main__": raise SystemExit(main())