Files
jpmschweitzerandClaude Opus 5 7f20bd303b feat(config): T-1282 — the validate domain, and a move that broke a root
reach validate content / checklist / ron / name-collisions. The three old
scripts are retired, their make targets with them.

Print statements go through the logging sink rather than a collector. The
validators emit their findings as console events as they run, so a long content
validation streams instead of going quiet and dumping at the end — the message
strings and their order are unchanged, only the destination. That also
satisfies the conformance rule forbidding print() in the package, which is what
forced the question.

validate-ron was three languages deep: bash dispatching on a flag, a Python
heredoc doing collision detection, cargo run for schema validation. Logic
embedded in a shell string cannot be imported, tested, or found by anything
that indexes Python, so it became Python; the cargo call became a guarded exec.
It also split into two verbs, because --check-name-collisions answered a
different question from the default path: whether the SET of cultures is
coherent, versus whether ONE file is well-formed.

The move broke something, quietly, which is the point of doing these one at a
time. validate-checklist computed ROOT as Path(__file__).parent.parent — the
repo root while it lived at tooling/validate-checklist, and tooling/domains
once moved. Both its schema and gauntlet paths silently repointed at nothing,
the gauntlet directory "did not exist", and it reported success having checked
zero files. Caught by running it beside the original: old exit 1, new exit 0.
Now config.repo_root(), and load_schema raises ReachError instead of calling
sys.exit, which a service must not do.

Parity on the live tree: content reproduces the original byte for byte
including its counts, name-collisions likewise. Tests pin what those runs
cannot reach — the detection path, since the repo currently has no collisions,
and the argument errors.

Two things found and left alone: validate-content FAILS on the live tree with
13 missing schemas, pre-existing and unrelated to this port; and the ticket's
claim that validate-content sits in the pre-commit hook is wrong — that hook
runs only check-fact-ids and pql decisions validate, so there was no shared
edit to coordinate.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-02 12:48:56 +02:00

571 lines
22 KiB
Python
Executable File

#!/usr/bin/env python3
"""Validate content YAML files against their JSON schemas and cross-references.
Pass 1: Schema validation (per-file against JSON Schema)
Pass 2: Cross-reference validation (9 checks across all files)
Schema mapping is by directory context:
campaign.yaml -> campaign.schema.json
system.yaml -> system.schema.json
station.yaml -> station.schema.json
district.yaml -> district.schema.json
npcs/*.yaml -> npc-profile.schema.json
locations/*.yaml -> location.schema.json
triangles/*.yaml -> triangle.schema.json
dialogue/**/*.yaml -> dialogue-pool.schema.json
monologue/**/*.yaml -> monologue-pool.schema.json
routines/*.yaml -> routine.schema.json
Files under global/ and content.yaml are skipped (no schema yet).
Exit code 0 = all valid, 1 = validation errors found.
"""
import json
import re
from pathlib import Path
import jsonschema
import yaml
from tooling.core import config, console
CONTENT_DIR = config.path("server", "content")
SCHEMA_DIR = CONTENT_DIR / "_schema"
# Map directory parent name (or filename) to schema file
FILENAME_SCHEMAS = {
"campaign.yaml": "campaign.schema.json",
"system.yaml": "system.schema.json",
"station.yaml": "station.schema.json",
"district.yaml": "district.schema.json",
}
DIR_SCHEMAS = {
"npcs": "npc-profile.schema.json",
"locations": "location.schema.json",
"triangles": "triangle.schema.json",
"dialogue": "dialogue-pool.schema.json",
"monologue": "monologue-pool.schema.json",
"routines": "routine.schema.json",
}
def resolve_schema(yaml_path: Path) -> Path | None:
"""Determine which schema applies to a content YAML file."""
name = yaml_path.name
if name in FILENAME_SCHEMAS:
return SCHEMA_DIR / FILENAME_SCHEMAS[name]
# Walk up parents to find a matching directory name
rel = yaml_path.relative_to(CONTENT_DIR)
for part in reversed(rel.parts[:-1]):
if part in DIR_SCHEMAS:
return SCHEMA_DIR / DIR_SCHEMAS[part]
return None
def _rel(path: Path) -> str:
"""Return path relative to CONTENT_DIR for display."""
try:
return str(path.relative_to(CONTENT_DIR))
except ValueError:
return str(path)
def _load_yaml(path: Path) -> dict | list | None:
"""Load a YAML file, returning None for empty/comment-only files."""
try:
with open(path) as f:
return yaml.safe_load(f)
except yaml.YAMLError:
return None
def _district_dir(path: Path) -> Path | None:
"""Find the district directory containing this file."""
parts = path.resolve().parts
for i, part in enumerate(parts):
if part == "districts" and i + 1 < len(parts):
return Path(*parts[: i + 2])
return None
class ContentIndex:
"""Builds an index of all defined entities for cross-referencing."""
def __init__(self, campaigns_dir: Path):
self.campaigns_dir = campaigns_dir
self.npcs: dict[str, Path] = {} # canonical_id -> defining file
self.locations: dict[str, set] = {} # district_dir str -> set of location slugs
self.triangle_slugs: dict[str, set] = {} # district_dir str -> set of triangle file stems
self.fact_ids: set[str] = set() # from knowledge catalogs
self.dialogue_files: list[tuple[Path, dict]] = []
self.monologue_files: list[tuple[Path, dict]] = []
self.npc_files: list[tuple[Path, dict]] = []
self.district_files: list[tuple[Path, dict]] = []
def build(self):
"""Scan all content files and populate the index."""
self._scan_npcs()
self._scan_districts()
self._scan_triangles()
self._scan_knowledge()
self._scan_dialogue()
self._scan_monologue()
def _scan_npcs(self):
for path in sorted(self.campaigns_dir.rglob("npcs/*.yaml")):
data = _load_yaml(path)
if data and isinstance(data, dict):
self.npc_files.append((path, data))
cid = data.get("canonical_id")
if cid:
self.npcs[cid] = path
def _scan_districts(self):
for path in sorted(self.campaigns_dir.rglob("district.yaml")):
data = _load_yaml(path)
if data and isinstance(data, dict):
self.district_files.append((path, data))
district_dir = str(path.parent.resolve())
locs = data.get("locations", [])
if isinstance(locs, list):
self.locations[district_dir] = set(locs)
def _scan_triangles(self):
for path in sorted(self.campaigns_dir.rglob("triangles/*.yaml")):
data = _load_yaml(path)
if data and isinstance(data, dict):
district_dir = str(path.parent.parent.resolve())
if district_dir not in self.triangle_slugs:
self.triangle_slugs[district_dir] = set()
self.triangle_slugs[district_dir].add(path.stem)
def _scan_knowledge(self):
knowledge_dir = CONTENT_DIR / "global" / "knowledge"
if not knowledge_dir.exists():
return
for path in sorted(knowledge_dir.glob("*.yaml")):
if path.name == "entity-attributes.yaml":
continue
data = _load_yaml(path)
if not data or not isinstance(data, dict):
# Stub catalogs (comment-only or flat key-value) don't parse
# as dicts. Fall back to regex extraction of fact_id lines.
try:
with open(path) as f:
for line in f:
line = line.strip()
if line.startswith("#") or not line:
continue
m = re.match(r"fact_id:\s*[\"']?([^\"'#]+)", line)
if m:
self.fact_ids.add(m.group(1).strip())
except OSError:
pass
continue
# Handle structured catalogs
self._extract_fact_ids_from_dict(data)
def _extract_fact_ids_from_dict(self, data, prefix=""):
"""Recursively extract fact_id values from knowledge catalog structures."""
if isinstance(data, dict):
for key, val in data.items():
if key == "fact_id" and isinstance(val, str):
self.fact_ids.add(val)
elif isinstance(val, (dict, list)):
self._extract_fact_ids_from_dict(val)
elif isinstance(data, list):
for item in data:
self._extract_fact_ids_from_dict(item)
def _scan_dialogue(self):
for path in sorted(self.campaigns_dir.rglob("dialogue/**/*.yaml")):
if path.name.startswith("."):
continue
data = _load_yaml(path)
if data and isinstance(data, dict):
self.dialogue_files.append((path, data))
def _scan_monologue(self):
for path in sorted(self.campaigns_dir.rglob("monologue/**/*.yaml")):
if path.name.startswith("."):
continue
data = _load_yaml(path)
if data and isinstance(data, dict):
self.monologue_files.append((path, data))
def validate_references(self) -> tuple[int, int]:
"""Check all cross-references. Returns (error_count, warning_count)."""
errors = 0
warnings = 0
errors += self._check_1_canonical_id_uniqueness()
errors += self._check_2_relationship_targets()
errors += self._check_3_location_slugs()
errors += self._check_4_dialogue_locations()
errors += self._check_5_fact_ids()
errors += self._check_6_triangle_membership()
warnings += self._check_7_npc_count()
errors += self._check_8_dialogue_line_ids()
warnings += self._check_9_bidirectional_relationships()
return errors, warnings
def _check_1_canonical_id_uniqueness(self) -> int:
"""Check that canonical_id values are unique across all NPC files."""
errors = 0
seen: dict[str, Path] = {}
for path, data in self.npc_files:
cid = data.get("canonical_id")
if not cid:
continue
if cid in seen:
console.event(f'XREF ERROR: duplicate canonical_id "{cid}"')
console.event(f" Defined in: {_rel(seen[cid])}")
console.event(f" Duplicate in: {_rel(path)}")
errors += 1
else:
seen[cid] = path
return errors
def _check_2_relationship_targets(self) -> int:
"""Check that all relationship targets resolve to known canonical_ids."""
errors = 0
for path, data in self.npc_files:
rels = data.get("relationships", [])
if not isinstance(rels, list):
continue
for rel in rels:
if not isinstance(rel, dict):
continue
target = rel.get("target")
if target and target not in self.npcs:
known = sorted(self.npcs.keys())
console.event(f'XREF ERROR: unresolved relationship target "{target}"')
console.event(f" In: {_rel(path)}")
console.event(f" Known canonical_ids: {', '.join(known[:10])}"
+ (f" (and {len(known) - 10} more)" if len(known) > 10 else ""))
errors += 1
return errors
def _check_3_location_slugs(self) -> int:
"""Check that district location slugs correspond to actual location files."""
errors = 0
for path, data in self.district_files:
locs = data.get("locations", [])
if not isinstance(locs, list):
continue
locations_dir = path.parent / "locations"
for slug in locs:
expected = locations_dir / f"{slug}.yaml"
if not expected.exists():
console.event(f'XREF ERROR: location slug "{slug}" not found')
console.event(f" In: {_rel(path)}")
console.event(f" Expected file: {_rel(expected)}")
errors += 1
return errors
def _check_4_dialogue_locations(self) -> int:
"""Check that dialogue file locations match district locations."""
errors = 0
for path, data in self.dialogue_files:
location = data.get("location")
if not location:
continue
district_dir = _district_dir(path)
if not district_dir:
continue
district_key = str(district_dir)
district_locs = self.locations.get(district_key, set())
if not district_locs:
console.event(f'XREF ERROR: dialogue location "{location}" references district with no locations declared')
console.event(f" In: {_rel(path)}")
console.event(f" District: {_rel(district_dir / 'district.yaml')}")
errors += 1
elif location not in district_locs:
console.event(f'XREF ERROR: dialogue location "{location}" not in district')
console.event(f" In: {_rel(path)}")
console.event(f" District locations: {', '.join(sorted(district_locs))}")
errors += 1
return errors
def _check_5_fact_ids(self) -> int:
"""Check that all referenced fact_ids resolve to knowledge catalogs."""
if not self.fact_ids:
console.event(" Check 5 (fact_ids): SKIPPED — no canonical fact_ids in knowledge catalogs yet")
return 0
errors = 0
# Collect all fact_id references from dialogue, monologue, NPC, and item files
refs = self._collect_all_fact_id_refs()
for fact_id, sources in sorted(refs.items()):
if fact_id not in self.fact_ids:
console.event(f'XREF ERROR: unknown fact_id "{fact_id}"')
for src in sources[:3]:
console.event(f" In: {_rel(src)}")
if len(sources) > 3:
console.event(f" ...and {len(sources) - 3} more files")
console.event(f" Canonical fact_ids: {len(self.fact_ids)} defined")
errors += 1
return errors
def _collect_all_fact_id_refs(self) -> dict[str, list[Path]]:
"""Collect all fact_id references from content files."""
refs: dict[str, list[Path]] = {}
def add_ref(fact_id: str, path: Path):
if fact_id:
refs.setdefault(fact_id, []).append(path)
# Dialogue: lines[].knowledge_grant.fact_id
for path, data in self.dialogue_files:
for line in data.get("lines", []):
if not isinstance(line, dict):
continue
kg = line.get("knowledge_grant")
if isinstance(kg, dict):
add_ref(kg.get("fact_id", ""), path)
# Monologue: entries with prerequisites.facts[].fact_id
for path, data in self.monologue_files:
self._walk_fact_id_refs(data, path, add_ref)
# NPC: information.knows[]
for path, data in self.npc_files:
info = data.get("information", {})
if isinstance(info, dict):
knows = info.get("knows", [])
if isinstance(knows, list):
for fid in knows:
if isinstance(fid, str):
add_ref(fid, path)
# Items: scan for fact_id in any nested structure
for path in sorted(self.campaigns_dir.rglob("items/*.yaml")):
data = _load_yaml(path)
if data:
self._walk_fact_id_refs(data, path, add_ref)
return refs
def _walk_fact_id_refs(self, data, path: Path, add_ref):
"""Recursively find fact_id references in nested structures."""
if isinstance(data, dict):
if "fact_id" in data and isinstance(data["fact_id"], str):
add_ref(data["fact_id"], path)
for val in data.values():
if isinstance(val, (dict, list)):
self._walk_fact_id_refs(val, path, add_ref)
elif isinstance(data, list):
for item in data:
if isinstance(item, (dict, list)):
self._walk_fact_id_refs(item, path, add_ref)
def _check_6_triangle_membership(self) -> int:
"""Check that NPC triangle_membership slugs correspond to triangle files."""
errors = 0
for path, data in self.npc_files:
memberships = data.get("triangle_membership", [])
if not isinstance(memberships, list):
continue
district_dir = _district_dir(path)
if not district_dir:
continue
district_key = str(district_dir)
available = self.triangle_slugs.get(district_key, set())
for slug in memberships:
if not isinstance(slug, str):
continue
if slug not in available:
console.event(f'XREF ERROR: triangle "{slug}" not found')
console.event(f" In: {_rel(path)}")
if available:
console.event(f" Available triangles: {', '.join(sorted(available))}")
else:
console.event(f" No triangles found in {_rel(district_dir / 'triangles')}")
errors += 1
return errors
def _check_7_npc_count(self) -> int:
"""Check that district npc_count matches actual NPC file count."""
warnings = 0
for path, data in self.district_files:
declared = data.get("npc_count")
if declared is None:
continue
npcs_dir = path.parent / "npcs"
if not npcs_dir.exists():
continue
actual = len(list(npcs_dir.glob("*.yaml")))
if declared != actual:
console.event("XREF WARNING: npc_count mismatch")
console.event(f" Declared: {declared}")
console.event(f" Actual NPC files: {actual}")
console.event(f" In: {_rel(path)}")
warnings += 1
return warnings
def _check_8_dialogue_line_ids(self) -> int:
"""Check that dialogue line IDs are unique across all dialogue files."""
errors = 0
# Cross-file uniqueness: same line ID in two files is an error
global_seen: dict[str, Path] = {}
for path, data in self.dialogue_files:
lines = data.get("lines", [])
if not isinstance(lines, list):
continue
local_seen: dict[str, int] = {}
for i, line in enumerate(lines, 1):
if not isinstance(line, dict):
continue
line_id = line.get("id")
if not line_id:
continue
# Per-file duplicate
if line_id in local_seen:
console.event(f'XREF ERROR: duplicate dialogue line id "{line_id}"')
console.event(f" In: {_rel(path)}")
console.event(f" First: line {local_seen[line_id]}")
console.event(f" Duplicate: line {i}")
errors += 1
else:
local_seen[line_id] = i
# Cross-file duplicate
if line_id in global_seen and global_seen[line_id] != path:
console.event(f'XREF ERROR: dialogue line id "{line_id}" used in multiple files')
console.event(f" First: {_rel(global_seen[line_id])}")
console.event(f" Also in: {_rel(path)}")
errors += 1
elif line_id not in global_seen:
global_seen[line_id] = path
return errors
def _check_9_bidirectional_relationships(self) -> int:
"""Check that NPC relationships have reciprocal entries.
WARNING severity: asymmetric relationships are valid by design (D-034
allows one-directional awareness). This check flags them for review,
not as errors."""
warnings = 0
# Build map: canonical_id -> set of relationship targets
npc_rels: dict[str, set[str]] = {}
for _path, data in self.npc_files:
cid = data.get("canonical_id")
if not cid:
continue
rels = data.get("relationships", [])
if isinstance(rels, list):
targets = set()
for rel in rels:
if isinstance(rel, dict) and "target" in rel:
targets.add(rel["target"])
npc_rels[cid] = targets
# Check reciprocity
checked = set()
for cid, targets in npc_rels.items():
for target in targets:
pair = tuple(sorted([cid, target]))
if pair in checked:
continue
checked.add(pair)
if target in self.npcs and cid not in npc_rels.get(target, set()):
console.event(f"XREF WARNING: {cid} has relationship to {target} but {target} has no reciprocal entry")
warnings += 1
return warnings
def schema_validate(campaigns_dir: Path) -> tuple[int, int, int]:
"""Pass 1: Schema validation. Returns (validated, skipped, errors)."""
errors = 0
validated = 0
skipped = 0
schema_cache: dict[str, dict] = {}
yaml_files = sorted(campaigns_dir.rglob("*.yaml"))
if not yaml_files:
console.event("No YAML files found under campaigns/")
return 0, 0, 1
for yaml_path in yaml_files:
schema_path = resolve_schema(yaml_path)
if schema_path is None:
skipped += 1
continue
if not schema_path.exists():
console.event(f"MISSING SCHEMA: {schema_path.name} for {yaml_path.relative_to(CONTENT_DIR)}")
errors += 1
continue
schema_key = str(schema_path)
if schema_key not in schema_cache:
with open(schema_path) as f:
schema_cache[schema_key] = json.load(f)
schema = schema_cache[schema_key]
try:
with open(yaml_path) as f:
data = yaml.safe_load(f)
except yaml.YAMLError as e:
console.event(f"YAML ERROR: {yaml_path.relative_to(CONTENT_DIR)}: {e}")
errors += 1
continue
if data is None:
skipped += 1
continue
try:
jsonschema.validate(instance=data, schema=schema)
validated += 1
except jsonschema.ValidationError as e:
rel = yaml_path.relative_to(CONTENT_DIR)
console.event(f"INVALID: {rel}")
console.event(f" Schema: {schema_path.name}")
console.event(f" Error: {e.message}")
if e.absolute_path:
console.event(f" Path: {'.'.join(str(p) for p in e.absolute_path)}")
errors += 1
return validated, skipped, errors
def validate() -> int:
"""Run both passes. Returns the exit code.
Replaces the old main(). Same logic, same messages, in the same order — but
they go out as EVENTS through core/console rather than through print, so
they stream as the validation runs, carry the invocation's job id, and are
rendered by the sink. A service does not decide to print (D-263).
"""
campaigns_dir = CONTENT_DIR / "campaigns"
if not campaigns_dir.exists():
console.event(f"No campaigns directory at {campaigns_dir}")
return 1
# Pass 1: Schema validation
validated, skipped, schema_errors = schema_validate(campaigns_dir)
console.event(f"\nPass 1 (schema): {validated} validated, {skipped} skipped, {schema_errors} errors")
if schema_errors > 0:
console.event(f"\nSchema validation failed ({schema_errors} errors) -- skipping cross-references")
return 1
# Pass 2: Cross-reference validation
console.event("\nPass 2 (cross-references):")
index = ContentIndex(campaigns_dir)
index.build()
xref_errors, xref_warnings = index.validate_references()
total_errors = schema_errors + xref_errors
console.event(f"\nValidated {validated} files: {total_errors} errors, {xref_warnings} warnings")
return (1 if total_errors > 0 else 0)