NEO-112: extend validate_content.py with E7M1 quest CI gates

scripts/validate_content.py

@@ -14,6 +14,7 @@ Validates:
 - 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)
 """
 
 from __future__ import annotations
@@ -39,6 +40,9 @@ 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"
 SKILLS_DIR = REPO_ROOT / "content/skills"
 MASTERY_DIR = REPO_ROOT / "content/mastery"
 ITEMS_DIR = REPO_ROOT / "content/items"
@@ -48,6 +52,7 @@ 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"
 
 # Slice 1 prototype lock (NEO-33): exact ids + category coverage after schema passes.
 PROTOTYPE_SLICE1_SKILL_IDS = frozenset({"salvage", "refine", "intrusion"})
@@ -152,6 +157,19 @@ PROTOTYPE_E5M3_REWARD_GRANTS: dict[str, int] = {
     "contract_handoff_token": 1,
 }
 
+# Epic 7 Slice 1 prototype lock (NEO-112): exact quest ids after schema passes.
+# Keep in sync with E7.M1 freeze table and future NEO-113 server loader.
+PROTOTYPE_E7M1_QUEST_IDS = frozenset(
+    {
+        "prototype_quest_gather_intro",
+        "prototype_quest_refine_intro",
+        "prototype_quest_combat_intro",
+        "prototype_quest_operator_chain",
+    }
+)
+PROTOTYPE_E7M1_CHAIN_QUEST_ID = "prototype_quest_operator_chain"
+PROTOTYPE_E7M1_CHAIN_TERMINAL_ITEM_ID = "contract_handoff_token"
+
 
 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."""
@@ -823,6 +841,236 @@ def _prototype_e5m3_encounter_cross_ref_gate(id_to_row: dict[str, dict]) -> str
     return None
 
 
+def _quest_def_validator() -> Draft202012Validator:
+    """Build a validator that resolves quest-def $ref to step and objective 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"))
+    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)),
+        ]
+    )
+    return Draft202012Validator(def_schema, registry=registry)
+
+
+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()
+                    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):
+                            obj_ids_seen: set[str] = set()
+                            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}",
+                                            file=sys.stderr,
+                                        )
+                                        errors += 1
+                                    else:
+                                        obj_ids_seen.add(oid)
+
+    return errors, seen_ids, id_to_row
+
+
+def _prototype_e7m1_quest_gate(seen_ids: dict[str, str]) -> str | None:
+    """Return a human-readable error if E7M1 quest contract fails, else None."""
+    ids = frozenset(seen_ids.keys())
+    if ids != PROTOTYPE_E7M1_QUEST_IDS:
+        return (
+            "error: prototype E7M1 expects exactly quest ids "
+            f"{sorted(PROTOTYPE_E7M1_QUEST_IDS)!r}, got {sorted(ids)!r}"
+        )
+    return None
+
+
+def _prototype_e7m1_quest_prerequisite_gate(id_to_row: dict[str, dict]) -> str | None:
+    """Return a human-readable error if prerequisite ids are unknown or cyclic, else None."""
+    known = frozenset(id_to_row.keys())
+    for qid, row in id_to_row.items():
+        prereqs = row.get("prerequisiteQuestIds")
+        if not isinstance(prereqs, list):
+            continue
+        for pid in prereqs:
+            if isinstance(pid, str) and pid not in known:
+                return (
+                    f"error: quest {qid!r}: prerequisiteQuestIds references unknown quest id {pid!r}"
+                )
+
+    visiting: set[str] = set()
+    visited: set[str] = set()
+
+    def dfs(node: str) -> str | None:
+        if node in visiting:
+            return f"error: cyclic prerequisiteQuestIds detected involving quest {node!r}"
+        if node in visited:
+            return None
+        visiting.add(node)
+        row = id_to_row.get(node, {})
+        prereqs = row.get("prerequisiteQuestIds", [])
+        if isinstance(prereqs, list):
+            for pid in prereqs:
+                if isinstance(pid, str):
+                    err = dfs(pid)
+                    if err:
+                        return err
+        visiting.remove(node)
+        visited.add(node)
+        return None
+
+    for qid in known:
+        err = dfs(qid)
+        if err:
+            return err
+    return None
+
+
+def _prototype_e7m1_quest_cross_ref_gate(id_to_row: dict[str, dict]) -> str | None:
+    """Return a human-readable error if objective catalog cross-refs fail, else None."""
+    for qid, row in id_to_row.items():
+        steps = row.get("steps")
+        if not isinstance(steps, list):
+            continue
+        for si, step in enumerate(steps):
+            if not isinstance(step, dict):
+                continue
+            objectives = step.get("objectives")
+            if not isinstance(objectives, list):
+                continue
+            for oi, obj in enumerate(objectives):
+                if not isinstance(obj, dict):
+                    continue
+                kind = obj.get("kind")
+                item_id = obj.get("itemId")
+                if isinstance(item_id, str) and kind in ("gather_item", "inventory_has_item"):
+                    if item_id not in PROTOTYPE_SLICE1_ITEM_IDS:
+                        return (
+                            f"error: quest {qid!r} steps[{si}] objectives[{oi}]: itemId {item_id!r} "
+                            "is not in the frozen prototype item catalog"
+                        )
+                recipe_id = obj.get("recipeId")
+                if isinstance(recipe_id, str) and kind == "craft_recipe":
+                    if recipe_id not in PROTOTYPE_SLICE3_RECIPE_IDS:
+                        return (
+                            f"error: quest {qid!r} steps[{si}] objectives[{oi}]: recipeId {recipe_id!r} "
+                            "is not in the frozen prototype recipe catalog"
+                        )
+                encounter_id = obj.get("encounterId")
+                if isinstance(encounter_id, str) and kind == "encounter_complete":
+                    if encounter_id not in PROTOTYPE_E5M3_ENCOUNTER_IDS:
+                        return (
+                            f"error: quest {qid!r} steps[{si}] objectives[{oi}]: encounterId "
+                            f"{encounter_id!r} is not in the frozen prototype encounter catalog"
+                        )
+    return None
+
+
+def _prototype_e7m1_chain_terminal_gate(id_to_row: dict[str, dict]) -> str | None:
+    """Return a human-readable error if operator chain terminal step fails, else None."""
+    chain = id_to_row.get(PROTOTYPE_E7M1_CHAIN_QUEST_ID)
+    if not isinstance(chain, dict):
+        return f"error: missing chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r}"
+    steps = chain.get("steps")
+    if not isinstance(steps, list) or len(steps) == 0:
+        return f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} must have at least one step"
+    last_step = steps[-1]
+    if not isinstance(last_step, dict):
+        return f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal step must be an object"
+    objectives = last_step.get("objectives")
+    if not isinstance(objectives, list) or len(objectives) != 1:
+        return (
+            f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal step must have "
+            "exactly one objective"
+        )
+    obj = objectives[0]
+    if not isinstance(obj, dict):
+        return (
+            f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal objective must be an object"
+        )
+    if obj.get("kind") != "inventory_has_item":
+        return (
+            f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal objective kind must be "
+            "'inventory_has_item'"
+        )
+    if obj.get("itemId") != PROTOTYPE_E7M1_CHAIN_TERMINAL_ITEM_ID:
+        return (
+            f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal itemId must be "
+            f"{PROTOTYPE_E7M1_CHAIN_TERMINAL_ITEM_ID!r}, got {obj.get('itemId')!r}"
+        )
+    if obj.get("quantity") != 1:
+        return (
+            f"error: chain quest {PROTOTYPE_E7M1_CHAIN_QUEST_ID!r} terminal quantity must be 1, "
+            f"got {obj.get('quantity')!r}"
+        )
+    return None
+
+
 def _prototype_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:
@@ -1158,6 +1406,15 @@ def main() -> int:
     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
 
     skill_schema = json.loads(SKILL_SCHEMA.read_text(encoding="utf-8"))
     skill_validator = Draft202012Validator(skill_schema)
@@ -1179,6 +1436,7 @@ def main() -> int:
     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()
 
     if not SKILLS_DIR.is_dir():
         print(f"error: missing directory {SKILLS_DIR}", file=sys.stderr)
@@ -1271,6 +1529,15 @@ def main() -> int:
         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
+
     seen_ids: dict[str, str] = {}
     id_to_category: dict[str, str] = {}
     errors = 0
@@ -1536,6 +1803,34 @@ def main() -> int:
         print(e5m3_encounter_cross_ref_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
+
+    e7m1_quest_err = _prototype_e7m1_quest_gate(quest_seen_ids)
+    if e7m1_quest_err:
+        print(e7m1_quest_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
+
     print(
         "content OK: "
         f"{len(skill_files)} skill catalog file(s), "
@@ -1549,6 +1844,7 @@ def main() -> int:
         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(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), "
@@ -1557,6 +1853,7 @@ def main() -> int:
         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(quest_seen_ids)} unique quest id(s), "
         f"{len(track_skill_ids)} mastery track(s)"
     )
     return 0