""" Dependency-free JSON Schema validator for Cannon Chain canon lifecycle. Validates: - form.state transitions (candidate -> survivor | archive_ghost) - audience_choice schema (survivor_id, stolen_trait, source_episode, source_issue, committed_at) - receipt schema (episode_id, form_id, provenance, signed_by, deltas) - signer allowlist (agent:vincent, agent:timmy, human:*) - Exactly one survivor per episode - Rejected forms promoted to archive_ghost - Illegal state transitions rejected Usage: from projects.slop_cannon.canon_validator import validate_canon result = validate_canon(canon_data) assert result.valid, result.errors """ import json import re from pathlib import Path from datetime import datetime CANON_PATH = Path(__file__).parent / "canon.json" VALID_STATES = {"candidate", "survivor", "archive_ghost"} VALID_TRANSITIONS = { "candidate": {"survivor", "archive_ghost"}, "survivor": set(), # terminal "archive_ghost": set(), # terminal } SIGNER_PATTERN = re.compile(r"^(agent:(?:vincent|timmy)|human:\w+)$") def validate_canon(data: dict) -> dict: """ Validate canon.json data against lifecycle schemas. Returns: {"valid": bool, "errors": list[str], "warnings": list[str]} """ errors = [] warnings = [] # --- Top-level structure --- if "schema_version" not in data: errors.append("Missing schema_version") if "forms" not in data: errors.append("Missing forms array") return {"valid": False, "errors": errors, "warnings": warnings} forms = data["forms"] if not isinstance(forms, list) or len(forms) == 0: errors.append("forms must be a non-empty array") return {"valid": False, "errors": errors, "warnings": warnings} form_ids = set() survivor_count = 0 form_id_map = {} for i, form in enumerate(forms): prefix = f"forms[{i}]" if "id" not in form: errors.append(f"{prefix}: missing id") continue fid = form["id"] form_id_map[fid] = i if fid in form_ids: errors.append(f"{prefix}: duplicate form id '{fid}'") form_ids.add(fid) # Required fields for field in ["power", "scar", "cost"]: if field not in form or not form[field]: errors.append(f"{prefix}: missing or empty '{field}'") # Palette if "palette" not in form or not isinstance(form["palette"], list): errors.append(f"{prefix}: missing or invalid palette") elif len(form["palette"]) < 3: errors.append(f"{prefix}: palette must have at least 3 colors, got {len(form['palette'])}") # Lifecycle state if "state" not in form: errors.append(f"{prefix}: missing lifecycle state") elif form["state"] not in VALID_STATES: errors.append(f"{prefix}: invalid state '{form['state']}', expected one of {VALID_STATES}") else: if form["state"] == "survivor": survivor_count += 1 # Exactly one survivor (or zero if no choice yet) if survivor_count > 1: errors.append(f"Too many survivors: {survivor_count} (expected 0 or 1)") # --- Episodes --- if "episodes" in data: episodes = data["episodes"] ready_count = sum(1 for ep in episodes if ep.get("state") == "ready") if ready_count > 1: errors.append(f"Too many ready episodes: {ready_count} (expected at most 1)") # Check ordering: first ready, rest locked found_ready = False for ep in episodes: if ep.get("state") == "ready": if found_ready: errors.append(f"Ready episode '{ep['id']}' after another ready episode") found_ready = True elif ep.get("state") != "locked": errors.append(f"Episode '{ep['id']}' has invalid state '{ep.get('state')}'") # --- Audience choice schema --- if data.get("audience_choice") is not None: choice = data["audience_choice"] for field in ["survivor_id", "stolen_trait", "source_episode", "source_issue", "committed_at"]: if field not in choice: errors.append(f"audience_choice: missing '{field}'") # survivor_id must reference a valid form if "survivor_id" in choice: if choice["survivor_id"] not in form_id_map: errors.append(f"audience_choice: survivor_id '{choice['survivor_id']}' not found in forms") # committed_at must be valid ISO timestamp if "committed_at" in choice: try: datetime.fromisoformat(choice["committed_at"]) except (ValueError, TypeError): errors.append(f"audience_choice: committed_at is not a valid ISO timestamp") # If choice is made, exactly one survivor required if survivor_count == 0: errors.append("audience_choice is set but no survivor found in forms") # --- Receipts --- if "receipts" in data: for i, receipt in enumerate(data["receipts"]): prefix = f"receipts[{i}]" for field in ["episode_id", "form_id", "provenance", "signed_by"]: if field not in receipt or not receipt[field]: errors.append(f"{prefix}: missing or empty '{field}'") # Signed-by allowlist if "signed_by" in receipt: if not SIGNER_PATTERN.match(receipt["signed_by"]): errors.append( f"{prefix}: invalid signed_by '{receipt['signed_by']}', " f"expected pattern: {SIGNER_PATTERN.pattern}" ) # Form_id must reference a valid form if "form_id" in receipt: if receipt["form_id"] not in form_id_map: errors.append(f"{prefix}: form_id '{receipt['form_id']}' not found in forms") # Deltas for delta in ["power_delta", "scar_delta", "cost_delta"]: if delta in receipt and not receipt[delta]: warnings.append(f"{prefix}: {delta} is present but empty") return { "valid": len(errors) == 0, "errors": errors, "warnings": warnings, } def validate_transition(form: dict, new_state: str) -> dict: """ Validate a state transition for a single form. Returns: {"valid": bool, "errors": list[str]} """ errors = [] current = form.get("state") if not current: errors.append("Form has no current state") return {"valid": False, "errors": errors} if current not in VALID_TRANSITIONS: errors.append(f"Invalid current state '{current}'") return {"valid": False, "errors": errors} if new_state not in VALID_STATES: errors.append(f"Invalid target state '{new_state}', expected one of {VALID_STATES}") return {"valid": False, "errors": errors} if new_state not in VALID_TRANSITIONS.get(current, set()): errors.append( f"Illegal transition: '{current}' -> '{new_state}'" ) return {"valid": False, "errors": errors} return {"valid": True, "errors": []} def load_canon(path: str = None) -> dict: """Load canon.json from the default path or a given path.""" target = Path(path) if path else CANON_PATH return json.loads(target.read_text()) def check_canon(path: str = None) -> dict: """Load and validate canon.json. Convenience entry point.""" data = load_canon(path) return validate_canon(data) if __name__ == "__main__": import sys result = check_canon() if result["valid"]: print(f"canon.json: VALID") if result["warnings"]: for w in result["warnings"]: print(f" WARNING: {w}") else: print(f"canon.json: INVALID") for e in result["errors"]: print(f" ERROR: {e}") sys.exit(1)