#!/usr/bin/python3 """ Verify Emotional Beats Checks emotional beat schedule compliance: 1. 11 required beats scheduled per CONSTANTS.md 2. Each beat hit within ±2 episode tolerance 3. Beats not rushed (minimum spacing) 4. Missing beats flagged with recovery recommendations Usage: python3 verify_emotional_beats.py Example: python3 verify_emotional_beats.py ./leviathan Returns: - Exit code 0: All beats on track - Exit code 1: Critical issues found """ import json import sys from pathlib import Path # Add engine tools to path _SCRIPT_DIR = Path(__file__).parent.resolve() sys.path.insert(0, str(_SCRIPT_DIR)) sys.path.insert(0, str(_SCRIPT_DIR.parent.parent)) # CLAUDE_PROJECTS, for recoil.* from recoil.core.paths import ProjectPaths # Import from single source of truth # Emotional beat schedule — must match CONSTANTS.md → Emotional Beat Schedule (11 Required Beats) # These are story-structural constants. If CONSTANTS.md changes, update this dict. # The ±2 tolerance is standard for emotional beat timing. BEAT_SCHEDULE = { "FIRST_CRACK": {"target": 10, "tolerance": 2, "description": "First vulnerability shown"}, "THRESHOLD": {"target": 15, "tolerance": 2, "description": "Point of no return"}, "DEEPENING": {"target": 20, "tolerance": 2, "description": "Investment despite themselves"}, "VULNERABILITY": {"target": 26, "tolerance": 2, "description": "First expressed care"}, "MIDPOINT": {"target": 30, "tolerance": 2, "description": "World/self fundamentally different"}, "FRACTURE": {"target": 33, "tolerance": 2, "description": "Core belief challenged"}, "BETRAYAL_DOUBT": {"target": 36, "tolerance": 2, "description": "Trust tested"}, "COST": {"target": 42, "tolerance": 2, "description": "Choose relationship over goal"}, "ALL_IS_LOST": {"target": 45, "tolerance": 2, "description": "Lowest point"}, "RECONCILIATION": {"target": 50, "tolerance": 2, "description": "Connection survives crisis"}, "RESOLUTION": {"target": 60, "tolerance": 2, "description": "Primary ache resolved"} } MINIMUM_BEAT_SPACING = 3 # Minimum episodes between beats def load_orchestrator_state(project_path: Path) -> dict | None: """Load orchestrator state if exists.""" state_file = ProjectPaths.from_root(project_path).state_dir / "orchestrator_state.json" if not state_file.exists(): return None try: with open(state_file, 'r') as f: return json.load(f) except json.JSONDecodeError as e: print(f"Error: Invalid JSON in orchestrator_state.json: {e}") return None def verify_beats(state: dict) -> list: """ Verify emotional beat compliance. Returns list of issues found. """ issues = [] beat_map = state.get("emotional_beat_map", {}) position = state.get("position", {}) current_episode = position.get("last_completed_episode", 0) # Track beat timing for spacing check hit_beats = [] for beat_name, schedule in BEAT_SCHEDULE.items(): target = schedule["target"] tolerance = schedule["tolerance"] beat_data = beat_map.get(beat_name, {}) status = beat_data.get("status", "pending") actual_ep = beat_data.get("actual_episode") # Check 1: Beat should have been hit by now if status == "pending" and current_episode >= target + tolerance: issues.append({ "type": "missed_beat", "beat": beat_name, "target": target, "tolerance": tolerance, "current": current_episode, "severity": "error", "recommendation": f"Beat '{beat_name}' was due by Ep {target + tolerance} but still pending. Insert recovery in next available episode." }) # Check 2: Beat hit outside tolerance if status == "hit" and actual_ep: hit_beats.append((beat_name, actual_ep)) min_ep = target - tolerance max_ep = target + tolerance if actual_ep < min_ep: issues.append({ "type": "rushed_beat", "beat": beat_name, "target": target, "actual": actual_ep, "severity": "warning", "recommendation": f"Beat '{beat_name}' hit at Ep {actual_ep}, earlier than range ({min_ep}-{max_ep}). May feel unearned." }) elif actual_ep > max_ep: issues.append({ "type": "late_beat", "beat": beat_name, "target": target, "actual": actual_ep, "severity": "warning", "recommendation": f"Beat '{beat_name}' hit at Ep {actual_ep}, later than range ({min_ep}-{max_ep}). Arc may feel slow." }) # Check 3: Beat coming up soon (reminder) if status == "pending" and target - tolerance <= current_episode + 5 <= target: issues.append({ "type": "upcoming_beat", "beat": beat_name, "target": target, "current": current_episode, "severity": "info", "recommendation": f"Beat '{beat_name}' due in {target - current_episode} episodes (Ep {target} ±{tolerance})." }) # Check 4: Beat spacing (ensure beats aren't too close together) hit_beats_sorted = sorted(hit_beats, key=lambda x: x[1]) for i in range(1, len(hit_beats_sorted)): prev_beat, prev_ep = hit_beats_sorted[i-1] curr_beat, curr_ep = hit_beats_sorted[i] spacing = curr_ep - prev_ep if spacing < MINIMUM_BEAT_SPACING: issues.append({ "type": "cramped_beats", "beats": f"{prev_beat} and {curr_beat}", "spacing": spacing, "severity": "warning", "recommendation": f"Beats '{prev_beat}' (Ep {prev_ep}) and '{curr_beat}' (Ep {curr_ep}) are only {spacing} episodes apart. May feel rushed." }) return issues def main(): if len(sys.argv) < 2: print("Usage: python3 verify_emotional_beats.py ") print("Example: python3 verify_emotional_beats.py ./leviathan") sys.exit(1) project_path = Path(sys.argv[1]).resolve() if not project_path.exists(): print(f"Error: Project path does not exist: {project_path}") sys.exit(1) state = load_orchestrator_state(project_path) if not state: print(f"Error: No orchestrator_state.json found. Run init_orchestrator_state.py first.") sys.exit(1) position = state.get("position", {}) current_ep = position.get("last_completed_episode", 0) beat_map = state.get("emotional_beat_map", {}) print(f"\n{'='*60}") print(f"EMOTIONAL BEATS CHECK") print(f"{'='*60}") print(f"\nProject: {project_path.name}") print(f"Current position: Episode {current_ep}") # Count beats hit beats_hit = sum(1 for b in beat_map.values() if b.get("status") == "hit") print(f"Beats hit: {beats_hit}/11") issues = verify_beats(state) # Beat status timeline print(f"\n{'─'*60}") print(f"BEAT TIMELINE:") print(f"{'─'*60}") for beat_name, schedule in BEAT_SCHEDULE.items(): target = schedule["target"] beat_data = beat_map.get(beat_name, {}) status = beat_data.get("status", "pending") actual = beat_data.get("actual_episode") if status == "hit": delta = actual - target delta_str = f"+{delta}" if delta > 0 else str(delta) if delta < 0 else "±0" print(f" ● Ep {actual:2d} ({delta_str}): {beat_name} — {schedule['description']}") elif status == "pending": if current_ep >= target: print(f" ○ Ep {target:2d} [DUE]: {beat_name} — {schedule['description']}") else: print(f" ○ Ep {target:2d}: {beat_name} — {schedule['description']}") # Issues if issues: # Separate by severity errors = [i for i in issues if i["severity"] == "error"] warnings = [i for i in issues if i["severity"] == "warning"] infos = [i for i in issues if i["severity"] == "info"] if errors or warnings: print(f"\n{'─'*60}") print(f"ISSUES FOUND:") print(f"{'─'*60}") for issue in errors + warnings: severity_icon = "✗" if issue["severity"] == "error" else "⚠" print(f"\n {severity_icon} [{issue['type'].upper()}] {issue.get('beat', issue.get('beats', ''))}") print(f" {issue['recommendation']}") if infos: print(f"\n{'─'*60}") print(f"UPCOMING:") print(f"{'─'*60}") for issue in infos: print(f" ℹ {issue['recommendation']}") print(f"\n{'='*60}") # Return non-zero if any errors has_errors = len(errors) > 0 sys.exit(1 if has_errors else 0) else: print(f"\n{'─'*60}") print(f"✓ All emotional beats on track") print(f"{'='*60}\n") sys.exit(0) if __name__ == "__main__": main()