"""Validate and summarize the Ghostlight corpus coverage ledger.""" from __future__ import annotations import argparse import json from collections import Counter, defaultdict from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_LEDGER = ROOT / "state" / "corpus-coverage.json" VALID_SCHEMA_VERSION = "ghostlight.corpus_coverage.v0" VALID_STATUSES = {"planned", "draft", "accepted_as_draft", "accepted", "needs_revision", "rejected", "superseded"} ACCEPTED_STATUSES = {"accepted", "accepted_as_draft"} VALID_TIMELINE_LANES = {"historical_grounded", "transition_grounded", "future_branch"} VALID_CANON_STATUSES = {"single_history", "branch_local", "promoted_branch", "template", "draft"} VALID_STORY_TYPES = { "founding_era_story", "day_in_the_life_story", "flashpoint_story", "movement_story", "future_branch_story", "trade_life_story", } VALID_REVIEW_STATUSES = {"not_run", "accepted", "accepted_as_draft", "needs_revision", "rejected", "blocked"} VALID_TRAINING_USABILITY = {"training_ready", "reference_only", "not_training_data"} REQUIRED_REVIEW_KEYS = {"schema", "narrative", "lore", "spatial", "visual", "if_artifact"} REQUIRED_ARTIFACTS_FOR_ACCEPTED = {"lore_digest", "agent_state", "coordinator", "ink", "training_sidecar"} class ValidationError(Exception): pass def load_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8-sig")) def require(condition: bool, message: str) -> None: if not condition: raise ValidationError(message) def require_string(value: Any, path: str) -> None: require(isinstance(value, str) and value.strip(), f"{path} must be a non-empty string") def require_string_array(value: Any, path: str, *, nonempty: bool = False) -> None: require(isinstance(value, list), f"{path} must be an array") require(not nonempty or bool(value), f"{path} must not be empty") for index, item in enumerate(value): require_string(item, f"{path}[{index}]") def require_keys(obj: dict[str, Any], keys: set[str], path: str) -> None: missing = sorted(keys - set(obj)) require(not missing, f"{path} missing required keys: {', '.join(missing)}") def validate_targets(targets: Any, path: str) -> None: require(isinstance(targets, dict), f"{path} must be an object") required = { "major_powers", "minor_powers", "movements", "pre_elysium_flashpoints", "defunct_absorbed_powers", "practical_first_broad_target", "major_power_requirement", "collision_requirement", } require_keys(targets, required, path) for key in ["major_powers", "minor_powers", "movements", "pre_elysium_flashpoints", "defunct_absorbed_powers"]: require(isinstance(targets[key], int) and targets[key] >= 0, f"{path}.{key} must be a non-negative integer") broad = targets["practical_first_broad_target"] require(isinstance(broad, dict), f"{path}.practical_first_broad_target must be an object") require_keys(broad, {"minimum", "maximum"}, f"{path}.practical_first_broad_target") require(isinstance(broad["minimum"], int) and broad["minimum"] >= 0, f"{path}.practical_first_broad_target.minimum must be a non-negative integer") require(isinstance(broad["maximum"], int) and broad["maximum"] >= broad["minimum"], f"{path}.practical_first_broad_target.maximum must be >= minimum") require_string(targets["major_power_requirement"], f"{path}.major_power_requirement") require_string(targets["collision_requirement"], f"{path}.collision_requirement") def validate_entry(entry: Any, path: str) -> list[str]: require(isinstance(entry, dict), f"{path} must be an object") required = { "fixture_id", "status", "timeline_lane", "story_type", "primary_subjects", "secondary_subjects", "collision_axes", "tonal_modes", "training_targets", "training_usability", "artifacts", "review", "economy_hooks", "open_lore_gaps", "notes", } require_keys(entry, required, path) require_string(entry["fixture_id"], f"{path}.fixture_id") require(entry["status"] in VALID_STATUSES, f"{path}.status is invalid") require(entry["timeline_lane"] in VALID_TIMELINE_LANES, f"{path}.timeline_lane is invalid") if "canon_status" in entry: require(entry["canon_status"] in VALID_CANON_STATUSES, f"{path}.canon_status is invalid") require(entry["story_type"] in VALID_STORY_TYPES, f"{path}.story_type is invalid") for optional in ["era", "location", "flashpoint", "notes"]: if optional in entry: require(isinstance(entry[optional], str), f"{path}.{optional} must be a string") require_string_array(entry["primary_subjects"], f"{path}.primary_subjects", nonempty=True) require_string_array(entry["secondary_subjects"], f"{path}.secondary_subjects") require_string_array(entry["collision_axes"], f"{path}.collision_axes", nonempty=True) require_string_array(entry["tonal_modes"], f"{path}.tonal_modes", nonempty=True) require_string_array(entry["training_targets"], f"{path}.training_targets", nonempty=True) usability = entry["training_usability"] require(isinstance(usability, dict), f"{path}.training_usability must be an object") require_keys(usability, {"overall"}, f"{path}.training_usability") for key, value in usability.items(): require_string(key, f"{path}.training_usability key") require(value in VALID_TRAINING_USABILITY, f"{path}.training_usability.{key} is invalid") require_string_array(entry["economy_hooks"], f"{path}.economy_hooks") require_string_array(entry["open_lore_gaps"], f"{path}.open_lore_gaps") artifacts = entry["artifacts"] require(isinstance(artifacts, dict), f"{path}.artifacts must be an object") for key, ref in artifacts.items(): require_string(key, f"{path}.artifacts key") require_string(ref, f"{path}.artifacts.{key}") require((ROOT / ref).exists(), f"{path}.artifacts.{key} does not exist: {ref}") if entry["status"] in ACCEPTED_STATUSES: missing_artifacts = sorted(REQUIRED_ARTIFACTS_FOR_ACCEPTED - set(artifacts)) require(not missing_artifacts, f"{path}.artifacts missing accepted fixture refs: {', '.join(missing_artifacts)}") review = entry["review"] require(isinstance(review, dict), f"{path}.review must be an object") for key, value in review.items(): require_string(key, f"{path}.review key") require(value in VALID_REVIEW_STATUSES, f"{path}.review.{key} is invalid") if entry["status"] in ACCEPTED_STATUSES: missing_review = sorted(REQUIRED_REVIEW_KEYS - set(review)) require(not missing_review, f"{path}.review missing accepted fixture review keys: {', '.join(missing_review)}") warnings: list[str] = [] if entry["status"] in ACCEPTED_STATUSES: weak_reviews = sorted(key for key, value in review.items() if value in {"not_run", "needs_revision", "rejected", "blocked"}) if weak_reviews: warnings.append(f"{entry['fixture_id']}: accepted fixture has weak review fields: {', '.join(weak_reviews)}") if len(set(entry["primary_subjects"] + entry["secondary_subjects"])) < 2: warnings.append(f"{entry['fixture_id']}: accepted fixture has fewer than two represented subjects") if entry["story_type"] == "future_branch_story" and entry.get("canon_status") != "branch_local": warnings.append(f"{entry['fixture_id']}: future_branch_story should usually be canon_status branch_local") if entry["training_usability"].get("overall") == "training_ready" and entry["status"] not in ACCEPTED_STATUSES: warnings.append(f"{entry['fixture_id']}: training_ready fixture is not accepted or accepted_as_draft") return warnings def validate_ledger(document: Any) -> tuple[list[dict[str, Any]], list[str]]: require(isinstance(document, dict), "ledger must be an object") require(document.get("schema_version") == VALID_SCHEMA_VERSION, "ledger schema_version is wrong") require_string(document.get("coverage_id"), "ledger.coverage_id") validate_targets(document.get("coverage_targets"), "ledger.coverage_targets") entries = document.get("entries") require(isinstance(entries, list), "ledger.entries must be an array") seen: set[str] = set() warnings: list[str] = [] for index, entry in enumerate(entries): entry_warnings = validate_entry(entry, f"ledger.entries[{index}]") fixture_id = entry["fixture_id"] require(fixture_id not in seen, f"ledger.entries[{index}].fixture_id is duplicated: {fixture_id}") seen.add(fixture_id) warnings.extend(entry_warnings) return entries, warnings def summarize(document: dict[str, Any], entries: list[dict[str, Any]], warnings: list[str]) -> str: targets = document["coverage_targets"] status_counts = Counter(entry["status"] for entry in entries) lane_counts = Counter(entry["timeline_lane"] for entry in entries) story_counts = Counter(entry["story_type"] for entry in entries) collision_counts = Counter(axis for entry in entries for axis in entry["collision_axes"]) tonal_counts = Counter(mode for entry in entries for mode in entry["tonal_modes"]) usability_counts = Counter(entry["training_usability"]["overall"] for entry in entries) subject_counts = Counter(subject for entry in entries for subject in entry["primary_subjects"] + entry["secondary_subjects"]) accepted_entries = [entry for entry in entries if entry["status"] in ACCEPTED_STATUSES] accepted_count = len(accepted_entries) target_min = targets["practical_first_broad_target"]["minimum"] target_max = targets["practical_first_broad_target"]["maximum"] lines = ["Ghostlight corpus coverage", ""] lines.append(f"Ledger: {document['coverage_id']}") lines.append(f"Entries: {len(entries)} total, {accepted_count} accepted/accepted-as-draft") lines.append(f"Practical first broad target: {target_min}-{target_max} accepted fixtures") if target_min: lines.append(f"Progress to minimum: {accepted_count}/{target_min} ({accepted_count / target_min:.1%})") lines.append("") def append_counter(title: str, counter: Counter[str]) -> None: lines.append(title) if counter: for key, count in counter.most_common(): lines.append(f"- {key}: {count}") else: lines.append("- none") lines.append("") append_counter("Status", status_counts) append_counter("Timeline lanes", lane_counts) append_counter("Story types", story_counts) append_counter("Top collision axes", collision_counts) append_counter("Tonal modes", tonal_counts) append_counter("Training usability", usability_counts) append_counter("Subjects represented", subject_counts) major_story_types: dict[str, set[str]] = defaultdict(set) for entry in entries: if entry["story_type"] in {"founding_era_story", "day_in_the_life_story"}: for subject in entry["primary_subjects"]: major_story_types[subject].add(entry["story_type"]) if major_story_types: lines.append("Major-power story coverage candidates") for subject, covered in sorted(major_story_types.items()): missing = sorted({"founding_era_story", "day_in_the_life_story"} - covered) if missing: lines.append(f"- {subject}: missing {', '.join(missing)}") else: lines.append(f"- {subject}: founding and day-in-life covered") lines.append("") lines.append("Gap signal") if accepted_count == 0: lines.append("- No accepted fixtures yet.") if accepted_count < target_min: lines.append(f"- Need {target_min - accepted_count} more accepted fixtures to reach the minimum broad target.") if not any(entry["timeline_lane"] == "future_branch" for entry in entries): lines.append("- No future_branch fixture coverage yet.") if not any(entry["story_type"] == "founding_era_story" for entry in entries): lines.append("- No founding-era story coverage yet.") if not any(entry["story_type"] == "day_in_the_life_story" for entry in entries): lines.append("- No day-in-the-life story coverage yet.") if not any(entry["economy_hooks"] for entry in entries): lines.append("- No economy hooks captured yet.") if warnings: lines.append("") lines.append("Warnings") for warning in warnings: lines.append(f"- {warning}") return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description="Validate and summarize the Ghostlight corpus coverage ledger.") parser.add_argument("path", nargs="?", type=Path, default=DEFAULT_LEDGER, help="Coverage ledger JSON path.") parser.add_argument("--json", action="store_true", help="Print the validated ledger JSON instead of a text summary.") args = parser.parse_args() document = load_json(args.path) entries, warnings = validate_ledger(document) if args.json: print(json.dumps(document, indent=2)) else: print(summarize(document, entries, warnings)) return 0 if __name__ == "__main__": raise SystemExit(main())