#!/usr/bin/python3 """ Quality Gate - Dramatic Infrastructure Validation Runs AFTER validate_batch.py passes mechanical V12 checks. Validates dramatic quality: Tier 1: PATTERN VARIETY (HARD FAIL) - No 4+ consecutive mid-action cliffhangers (max 3 allowed) - No 4+ consecutive aftermath cliffhangers (max 3 allowed) - No 4+ consecutive silent hooks (max 3 allowed) - No 4+ consecutive dialogue hooks (max 3 allowed) Tier 2: DRAMATIC BEATS (HARD FAIL) - Emotional beats present at scheduled episodes - Continuity maintained (episode continues from previous cliffhanger) - MUST CONTAIN items present in required episodes Tier 3: TRACKING (Report Only) - Hook ratio (silent vs dialogue) - Cliffhanger ratio (mid-action vs aftermath) - Thread opportunities Usage: python3 quality_gate.py Example: python3 quality_gate.py ./leviathan 1 Returns: - Exit code 0: All quality checks pass - Exit code 1: One or more quality checks fail (must regenerate) """ import json import sys import re from pathlib import Path # Import constants from engine (single source of truth) try: _SCRIPT_DIR = Path(__file__).parent.resolve() sys.path.insert(0, str(_SCRIPT_DIR.parent.parent / 'tools')) from engine_constants import ( GENERATION_BATCH_SIZE, MAX_CONSECUTIVE_SAME_TYPE, PILOT_EPISODE_COUNT, ANTHROPIC_HAIKU, get_anthropic_client, call_anthropic, parse_llm_field, extract_script_content, ) except ImportError: GENERATION_BATCH_SIZE = 5 MAX_CONSECUTIVE_SAME_TYPE = 3 PILOT_EPISODE_COUNT = 10 ANTHROPIC_HAIKU = "claude-haiku-4-5-20251001" get_anthropic_client = lambda: None call_anthropic = lambda client, model, prompt, max_tokens=200: None parse_llm_field = lambda result, field, expected=None: None extract_script_content = lambda content: content # Emotional beat schedule - target episodes with ±2 tolerance (per CONSTANTS.md) # Format: (target_episode, beat_name, tolerance) EMOTIONAL_BEAT_SCHEDULE = [ (10, "First Crack", 2), (15, "Threshold", 2), (20, "Deepening", 2), (26, "Vulnerability", 2), (30, "Midpoint", 2), (32, "Fracture", 2), # CONSTANTS.md says 32-33 (33, "Fracture continues", 2), (36, "Betrayal/Doubt", 2), (42, "Cost", 2), (45, "All Is Lost", 2), (50, "Reconciliation", 2), (59, "Final emotional beat", 2), (60, "Resolution", 2), ] # Maximum consecutive same-type before violation # Per CONSTANTS.md: "Max 3 consecutive allowed; 4+ is violation" # MAX_CONSECUTIVE_SAME_TYPE = 3 from engine_constants, so violation threshold is 3+1=4 MAX_CONSECUTIVE = MAX_CONSECUTIVE_SAME_TYPE + 1 # Pilot window: Episodes 1-10 where cliffhanger variety is suspended # Per CONSTANTS.md: "During pilot (Ep 1-10): All MID-ACTION cliffhangers allowed" PILOT_WINDOW_END = PILOT_EPISODE_COUNT def extract_script_content(content): """Extract the fountain script content from the episode file. Strips footer metadata (CLIFFHANGER TYPE, HOOK TYPE, NEXT lines) that appears after the final --- separator. """ fountain_match = re.search(r'```fountain\s*(.*?)```', content, re.DOTALL) if fountain_match: return fountain_match.group(1) script_match = re.search(r'## SCRIPT\s*(.*?)(?=##|$)', content, re.DOTALL) if script_match: return script_match.group(1) # Strip footer metadata after final --- separator parts = content.rsplit('\n---\n', 1) return parts[0] if len(parts) > 1 else content def extract_cliffhanger_metadata(content): """ Extract cliffhanger type from explicit metadata. Supports formats: - New: **CLIFFHANGER TYPE:** Mid-Action (M-CF) - Confrontation - Legacy: **CLIFFHANGER TYPE:** MidAction (M) Returns (main_type, subtype_code, subtype_name) or (None, None, None) if not found. """ # Try new format first: Mid-Action (M-CF) - Confrontation new_match = re.search( r'\*\*CLIFFHANGER TYPE:\*\*\s*(Mid-Action|Aftermath)\s*\(([A-Z]-[A-Z]{2})\)\s*-\s*(.+?)(?:\n|\*\*)', content, re.IGNORECASE ) if new_match: main_type = new_match.group(1).lower().replace('-', '') # "midaction" or "aftermath" if 'mid' in main_type: main_type = 'mid-action' else: main_type = 'aftermath' return main_type, new_match.group(2), new_match.group(3).strip() # Try legacy format: MidAction (M) or Aftermath (A) legacy_match = re.search( r'\*\*CLIFFHANGER TYPE:\*\*\s*(\w+)\s*\((\w)\)', content ) if legacy_match: type_name = legacy_match.group(1).lower() code = legacy_match.group(2) # Map legacy codes to main types if code == 'M' or 'mid' in type_name: return 'mid-action', code, type_name elif code in ['A', 'C', 'R'] or 'aftermath' in type_name: # C (Consequence) and R (Revelation) are Aftermath subtypes return 'aftermath', code, type_name else: return 'mid-action', code, type_name # Default return None, None, None def detect_cliffhanger_type(script, full_content=None): """ Detect if cliffhanger is MID-ACTION or AFTERMATH. First tries to read explicit metadata, then falls back to heuristic detection. MID-ACTION: Ends in the middle of action (trigger pull, fall, attack) AFTERMATH: Ends on reaction, quip, silence after action """ # Try explicit metadata first if full_content: main_type, subtype_code, subtype_name = extract_cliffhanger_metadata(full_content) if main_type: return main_type, f"Metadata: {main_type} ({subtype_code}) - {subtype_name}" # Fall back to heuristic detection # Find THE CLIFFHANGER section cliff_match = re.search( r'#.*THE CLIFFHANGER.*?\n(.*?)(?=#|\Z)', script, re.DOTALL | re.IGNORECASE ) if not cliff_match: return "unknown", "No CLIFFHANGER section found" cliff_content = cliff_match.group(1).strip() lines = [l.strip() for l in cliff_content.split('\n') if l.strip()] if not lines: return "unknown", "Empty CLIFFHANGER section" # Get last meaningful line (not metadata or empty) last_line = None for line in reversed(lines): if not line.startswith('**') and not line.startswith('---'): last_line = line break if not last_line: return "unknown", "No content in CLIFFHANGER" # Aftermath indicators: dialogue, reaction descriptions, quiet moments aftermath_patterns = [ r'[A-Z]{2,}$', # Ends with character name (next line is dialogue) r'\.\s*$', # Ends with period (completed thought) r'["\'].*["\']', # Contains quoted dialogue r'(smiles|nods|turns|walks away|silence|stares|looks|whispers)', # Reaction verbs r'(beat|pause|moment|quietly)', # Quiet indicators ] # Mid-action indicators: incomplete action, cut in motion midaction_patterns = [ r'(—|--|\.\.\.)\s*$', # Ends with dash or ellipsis (interrupted) r'(fires|shoots|falls|crashes|explodes|lunges|strikes)', # Active verbs r'(reaching|running|falling|swinging|pulling)', # -ing verbs (in motion) r'(just as|before|the moment)', # Cut timing words r'SMASH CUT|CUT TO BLACK', # Abrupt cuts ] last_line_lower = last_line.lower() # Check for mid-action indicators for pattern in midaction_patterns: if re.search(pattern, last_line, re.IGNORECASE): return "mid-action", f"Detected: {last_line[:50]}..." # Check for aftermath indicators for pattern in aftermath_patterns: if re.search(pattern, last_line, re.IGNORECASE): return "aftermath", f"Detected: {last_line[:50]}..." # Default to mid-action if unclear (conservative) return "mid-action", f"Default (unclear): {last_line[:50]}..." def detect_hook_type(script): """ Detect if hook is SILENT or DIALOGUE. SILENT: Pure action/visual, no spoken words DIALOGUE: Contains character speaking """ hook_match = re.search( r'#.*THE HOOK.*?\n(.*?)(?=#|\Z)', script, re.DOTALL | re.IGNORECASE ) if not hook_match: return "unknown", "No HOOK section found" hook_content = hook_match.group(1) lines = hook_content.split('\n') # Check for character names (ALL CAPS) which indicate dialogue for line in lines: stripped = line.strip() if stripped and stripped.isupper() and len(stripped) < 30: if not stripped.startswith('INT.') and not stripped.startswith('EXT.'): if not stripped.startswith('.') and not stripped.startswith('#'): # Skip common non-character words if stripped not in ['ECU', 'CU', 'MCU', 'MS', 'WS', 'POV', 'SFX', 'VFX', 'INSERT', 'CONTINUOUS', 'LATER', 'NIGHT', 'DAY', 'MORNING', 'EVENING']: return "dialogue", f"Speaker: {stripped}" return "silent", "No dialogue detected" def check_pattern_variety(project_path, batch_num, current_episodes): """ Check for 3+ consecutive same-type patterns. Looks at previous episodes + current batch to detect violations. Returns (passed, violations_list) """ violations = [] # Build history of cliffhanger types and hook types cliffhanger_history = [] hook_history = [] # Calculate episode range ep_start = (batch_num - 1) * GENERATION_BATCH_SIZE + 1 ep_end = batch_num * GENERATION_BATCH_SIZE # Load previous episodes for context (need at least MAX_CONSECUTIVE-1 previous) episodes_dir = project_path / "episodes" lookback = MAX_CONSECUTIVE for ep_num in range(max(1, ep_start - lookback), ep_end + 1): ep_file = episodes_dir / f"ep_{ep_num:03d}.md" if not ep_file.exists(): continue content = ep_file.read_text() script = extract_script_content(content) cliff_type, _ = detect_cliffhanger_type(script, content) hook_type, _ = detect_hook_type(script) cliffhanger_history.append((ep_num, cliff_type)) hook_history.append((ep_num, hook_type)) # Check for consecutive patterns in cliffhangers # PILOT WINDOW EXEMPTION: Episodes 1-10 allow all mid-action cliffhangers if len(cliffhanger_history) >= MAX_CONSECUTIVE: for i in range(len(cliffhanger_history) - MAX_CONSECUTIVE + 1): window = cliffhanger_history[i:i + MAX_CONSECUTIVE] types = [t for _, t in window] eps_in_window = [e for e, _ in window] # Check if all same type and includes current batch if len(set(types)) == 1 and types[0] != "unknown": # PILOT WINDOW: Skip mid-action violations if all episodes are <= PILOT_WINDOW_END if types[0] == "mid-action" and all(e <= PILOT_WINDOW_END for e in eps_in_window): continue # Exempt from pattern variety during pilot # Check if any episode in window is in current batch if any(ep_start <= e <= ep_end for e in eps_in_window): violations.append( f"Cliffhanger variety: {MAX_CONSECUTIVE}+ consecutive {types[0]} " f"(Episodes {eps_in_window[0]}-{eps_in_window[-1]})" ) break # One violation is enough # Check for consecutive patterns in hooks if len(hook_history) >= MAX_CONSECUTIVE: for i in range(len(hook_history) - MAX_CONSECUTIVE + 1): window = hook_history[i:i + MAX_CONSECUTIVE] types = [t for _, t in window] if len(set(types)) == 1 and types[0] != "unknown": eps_in_window = [e for e, _ in window] if any(ep_start <= e <= ep_end for e in eps_in_window): violations.append( f"Hook variety: {MAX_CONSECUTIVE}+ consecutive {types[0]} " f"(Episodes {eps_in_window[0]}-{eps_in_window[-1]})" ) break return len(violations) == 0, violations, cliffhanger_history, hook_history def check_emotional_beats(project_path, batch_num): """ Check that scheduled emotional beats are present within ±2 episode tolerance. Per CONSTANTS.md, each beat has a target episode and ±2 tolerance window. A beat is satisfied if ANY episode within the tolerance window contains emotional content matching the beat. Returns (passed, violations_list) """ violations = [] ep_start = (batch_num - 1) * GENERATION_BATCH_SIZE + 1 ep_end = batch_num * GENERATION_BATCH_SIZE episodes_dir = project_path / "episodes" # Emotional content indicators emotional_indicators = [ r'(vulnerability|vulnerable)', r'(emotion|emotional)', r'(tears|crying|wept)', r'(confession|confess)', r'(trust|trusted)', r'(betrayal|betrayed)', r'(sacrifice|sacrificed)', r'(forgive|forgiveness)', r'(love|loved)', r'(grief|grieving)', r'(reconcil|reconciliation)', r'(truth|honest)', r'(fear|afraid|scared)', r'(hope|hopeful|hopeless)', r'(broken|breaking)', r'(healing|healed)', r'(whispers|softly|gently)', r'(embrace|embraces|held)', r'(silent|silence|pause|beat)', r'(memory|remember)', ] for target_ep, beat_name, tolerance in EMOTIONAL_BEAT_SCHEDULE: # Check if this beat's tolerance window overlaps with the current batch beat_window_start = target_ep - tolerance beat_window_end = target_ep + tolerance # Skip beats whose window doesn't overlap this batch if beat_window_end < ep_start or beat_window_start > ep_end: continue # Only check this beat if the target episode falls within or near this batch # (we check when the batch contains the target or the last episode in the window) # To avoid duplicate checking across batches, only trigger when the target # episode itself falls within this batch's range if target_ep < ep_start or target_ep > ep_end: continue # Search the tolerance window for emotional content found_emotional = False search_start = max(1, beat_window_start) search_end = min(60, beat_window_end) for check_ep in range(search_start, search_end + 1): ep_file = episodes_dir / f"ep_{check_ep:03d}.md" if not ep_file.exists(): continue content = ep_file.read_text() script = extract_script_content(content) script_lower = script.lower() for pattern in emotional_indicators: if re.search(pattern, script_lower): found_emotional = True break if found_emotional: break if not found_emotional: # Check if the target episode file itself exists target_file = episodes_dir / f"ep_{target_ep:03d}.md" if not target_file.exists(): violations.append( f"Episode {target_ep}: Missing (required beat: {beat_name}, " f"window: ep {search_start}-{search_end})" ) else: violations.append( f"Episode {target_ep} (±{tolerance}): Missing emotional content " f"(required beat: {beat_name}, checked ep {search_start}-{search_end})" ) return len(violations) == 0, violations # --------------------------------------------------------------------------- # LLM-based Dangling Cause Check (Scaffolding Gate G2) # Replaces word-overlap continuity with forward-momentum verification. # Uses Haiku per episode boundary: "What unresolved threat/clock forces # action in the next episode?" # --------------------------------------------------------------------------- _DANGLING_CAUSE_PROMPT = """You are evaluating the transition between two consecutive microdrama episodes for forward momentum. RULE (Gulino's Dangling Cause): No episode ends at rest. Every episode boundary must contain a physical threat, emotional revelation, or ticking clock that demands immediate resolution. "And then" is failure. "Therefore" or "but" is success. Here is the CLIFFHANGER (final section) of the previous episode: {cliffhanger_text} Here is the HOOK (opening section) of the next episode: {hook_text} Answer these questions: 1. What specific unresolved threat, revelation, or ticking clock in the cliffhanger forces immediate action? 2. Does the hook pick up on this forward momentum? (YES or NO) Output EXACTLY in this format: CAUSE: [the specific dangling cause, or NONE if absent] PICKUP: [YES or NO] REASON: [one sentence]""" def check_dangling_cause(project_path, batch_num): """ Check that each episode boundary has forward momentum (Gulino). Uses Haiku to verify that the cliffhanger creates a dangling cause and the hook picks it up. Falls back to word-overlap if API unavailable. LLM calls run in parallel via ThreadPoolExecutor. Returns (passed, violations_list) """ from concurrent.futures import ThreadPoolExecutor violations = [] ep_start = (batch_num - 1) * GENERATION_BATCH_SIZE + 1 ep_end = batch_num * GENERATION_BATCH_SIZE episodes_dir = project_path / "episodes" client = get_anthropic_client() if callable(get_anthropic_client) else None # Collect all boundary pairs that need checking boundaries = [] for ep_num in range(ep_start, ep_end + 1): if ep_num == 1: continue prev_file = episodes_dir / f"ep_{ep_num - 1:03d}.md" curr_file = episodes_dir / f"ep_{ep_num:03d}.md" if not prev_file.exists() or not curr_file.exists(): continue prev_content = prev_file.read_text() curr_content = curr_file.read_text() prev_script = extract_script_content(prev_content) curr_script = extract_script_content(curr_content) cliff_match = re.search( r'#.*THE CLIFFHANGER.*?\n(.*?)(?=#|\Z)', prev_script, re.DOTALL | re.IGNORECASE ) hook_match = re.search( r'#.*THE HOOK.*?\n(.*?)(?=#|\Z)', curr_script, re.DOTALL | re.IGNORECASE ) if not cliff_match or not hook_match: continue prev_cliff = cliff_match.group(1).strip() curr_hook = hook_match.group(1).strip() boundaries.append((ep_num, prev_cliff, curr_hook)) if client is None: # Fallback: word-overlap heuristic — advisory only (not a hard fail) # The LLM-based check is the authoritative gate; without API key, # scene transitions that change location will false-positive here. warnings = [] for ep_num, prev_cliff, curr_hook in boundaries: # Strip sluglines from hook text (INT./EXT. lines aren't narrative content) hook_narrative = re.sub(r'^(INT\.|EXT\.).*$', '', curr_hook, flags=re.MULTILINE).strip() cliffhanger_words = set(re.findall(r'\b[a-z]{4,}\b', prev_cliff.lower())) hook_words = set(re.findall(r'\b[a-z]{4,}\b', hook_narrative.lower())) overlap = cliffhanger_words & hook_words if len(overlap) < 1: prev_last = [l for l in prev_cliff.split('\n') if l.strip()][-1][:60] if prev_cliff else "" curr_first = [l for l in curr_hook.split('\n') if l.strip()][0][:60] if curr_hook else "" warnings.append( f"Episode {ep_num}: Possible continuity break (word-overlap fallback)\n" f" Prev cliffhanger: \"{prev_last}...\"\n" f" Current hook: \"{curr_first}...\"" ) # Word-overlap fallback is advisory — print warnings but don't fail if warnings: print(" [WARN] Dangling Cause (word-overlap fallback, advisory only):") for w in warnings: print(f" - {w}") return True, [] # Pass — fallback is not authoritative # LLM-based dangling cause checks — run all in parallel def _check_boundary(args): ep_num, prev_cliff, curr_hook = args prompt = _DANGLING_CAUSE_PROMPT.format( cliffhanger_text=prev_cliff, hook_text=curr_hook, ) try: response = client.messages.create( model=ANTHROPIC_HAIKU, max_tokens=200, messages=[{"role": "user", "content": prompt}], ) result = response.content[0].text.strip() cause = parse_llm_field(result, "CAUSE") pickup = parse_llm_field(result, "PICKUP", ["YES", "NO"]) reason = parse_llm_field(result, "REASON") or "" if cause and "NONE" in cause.upper(): return f"Episode {ep_num - 1}: No dangling cause — episode ends at rest.\n {reason}" elif pickup == "NO": return f"Episode {ep_num}: Hook doesn't pick up previous momentum.\n {reason}" except Exception: pass return None with ThreadPoolExecutor(max_workers=len(boundaries) or 1) as executor: results = list(executor.map(_check_boundary, boundaries)) violations.extend(v for v in results if v is not None) return len(violations) == 0, violations def check_transition_patterns(project_path, batch_num): """ Check for HARD FAIL transition patterns: - Time skip after Mid-Action cliffhanger - Location change after Mid-Action without explanation Returns (passed, hard_fails, warnings) """ hard_fails = [] warnings = [] ep_start = (batch_num - 1) * GENERATION_BATCH_SIZE + 1 ep_end = batch_num * GENERATION_BATCH_SIZE episodes_dir = project_path / "episodes" # Time skip phrases that indicate non-immediate pickup time_skip_phrases = [ 'later', 'hours later', 'days later', 'the next morning', 'when she woke', 'when he woke', 'twenty-six days', 'twenty-seven days' ] for ep_num in range(ep_start, ep_end + 1): if ep_num == 1: continue prev_file = episodes_dir / f"ep_{ep_num - 1:03d}.md" curr_file = episodes_dir / f"ep_{ep_num:03d}.md" if not prev_file.exists() or not curr_file.exists(): continue prev_content = prev_file.read_text() curr_content = curr_file.read_text() # Get cliffhanger type from metadata (supports both new and legacy format) main_type, subtype_code, _ = extract_cliffhanger_metadata(prev_content) # For transition checks, we only care if it's mid-action or aftermath # Mid-action (M or M-*) requires immediate pickup if main_type == 'mid-action': prev_cliff_type = 'M' elif main_type == 'aftermath': prev_cliff_type = 'A' else: prev_cliff_type = '?' # Extract hook text hook_match = re.search( r'#.*THE HOOK.*?\n(.*?)(?=#|\Z)', curr_content, re.DOTALL | re.IGNORECASE ) curr_hook = hook_match.group(1).strip().lower() if hook_match else "" # Extract locations prev_cliff_match = re.search( r'#.*THE CLIFFHANGER.*?\n(.*?)(?=#|\Z)', prev_content, re.DOTALL | re.IGNORECASE ) prev_cliff_section = prev_cliff_match.group(1) if prev_cliff_match else "" prev_loc_match = re.search(r'(?:INT\.|EXT\.)\s*(.+?)(?:\s*-|$)', prev_cliff_section, re.MULTILINE) curr_loc_match = re.search(r'(?:INT\.|EXT\.)\s*(.+?)(?:\s*-|$)', curr_content[:500], re.MULTILINE) prev_location = prev_loc_match.group(1).strip().upper() if prev_loc_match else None curr_location = curr_loc_match.group(1).strip().upper() if curr_loc_match else None # Rule 1: Mid-Action must have immediate pickup (no time skip) if prev_cliff_type == 'M': hook_start = curr_hook[:150] for phrase in time_skip_phrases: if phrase in hook_start: hard_fails.append( f"Episode {ep_num}: Time skip ('{phrase}') after Mid-Action cliffhanger\n" f" Mid-Action cliffhangers require immediate pickup" ) break # Rule 2: Location should match for Mid-Action if prev_location and curr_location: if prev_location != curr_location and prev_location not in curr_location: hard_fails.append( f"Episode {ep_num}: Location change after Mid-Action\n" f" Cliffhanger: {prev_location}\n" f" Hook: {curr_location}\n" f" Must continue in same location or show transition" ) return len(hard_fails) == 0, hard_fails, warnings def check_must_contain(project_path, batch_num): """ Check that MUST CONTAIN items from episode_arc.md are present. Returns (passed, violations_list) """ violations = [] ep_start = (batch_num - 1) * GENERATION_BATCH_SIZE + 1 ep_end = batch_num * GENERATION_BATCH_SIZE # Load episode_arc.md arc_file = project_path / "bible" / "episode_arc.md" if not arc_file.exists(): return True, [] # No arc file, skip check arc_content = arc_file.read_text() episodes_dir = project_path / "episodes" for ep_num in range(ep_start, ep_end + 1): # Find MUST CONTAIN for this episode in arc # Pattern: looking for episode entry with MUST CONTAIN marker ep_pattern = rf'\|\s*{ep_num}\s*\|.*?MUST CONTAIN[:\s]*([^|]+)\|' match = re.search(ep_pattern, arc_content, re.IGNORECASE) if not match: # Try alternative format ep_pattern2 = rf'Episode\s*{ep_num}.*?MUST CONTAIN[:\s]*([^\n]+)' match = re.search(ep_pattern2, arc_content, re.IGNORECASE) if not match: continue # No MUST CONTAIN for this episode must_contain = match.group(1).strip() if not must_contain or must_contain == '-': continue # Check episode content ep_file = episodes_dir / f"ep_{ep_num:03d}.md" if not ep_file.exists(): violations.append(f"Episode {ep_num}: Missing (MUST CONTAIN: {must_contain})") continue content = ep_file.read_text().lower() # Check if MUST CONTAIN items appear in episode # Split by comma or semicolon for multiple items items = re.split(r'[,;]', must_contain) for item in items: item = item.strip().lower() if item and item not in content: # Try fuzzy match (key words) key_words = [w for w in item.split() if len(w) > 3] if key_words and not all(w in content for w in key_words): violations.append( f"Episode {ep_num}: MUST CONTAIN missing: \"{item.strip()}\"" ) return len(violations) == 0, violations def calculate_ratios(cliffhanger_history, hook_history): """Calculate running ratios for tracking.""" cliff_types = [t for _, t in cliffhanger_history if t != "unknown"] hook_types = [t for _, t in hook_history if t != "unknown"] midaction_count = sum(1 for t in cliff_types if t == "mid-action") aftermath_count = sum(1 for t in cliff_types if t == "aftermath") silent_count = sum(1 for t in hook_types if t == "silent") dialogue_count = sum(1 for t in hook_types if t == "dialogue") total_cliff = midaction_count + aftermath_count total_hook = silent_count + dialogue_count return { "cliffhanger": { "mid_action": midaction_count, "aftermath": aftermath_count, "mid_action_pct": round(midaction_count / total_cliff * 100, 1) if total_cliff > 0 else 0, "aftermath_pct": round(aftermath_count / total_cliff * 100, 1) if total_cliff > 0 else 0, }, "hook": { "silent": silent_count, "dialogue": dialogue_count, "silent_pct": round(silent_count / total_hook * 100, 1) if total_hook > 0 else 0, "dialogue_pct": round(dialogue_count / total_hook * 100, 1) if total_hook > 0 else 0, } } def main(): if len(sys.argv) < 3: print("Usage: python3 quality_gate.py ") print("Example: python3 quality_gate.py ./leviathan 1") sys.exit(1) project_path = Path(sys.argv[1]).resolve() batch_num = int(sys.argv[2]) if not project_path.exists(): print(f"Error: Project path does not exist: {project_path}") sys.exit(1) ep_start = (batch_num - 1) * GENERATION_BATCH_SIZE + 1 ep_end = batch_num * GENERATION_BATCH_SIZE print(f"\n{'='*60}") print(f"QUALITY GATE: Batch {batch_num} (Episodes {ep_start}-{ep_end})") print(f"Project: {project_path.name}") print(f"{'='*60}") all_passed = True tier1_violations = [] tier2_violations = [] # TIER 1: Pattern Variety print(f"\nTIER 1: PATTERN VARIETY") print("-" * 40) variety_passed, variety_violations, cliff_hist, hook_hist = check_pattern_variety( project_path, batch_num, [] ) if variety_passed: print(" [PASS] No pattern violations detected") else: print(" [FAIL] Pattern violations found:") for v in variety_violations: print(f" - {v}") tier1_violations.extend(variety_violations) all_passed = False # TIER 2: Dramatic Beats print(f"\nTIER 2: DRAMATIC BEATS") print("-" * 40) # Check emotional beats emotional_passed, emotional_violations = check_emotional_beats(project_path, batch_num) if emotional_passed: print(" [PASS] Emotional beats: All scheduled beats present") else: print(" [FAIL] Emotional beats:") for v in emotional_violations: print(f" - {v}") tier2_violations.extend(emotional_violations) all_passed = False # Check dangling cause (replaces word-overlap continuity) dangling_passed, dangling_violations = check_dangling_cause(project_path, batch_num) if dangling_passed: print(" [PASS] Dangling Cause: Forward momentum maintained at episode boundaries") else: print(" [FAIL] Dangling Cause violations:") for v in dangling_violations: print(f" - {v}") tier2_violations.extend(dangling_violations) all_passed = False # Check transition patterns (HARD FAIL) transition_passed, transition_hard_fails, transition_warnings = check_transition_patterns( project_path, batch_num ) if transition_passed: print(" [PASS] Transition patterns: All transitions valid") else: print(" [FAIL] Transition pattern violations:") for v in transition_hard_fails: print(f" - {v}") tier2_violations.extend(transition_hard_fails) all_passed = False # Check MUST CONTAIN must_passed, must_violations = check_must_contain(project_path, batch_num) if must_passed: print(" [PASS] MUST CONTAIN: All required elements present") else: print(" [FAIL] MUST CONTAIN:") for v in must_violations: print(f" - {v}") tier2_violations.extend(must_violations) all_passed = False # TIER 3: Tracking (always report, never blocks) print(f"\nTIER 3: TRACKING INFO") print("-" * 40) ratios = calculate_ratios(cliff_hist, hook_hist) print(f" Cliffhanger distribution (series so far):") print(f" Mid-action: {ratios['cliffhanger']['mid_action_pct']}% ({ratios['cliffhanger']['mid_action']} episodes)") print(f" Aftermath: {ratios['cliffhanger']['aftermath_pct']}% ({ratios['cliffhanger']['aftermath']} episodes)") print(f" Hook distribution (series so far):") print(f" Silent: {ratios['hook']['silent_pct']}% ({ratios['hook']['silent']} episodes)") print(f" Dialogue: {ratios['hook']['dialogue_pct']}% ({ratios['hook']['dialogue']} episodes)") # Final result print(f"\n{'='*60}") if all_passed: print(f"QUALITY GATE: PASSED") print(f"Batch {batch_num} meets dramatic quality standards") print(f"{'='*60}\n") sys.exit(0) else: print(f"QUALITY GATE: FAILED") print(f"\nREGENERATION REQUIRED:") if tier1_violations: print(f"\n TIER 1 (Pattern Variety):") for v in tier1_violations: print(f" - {v}") if tier2_violations: print(f"\n TIER 2 (Dramatic Beats):") for v in tier2_violations: print(f" - {v}") print(f"\nFix the issues above and re-run validation.") print(f"{'='*60}\n") sys.exit(1) if __name__ == "__main__": main()