#!/usr/bin/env python3 """ Script Doctor — Full-series diagnostic via Gemini API. Bundles the complete episode corpus + character bible + treatment, sends to Gemini for series-level analysis, and produces a structured revision brief with annotation-ready output for /revise. Usage: python3 script_doctor.py [project] --full # RECOMMENDED: automated full pipeline python3 script_doctor.py [project] --bundle # Preview corpus payload python3 script_doctor.py [project] --diagnose # Run broad diagnostic only python3 script_doctor.py [project] --verify # Post-revision verification python3 script_doctor.py [project] --focus voice,arc_earning # Focused analysis python3 script_doctor.py [project] --to-annotations # Extract annotations from brief python3 script_doctor.py [project] --deep-fix F002 # Creative fix for structural finding Requires: pip install google-generativeai export GEMINI_API_KEY="your-key-here" """ import argparse import json import os import re import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Optional # Add engine tools to path for imports (matches init_orchestrator_state.py) _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 cost_tracker import CostTracker # noqa: E402 (import after sys.path bootstrap) from recoil.core.model_profiles import get_model # noqa: E402 from recoil.core.paths import ProjectPaths, ProjectsRootUnresolvable # noqa: E402 # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- # Gemini model — 3.1 Pro for best reasoning on script work DEFAULT_MODEL = get_model("pro", "text") # Fallback for very large corpora LARGE_CONTEXT_MODEL = get_model("large_context", "text") # Approximate tokens per word (conservative estimate for English) TOKENS_PER_WORD = 1.3 # Context window sizes CONTEXT_LIMITS = { "gemini-3-pro-preview": 1_000_000, "gemini-3-pro": 1_000_000, "gemini-3.1-pro-preview": 1_000_000, "gemini-1.5-pro": 2_000_000, } # --------------------------------------------------------------------------- # Corpus Bundling # --------------------------------------------------------------------------- def _project_paths_or_exit(project: str) -> ProjectPaths: """Resolve project paths with script_doctor's existing fail-loud CLI style.""" try: return ProjectPaths.for_project(project) except (FileNotFoundError, ProjectsRootUnresolvable) as exc: print(f"ERROR: Project directory not found: {project} ({exc})", file=sys.stderr) sys.exit(1) def bundle_corpus(project: str) -> dict: """ Read all project files and bundle into a single payload dict. Returns: { "episodes": {1: "text...", 2: "text...", ...}, "characters": "text...", "treatment": "text...", "orchestration": "text...", "episode_count": 60, "word_count": 30000, "estimated_tokens": 39000, } """ pp = _project_paths_or_exit(project) # --- Episodes --- episodes_dir = pp.episodes_dir if not episodes_dir.is_dir(): print(f"ERROR: Episodes directory not found: {episodes_dir}", file=sys.stderr) sys.exit(1) episode_files = sorted(episodes_dir.glob("ep_*.md")) if not episode_files: print(f"ERROR: No episode files found in {episodes_dir}", file=sys.stderr) sys.exit(1) episodes = {} total_words = 0 for ep_file in episode_files: # Extract episode number from filename match = re.search(r"ep_(\d+)", ep_file.name) if match: ep_num = int(match.group(1)) text = ep_file.read_text(encoding="utf-8") episodes[ep_num] = text total_words += len(text.split()) # --- Characters --- chars_path = pp.bible_dir / "characters.md" characters = "" if chars_path.exists(): characters = chars_path.read_text(encoding="utf-8") total_words += len(characters.split()) # --- Treatment --- treatment_path = pp.treatment_path treatment = "" if treatment_path.exists(): treatment = treatment_path.read_text(encoding="utf-8") total_words += len(treatment.split()) # --- Orchestration --- orch_path = pp.project_root / "ORCHESTRATION.md" orchestration = "" if orch_path.exists(): orchestration = orch_path.read_text(encoding="utf-8") total_words += len(orchestration.split()) estimated_tokens = int(total_words * TOKENS_PER_WORD) return { "episodes": episodes, "characters": characters, "treatment": treatment, "orchestration": orchestration, "episode_count": len(episodes), "word_count": total_words, "estimated_tokens": estimated_tokens, } def format_corpus_payload(corpus: dict, focus: Optional[list] = None) -> str: """ Format the bundled corpus into a single text payload for Gemini, including the series-level diagnostic prompt with annotation-ready output. The prompt uses 9 series-level assessment dimensions as a reporting vocabulary (not a checklist), and requests annotations directly in /revise format — eliminating the brittle quote-matching conversion layer. """ sections = [] # ─── ROLE + CONTEXT ─────────────────────────────────────────────── sections.append( "# ROLE\n\n" "You are a veteran script doctor. You've been hired to review an entire " "microdrama series — all episodes, start to finish — and deliver an honest, " "specific diagnostic.\n\n" "This series was generated by an AI scripting engine in 5-episode batches " "with context reloads between batches. The generating model never saw more " "than ~10 episodes at once. Your advantage: you can see ALL of it simultaneously. " "The systemic issues — drift, repetition, unearned arcs, convergence — are " "invisible at the batch level but obvious at the series level. That's your job.\n" ) # ─── VITALITY FRAME ────────────────────────────────────────────── sections.append( "# THE VITALITY STANDARD\n\n" "Before you start cataloguing problems, understand the bar: these scripts " "need to feel ALIVE. Engaging. Fun. Surprising. Worth watching.\n\n" "If the series feels like a formula — if it grinds instead of breathes, " "if there's no humor or play or unpredictability, if the emotional beats " "land like checkboxes rather than punches — that's a problem as serious as " "any plot hole. Flag it.\n\n" "A technically correct series that's boring is a failure. A series with " "minor continuity issues that makes you want to watch the next episode is " "a success with fixable problems. Keep this lens active throughout.\n" ) # ─── OPEN DIAGNOSTIC QUESTION ──────────────────────────────────── sections.append( "# YOUR TASK\n\n" "Read this entire series and tell me everything that's wrong with it. " "Be honest and specific. Don't soften findings to be polite — the goal " "is to make this series better.\n\n" "When you report findings, classify them using the assessment dimensions " "below. But if you find something that doesn't fit any category, report " "it under 'uncategorized' with your own severity assessment. The dimensions " "are a vocabulary for filing what you find, not a checklist constraining " "what you look for.\n" ) # ─── FOCUS INSTRUCTION (if applicable) ──────────────────────────── if focus: focus_str = ", ".join(focus) sections.append( f"**FOCUS AREAS:** Limit your analysis to: {focus_str}. " "Skip other dimensions.\n" ) # ─── ASSESSMENT DIMENSIONS ──────────────────────────────────────── sections.append( "# ASSESSMENT DIMENSIONS\n\n" "Use these 9 dimensions to classify your findings. Each includes detection " "guidance — what to look for, not a checklist to complete.\n\n" "## 1. voice\n" "Character speech pattern integrity across the full series.\n" "- **Drift:** Characters gradually losing their distinctive voice over 60 episodes\n" "- **Convergence:** Multiple characters starting to sound alike (especially in Act 3)\n" "- **Calcification:** Behavioral DNA rules becoming caricature through rigid adherence " "(e.g., a verbal tic that should break in emotional moments but never does)\n" "- **\"Never say\" violations:** Characters using words/phrases their bible explicitly forbids\n" "- **Missed breaks:** Moments where BREAKING the voice pattern would serve the story " "(pattern-break IS the character moment) but the script plays it safe\n\n" "## 2. pattern_fatigue\n" "Repetition that grates at series scale.\n" "- **Catchphrase overuse:** Exact count of signature phrases — which are earned anchors, " "which are filler?\n" "- **Physical tic repetition:** Same physical behavior described the same way too often\n" "- **Idiom loops:** Characters cycling through the same 3-4 idioms\n" "- **Action description patterns:** Repeated imagery (e.g., every falling sequence, " "every targeting reticle, every 'eyes narrow')\n" "- **Structural rhythm monotony:** Every episode hitting the same emotional shape, " "no valley episodes, no surprise structures\n\n" "## 3. arc_earning\n" "Character transformations AND relationships: are the emotional moves earned?\n" "- **Skipped beats:** Transformations that jump steps (disillusionment → altruism " "with no transactional bridge)\n" "- **Unearned relationship shifts:** Characters suddenly trusting/caring without " "sufficient shared experience\n" "- **Missing milestones:** Key emotional beats (vulnerability, betrayal, reconciliation) " "that the arc needs but never delivers\n" "- **Plot convenience vs character logic:** Does the arc use the character's own " "Behavioral DNA logic, or does it rely on plot forcing a change?\n" "- **Relationship earning schedule:** Map actual relationship progress against the " "earning curve — too fast? Too slow? Skipping levels?\n\n" "## 4. continuity\n" "Thread integrity, knowledge consistency, world logic.\n" "- **Planted-but-unresolved:** Threads seeded early that never pay off\n" "- **Resolved-too-neatly:** Threads that wrap up in ways that feel contrived\n" "- **Knowledge consistency:** Characters knowing things they shouldn't, or " "forgetting things they should know\n" "- **World logic:** Rules established early that are violated later\n" "- **Batch-boundary artifacts:** Continuity breaks that align with the 5-episode " "batch boundaries (strong signal of context-reload drift)\n\n" "## 5. texture_tone_vitality\n" "Register variety, energy distribution, does this series feel alive?\n" "- **Kill Box fatigue:** Does the rigid 90-second structure become oppressive? " "Are there moments where the format should breathe but can't?\n" "- **Breathing room:** Are there enough quiet/valley episodes between peaks? " "Or is it relentless intensity that numbs the audience?\n" "- **Surprise distribution:** Do any episodes genuinely surprise? Or is every " "twist telegraphed by the formula?\n" "- **Humor and play:** Is there any? Should there be more? Characters joking, " "banter, moments of lightness that make the heavy beats hit harder?\n" "- **Tone variety:** Does the series hit different emotional registers (dread, " "wonder, tenderness, fury) or flatten to one sustained note?\n\n" "## 6. exposition_load\n" "Dialogue doing work that images should do.\n" "- **Dialogue-as-narration:** Characters explaining what's happening instead of " "the action showing it\n" "- **'As you know' patterns:** Characters telling each other things they already know " "for the audience's benefit\n" "- **Missed visual storytelling:** Moments where a physical action, prop, or " "environmental detail could replace dialogue exposition\n" "- **Emotion-telling:** 'She felt scared' instead of showing fear through behavior\n\n" "## 7. predictable_reversal\n" "Value turn predictability — the pendulum problem (McKee).\n" "- **Pendulum pattern:** Reversals that alternate mechanically " "(good→bad→good→bad) without escalation or surprise\n" "- **Telegraphed turns:** Reversals the audience can predict 20 seconds " "in advance because the pattern is established\n" "- **Same-polarity repetition:** Multiple consecutive episodes ending on " "the same emotional valence (all bad, all good)\n" "- **Missing ironic reversals:** Turns where the character gets what they " "WANTED but not what they NEEDED, or vice versa\n" "- **Value turn variety:** Are reversals hitting different value dimensions " "(trust/betrayal, power/weakness, knowledge/ignorance) or always the same one?\n\n" "## 8. on_the_nose\n" "Characters directly stating themes, emotions, or subtext (Seger).\n" "- **Emotion statements:** Characters saying 'I'm angry' instead of showing anger " "through behavior\n" "- **Theme announcements:** Characters articulating the theme of the episode or " "series directly ('This is really about trust')\n" "- **Subtext made text:** Dialogue that says what should be communicated through " "action, silence, or implication\n" "- **Motivation narration:** Characters explaining WHY they're doing something " "instead of just doing it\n" "- **Relationship labeling:** 'You're my friend now' instead of earned behavioral " "shift. 'I don't trust you' instead of shown distrust.\n\n" "## 9. integration\n" "A-story / B-story convergence and thematic unity (Aronson).\n" "- **Parallel without convergence:** A-story and B-story run independently and " "never meaningfully intersect\n" "- **Missing thematic echo:** B-story doesn't illuminate or complicate the A-story " "theme from a different angle\n" "- **Finale integration failure:** The resolution doesn't use lessons/tools/relationships " "from the B-story to resolve the A-story\n" "- **Late convergence only:** Stories only connect in the last few episodes rather than " "building toward convergence throughout\n" "- **One-way dependency:** A-story affects B-story but not vice versa (or the reverse)\n\n" "## uncategorized\n" "Anything that doesn't fit the above. Use your judgment for severity.\n\n" ) # ─── CATEGORY → ACTION TYPE MAPPING ─────────────────────────────── sections.append( "# FINDING → ANNOTATION TYPE MAPPING\n\n" "Each finding must include specific annotations — exact locations in the episodes " "where the problem manifests, with the action type that should fix it.\n\n" "| Problem Type | Annotation Action | When to Use |\n" "|---|---|---|\n" "| Text substitution (catchphrase swap, voice fix, exposition rewrite) | REWRITE | " "The fix is replacing specific text with better text |\n" "| Remove text entirely (redundant exposition, filler) | DELETE | " "The fix is cutting, not replacing |\n" "| Structural issue (unearned arc, pacing, missing beat) | FLAG | " "The fix requires creative rewriting beyond simple substitution |\n\n" "**REWRITE and DELETE annotations must include exact `selected_text` quotes from the episodes.** " "You have the full episode text — quote it exactly as it appears.\n\n" "**FLAG annotations** mark locations where structural work is needed. " "The `selected_text` should identify the relevant passage, and the `note` should " "explain what's wrong and what kind of fix is needed.\n" ) # ─── OUTPUT SCHEMA ──────────────────────────────────────────────── sections.append( "# OUTPUT SCHEMA\n\n" "Return ONLY valid JSON (no markdown fences, no commentary outside the JSON).\n\n" "```\n" "{\n" ' "version": "2.0",\n' ' "summary": "2-3 sentence overall assessment of series quality and vitality",\n' ' "findings": [\n' " {\n" ' "id": "F001",\n' ' "dimension": "voice|pattern_fatigue|arc_earning|continuity|texture_tone_vitality|exposition_load|predictable_reversal|on_the_nose|integration|uncategorized",\n' ' "severity": "P1|P2|P3",\n' ' "title": "Short descriptive title",\n' ' "description": "Detailed explanation of the issue across the series. Include exact counts for repetition. Include specific episode ranges for drift.",\n' ' "episodes": [1, 4, 7, 14],\n' ' "annotations": [\n' " {\n" ' "episode": 4,\n' ' "line": 0,\n' ' "action": "REWRITE|DELETE|FLAG",\n' ' "selected_text": "Exact quote from the episode text",\n' ' "note": "[F001/P1] What is wrong. REPLACEMENT: What it should say instead. (For REWRITE, always include replacement text.)"\n' " }\n" " ]\n" " }\n" " ],\n" ' "character_grades": {\n' ' "CHARACTER_NAME": {\n' ' "grade": "A+|A|A-|B+|B|B-|C+|C|C-|D",\n' ' "arc_summary": "Brief arc assessment — earned beats, skipped beats, overall trajectory",\n' ' "voice_notes": "Brief voice assessment — distinctiveness, drift, calcification",\n' ' "voice_diversification": "Specific suggestions for expanding this character\'s voice palette. Name 2-3 new speech modes or registers they should use in specific emotional contexts."\n' " }\n" " },\n" ' "workflow_observations": [\n' " {\n" ' "target": "characters.md|treatment.md|generation_instructions|format_rules",\n' ' "observation": "What should change in the upstream document to prevent this issue in future generations",\n' ' "rationale": "Why this is a system-level fix, not just an episode-level fix"\n' " }\n" " ],\n" ' "stats": {\n' ' "total_findings": 12,\n' ' "p1_count": 3,\n' ' "p2_count": 5,\n' ' "p3_count": 4,\n' ' "total_annotations": 45\n' " }\n" "}\n" "```\n\n" ) # ─── ANNOTATION FORMAT RULES ────────────────────────────────────── sections.append( "# ANNOTATION FORMAT RULES\n\n" "Each annotation in a finding's `annotations` array must follow these rules:\n\n" "- **`episode`**: Integer episode number\n" "- **`line`**: Set to 0 (line matching happens downstream when /revise searches " "for selected_text)\n" "- **`action`**: One of REWRITE, DELETE, or FLAG\n" "- **`selected_text`**: EXACT text from the episode. Copy it character-for-character. " "This is the text that will be found and replaced/deleted/flagged. For multi-line " "passages, include the full passage with newlines.\n" "- **`note`**: Format as `[FINDING_ID/SEVERITY] Description. Fix guidance.`\n" " - For REWRITE: MUST include replacement text. Format: `[F001/P2] Issue description. " "REPLACEMENT: \"The actual line(s) that should replace the selected text.\"` " "Do not just describe what should change — write the replacement.\n" " - For DELETE: explain why this should be cut and what (if anything) should fill the gap\n" " - For FLAG: explain the structural issue, what kind of creative fix is needed, and " "sketch what the rewritten scene should accomplish\n\n" ) # ─── SEVERITY DEFINITIONS ───────────────────────────────────────── sections.append( "# SEVERITY DEFINITIONS\n\n" "- **P1 — Must fix.** Actively damages the series. Unearned arcs that break " "audience trust. Repetition so heavy it becomes a meme. Voice collapse that " "makes characters interchangeable. Structural gaps that confuse the story.\n" "- **P2 — Should fix.** Noticeable quality issue that a careful viewer catches. " "Moderate repetition, pacing drag, exposition that slows momentum, voice drift " "that's distracting but not breaking.\n" "- **P3 — Consider.** Polish-level. Minor variety improvements, subtle exposition " "reduction, small voice refinements. Good to fix if the budget allows.\n\n" ) # ─── EVIDENCE REQUIREMENTS ──────────────────────────────────────── sections.append( "# EVIDENCE REQUIREMENTS\n\n" "Every finding MUST include:\n" "- Specific episode numbers where the issue manifests\n" "- Exact quotes from the episodes (not paraphrases)\n" "- For pattern_fatigue: exact counts (e.g., 'phrase X appears 28 times across " "episodes 1, 4, 7, 14, ...')\n" "- For arc_earning: the beat sequence showing where the gap is\n" "- For voice: side-by-side examples showing drift/convergence\n\n" "Findings without specific evidence will be ignored.\n\n" "## DEPTH EXPECTATIONS\n\n" "You are reviewing a 60-episode series (~27,000 words). A thorough diagnostic should " "produce:\n" "- **12-25 findings** across the 9 dimensions (plus uncategorized)\n" "- **40-80 annotations** total — specific line-level fixes across the series\n" "- **Every REWRITE annotation** must include replacement text in the note " "(not just 'fix this' — write what it should say instead)\n" "- **Catchphrase/tic findings** must include a KEEP/CUT list: which instances are " "earned anchors (KEEP) and which are overuse (CUT with episode numbers)\n" "- **Voice findings** must include alternative phrasing — don't just say 'Varek sounds " "like a calculator', write what he SHOULD say in that moment\n\n" "A diagnostic with fewer than 10 findings or 30 annotations is too shallow. " "You have the full series — use it.\n\n" ) # ═══ CORPUS (4 TIERS) ════════════════════════════════════════════ # ─── TIER 1: Engine Constraints Briefing ────────────────────────── sections.append( f"{'='*60}\n" "# TIER 1 — ENGINE CONSTRAINTS BRIEFING\n" f"{'='*60}\n\n" "Understanding the production system is critical to diagnosing correctly. " "Some 'problems' are features of the format. Others are artifacts of the " "batch generation model. This section gives you the full framework so you " "can distinguish features from bugs.\n\n" "## Kill Box Structure (90 seconds per episode)\n" "Every episode follows a rigid 5-beat structure within ~450-500 words:\n" "- HOOK [00:00-00:05]: 5 sec — immediate grab\n" "- SETUP [00:05-00:15]: 10 sec — establish stakes\n" "- ESCALATION [00:15-00:40]: 25 sec — build pressure\n" "- TURN [00:40-00:70]: 30 sec — reversal/revelation\n" "- CLIFFHANGER [00:70-00:90]: 20 sec — unresolved ending\n\n" "## Numeric Constraints\n" "- 450-500 words per episode (total file words including headers)\n" "- Dialogue: ≤40% of word count, max 8 exchanges\n" "- Hooks: 70-85% silent / 15-30% dialogue\n" "- Cliffhangers: 70-85% mid-action / 15-30% aftermath\n" "- Max 3 consecutive episodes with same hook/cliffhanger type\n\n" "## Batch Generation Model\n" "Episodes were generated in batches of 5 with full context reloads between " "batches. The generating model never saw more than ~10 episodes of context. " "This is WHY systemic issues exist:\n" "- **Batch-boundary drift:** Voice, tone, and energy can shift at ep 5/10/15/20... boundaries\n" "- **No cross-batch visibility:** A catchphrase used 3x in batch 1 seems fine — but if " "every batch uses it 3x, that's 36 times across the series\n" "- **Thread fragility:** Threads planted in batch 2 may be forgotten by batch 6 " "when the model reloads context from a summary\n\n" "## Hook & Cliffhanger Taxonomy\n" "The engine defines specific hook and cliffhanger types. When evaluating variety, " "use this vocabulary:\n\n" "**Hook types (episode openings):**\n" "- SILENT hooks (70-85% of episodes): Visceral Image (pure visual), Found Object " "(character discovers something), Environmental Shift (world changes), Aftermath " "(consequences of last episode), Physical Action (mid-motion), Sensory Detail " "(non-visual sense), Countdown/Timer (ticking clock)\n" "- DIALOGUE hooks (15-30%): Mid-Conversation (join argument in progress), " "Question Hook (unanswered question), Declaration (character states intent), " "Contradiction (one line subverts another)\n\n" "**Cliffhanger types (episode endings):**\n" "- MID-ACTION (70-85%): Physical Peril, Choice Point, Discovery, Confrontation, " "Countdown Expiry\n" "- AFTERMATH (15-30%): Quiet Devastation, Pyrrhic Victory, Revelation Sinks In, " "Changed Dynamic, New Normal, Ominous Silence\n\n" "**Pattern rule:** No more than 3 consecutive episodes with the same hook type or " "cliffhanger type. If every episode ends on Physical Peril, that's a bug.\n\n" "## Variety & Surprise Toolkit\n" "The engine instructs writers to include variety through these mechanisms. When you " "find monotony, reference which toolkit element is missing:\n\n" "- **Wildcard beats:** 1 per episode — unexpected moments that break the formula " "(character does something out of pattern, world reveals a surprise, tone shifts)\n" "- **Tone variety:** Episodes should vary across dread, wonder, tenderness, fury, " "dark humor. If the series sits on one note, that's a structural problem.\n" "- **Structure variation:** Not every episode needs to follow the same emotional shape. " "Some should be valleys (quiet, character-driven, reflective) between peaks.\n" "- **Visual palette rotation:** Action descriptions should cycle through different " "sensory anchors — not always 'eyes glow' and 'sparks fly.' Use sound, texture, " "temperature, smell, scale shifts.\n" "- **Character surprise moments:** Characters should occasionally contradict their own " "established patterns in earned ways — the tough character showing tenderness, " "the analytical character making an emotional choice.\n\n" "## Emotional Architecture\n" "The engine uses a structured emotional system. When evaluating whether arcs are " "earned or emotional beats land, reference this framework:\n\n" "- **Emotional anchors:** Specific objects, gestures, or phrases that accumulate " "meaning through repetition. A scar, a phrase, a physical gesture that means more " "each time it appears. These are INTENTIONAL — but only if they evolve.\n" "- **Dual engine:** Every episode should engage both PLOT (external stakes) and " "EMOTION (internal stakes). If an episode is pure plot mechanics with no emotional " "movement, flag it.\n" "- **Relationship earning stages:** Relationships must progress through 5 levels:\n" " 1. TRANSACTIONAL (pure self-interest)\n" " 2. RELUCTANT ALLIANCE (shared enemy)\n" " 3. GRUDGING RESPECT (competence recognized)\n" " 4. UNSPOKEN BOND (actions speak)\n" " 5. DECLARED PARTNERSHIP (explicit commitment)\n" " Skipping levels is a P1 arc_earning issue. Rushing through levels is P2.\n" "- **Catharsis economy:** Emotional payoffs must be rationed. If characters have " "heart-to-hearts every 3 episodes, they stop being earned. The scarcity of " "vulnerability is what makes it powerful.\n" "- **The Ache:** Quiet moments where the audience FEELS the weight of unspoken emotion. " "These should exist in the series — if every emotional beat is screamed, none land.\n\n" "## Behavioral DNA System\n" "Characters are built from a structured template. Each character has:\n" "- **On-screen behaviors (3+):** Filmable actions the audience SEES (not backstory)\n" "- **Stress behavior:** What they do under pressure (must be surprising, not generic)\n" "- **Signature line:** A recurring phrase that passes the swap test " "(if another character could say it, it fails)\n" "- **Orthogonal trait:** An interest/behavior unrelated to the plot (makes them human)\n" "- **Contradiction:** Two opposing traits that create internal tension\n" "- **Voice DNA:** Speech patterns, verbal tics, sentence structure, emotional tells\n" "- **Need layers:** Surface want vs deep need (character may not know their own deep need)\n" "- **Anti-patterns:** Things the character would NEVER say or do\n\n" "Characters are INSTRUCTED to express specific tics, idioms, and signature lines. " "Repetition of character elements is partly a directive, not random — but the " "batch model can't see cumulative frequency. A tic that appears 3x per batch is " "intentional; 3x per batch × 12 batches = 36 total is probably excessive. " "Your job is to distinguish intentional character expression from cumulative " "overuse.\n\n" "**Critical behavioral checks:**\n" "- Are on-screen behaviors actually shown, or just told through dialogue?\n" "- Does the stress behavior appear when it should? Is it always the same, or " "does it evolve?\n" "- Is the signature line earned through scarcity, or cheapened through overuse?\n" "- Does the orthogonal trait appear at all? It should surface 2-3 times across " "60 episodes.\n" "- Does the contradiction create visible tension, or is one side always dominant?\n" "- When a character BREAKS their voice pattern, is it at an emotionally justified " "moment? Pattern-break IS the character moment.\n" ) # ─── TIER 2: Intended Story ─────────────────────────────────────── sections.append( f"\n{'='*60}\n" "# TIER 2 — INTENDED STORY\n" f"{'='*60}\n" ) if corpus["characters"]: sections.append( "\n## Character Bible\n\n" "Character voices, Behavioral DNA, arcs, relationships:\n\n" f"{corpus['characters']}\n" ) if corpus["treatment"]: treatment_words = len(corpus["treatment"].split()) if treatment_words > 15000: sections.append( "\n## Treatment (truncated to first 15,000 words)\n\n" f"{' '.join(corpus['treatment'].split()[:15000])}\n" "\n[Treatment truncated for token budget.]\n" ) else: sections.append( "\n## Treatment\n\n" f"{corpus['treatment']}\n" ) if corpus["orchestration"]: sections.append( "\n## Orchestration Rules\n\n" "Project-specific overrides and constraints:\n\n" f"{corpus['orchestration']}\n" ) # ─── TIER 3: Audience Context ───────────────────────────────────── sections.append( f"\n{'='*60}\n" "# TIER 3 — AUDIENCE CONTEXT\n" f"{'='*60}\n\n" "**Target demographic:** Men 18-35\n" "**Genre expectations:** Competence porn, high-stakes economics, " "underdog outsmarts the system, action over romance\n" "**Format:** Vertical video, 90 seconds per episode, 60 episodes\n\n" "**What's a feature vs a bug for this audience:**\n" "- Fast pacing is a FEATURE — but relentless identical pacing is a bug\n" "- Cliffhangers are a FEATURE — but predictable cliffhanger patterns are a bug\n" "- Competent protagonists are a FEATURE — but flawless protagonists are a bug\n" "- Signature catchphrases are a FEATURE — but catchphrase overuse is a bug\n" "- Action density is a FEATURE — but no breathing room is a bug\n" ) # ─── TIER 4: Episodes ───────────────────────────────────────────── sections.append( f"\n{'='*60}\n" "# TIER 4 — COMPLETE EPISODE SCRIPTS\n" f"{'='*60}\n" ) for ep_num in sorted(corpus["episodes"].keys()): sections.append( f"\n{'='*60}\n" f"EPISODE {ep_num}\n" f"{'='*60}\n\n" f"{corpus['episodes'][ep_num]}\n" ) return "\n".join(sections) def format_transition_payload(corpus: dict) -> str: """ Format the transition analysis prompt (Pass 1: Structural). Checks TWO levels of transition: 1. INTER-EPISODE: Every boundary (cliffhanger N → hook N+1) 2. INTRA-EPISODE: Every Kill Box section boundary within each episode (HOOK→SETUP→ESCALATION→TURN→CLIFFHANGER) Both levels use the THEREFORE/BUT vs AND THEN framework. Returns findings in the same JSON schema as the main diagnostic so merge_briefs() and extract_annotations() work without modification. Inter-episode findings use T-prefixed IDs (T001, T002...). Intra-episode findings use I-prefixed IDs (I001, I002...). """ sections = [] # ─── ROLE ────────────────────────────────────────────────────────── sections.append( "# ROLE\n\n" "You are a script continuity editor analyzing a 60-episode vertical " "microdrama series. Each episode is ~90 seconds. Viewers watch " "consecutively.\n\n" "You analyze transitions at TWO levels:\n" "1. **INTER-EPISODE** — Every boundary between episodes " "(cliffhanger N → hook N+1). These must feel inevitable.\n" "2. **INTRA-EPISODE** — Every boundary between Kill Box sections " "within each episode (HOOK→SETUP→ESCALATION→TURN→CLIFFHANGER). " "These must feel driven, not convenient.\n\n" "This series was generated in 5-episode batches. The generating model " "never saw more than ~10 episodes at once. Batch boundaries (ep 5/6, " "10/11, 15/16, etc.) are high-risk for transition problems.\n" ) # ─── THE RULE ────────────────────────────────────────────────────── sections.append( "# THE RULE: THEREFORE / BUT — NEVER \"AND THEN\"\n\n" "This applies to BOTH inter-episode and intra-episode transitions.\n\n" "**THEREFORE** — The previous beat causes or necessitates the next.\n" "Example: \"Kian's battery dies\" → THEREFORE → \"Jinx drags his dead " "chassis to safety\"\n\n" "**BUT** — The next beat complicates, subverts, or contradicts expectations.\n" "Example: \"They escape the Collectors\" → BUT → \"The person Jinx brings " "Kian to is terrified of him\"\n\n" "**\"AND THEN\" is a failure.** The next beat merely follows chronologically " "with no causal link.\n" "Example: \"Emotional confrontation in compartment\" → AND THEN → " "\"They're at a workshop\" (how? why?)\n\n" "This is the South Park rule: Trey Parker and Matt Stone's writing " "principle. Scenes connect with \"therefore\" or \"but,\" never " "\"and then.\" AND THEN creates lazy storytelling. THEREFORE and BUT " "create drama, surprise, suspense, causality, and resolution.\n" ) # ─── INTER-EPISODE CHECKS ───────────────────────────────────────── sections.append( "# PART 1: INTER-EPISODE TRANSITIONS\n\n" "For every transition from Episode N to Episode N+1, evaluate:\n\n" "1. **Causal Logic** — Is the connection THEREFORE, BUT, or AND THEN?\n" " If AND THEN: this is a finding. The cliffhanger needs a bridge line, " "or the hook needs to acknowledge how we got here.\n\n" "2. **Spatial Continuity** — Does the location change make sense?\n" " If Episode N ends at Location A and N+1 opens at Location B:\n" " - Is there any indication of travel/transition?\n" " - Could the audience infer how they got there?\n" " - Or does it feel like a teleport?\n\n" "3. **Character Positioning** — When a new character appears, do we " "know why?\n" " - How did they find the protagonists?\n" " - Were they established as being nearby?\n" " - Or do they just materialize?\n\n" "4. **Emotional Continuity** — Does the emotional state carry across?\n" " - If Episode N ends on a heavy emotional beat, does N+1 acknowledge it?\n" " - Or does the emotional thread get dropped?\n\n" "5. **Off-Screen Resolution** — Are cliffhanger threats resolved on screen?\n" " - If Episode N ends with a blade to someone's throat, Episode N+1 " "MUST show how they escape — not skip to \"they already escaped.\"\n" " - Skipping cliffhanger resolution makes the previous threat feel " "meaningless and breaks audience trust.\n\n" "Inter-episode findings use **T-prefixed IDs** (T001, T002, ...).\n" ) # ─── INTRA-EPISODE CHECKS ───────────────────────────────────────── sections.append( "# PART 2: INTRA-EPISODE TRANSITIONS (Kill Box Sections)\n\n" "Each episode follows a 5-section Kill Box structure:\n" "1. **THE HOOK** [00:00-00:05] — Immediate tension or mystery\n" "2. **THE SETUP** [00:05-00:15] — Establish stakes, character in conflict\n" "3. **THE ESCALATION** [00:15-00:40] — Pressure increases, new obstacle\n" "4. **THE TURN** [00:40-00:70] — Unexpected shift, stakes raised\n" "5. **THE CLIFFHANGER** [00:70-00:90] — Cut mid-action or aftermath\n\n" "For every boundary between adjacent Kill Box sections within each episode, " "evaluate whether the transition is THEREFORE, BUT, or AND THEN.\n\n" "**What to check:**\n" "- **Causal Logic** — Does the next section arise from the previous one? " "Or does it just happen next?\n" "- **Escalation Quality** — Does pressure genuinely increase? Flat " "transitions (same stakes, new activity) are AND THEN.\n" "- **Spatial Logic** — If the location changes within the episode, is it " "motivated by what just happened?\n" "- **Convenience** — Does an obstacle, character, or location appear " "conveniently to serve the plot? A drainage tunnel that shows up just " "because the writer needs water, a character who arrives just in time " "with no setup — these feel like cheats even if the audience can't " "articulate why.\n\n" "**Be strict about convenience.** Little moments of plot convenience — " "an unlocked door, a perfectly timed arrival, an obstacle that appears " "without setup — make the audience feel cheated even when they can't " "name what's wrong. Flag them.\n\n" "Intra-episode findings use **I-prefixed IDs** (I001, I002, ...).\n" ) # ─── ANNOTATION GUIDANCE ─────────────────────────────────────────── sections.append( "# ANNOTATION TYPES FOR TRANSITIONS\n\n" "Each finding should include specific annotations:\n\n" "- **REWRITE** — For text that needs a causal bridge added or " "spatial/causal grounding. Include the REPLACEMENT text.\n" " - Inter-episode example: Add \"I know someone who can tell us " "what you are\" to a cliffhanger to bridge to the next scene.\n" " - Intra-episode example: Add a line showing WHY the character " "goes to the drainage tunnel, not just that they do.\n" "- **FLAG** — For structural issues requiring creative rewriting " "(off-screen resolutions, episode reordering, missing beats).\n" " - Example: \"Standoff resolution skipped. Need to show escape.\"\n\n" "**Quote selected_text EXACTLY from the episode.** For inter-episode " "cliffhanger fixes, quote the last action/dialogue line. For hook " "fixes, quote the first action/dialogue line of the next episode. " "For intra-episode fixes, quote the first line of the section that " "has the weak transition.\n" ) # ─── OUTPUT SCHEMA ───────────────────────────────────────────────── sections.append( "# OUTPUT SCHEMA\n\n" "Return ONLY valid JSON (no markdown fences). Both inter-episode " "and intra-episode findings go in the same findings array:\n\n" "```\n" "{\n" ' "version": "2.0",\n' ' "summary": "Overall assessment of structural causality",\n' ' "findings": [\n' " {\n" ' "id": "T001",\n' ' "dimension": "transitions",\n' ' "severity": "P1|P2|P3",\n' ' "title": "EP N → EP N+1: Short description",\n' ' "description": "What the connector is, what breaks, why it matters.",\n' ' "episodes": [N, N+1],\n' ' "connector": "AND THEN|THEREFORE|BUT",\n' ' "annotations": [\n' " {\n" ' "episode": N,\n' ' "line": 0,\n' ' "action": "REWRITE|FLAG",\n' ' "selected_text": "Exact quote from the episode",\n' ' "note": "[T001/P1] Issue. REPLACEMENT: \\"Fixed text.\\""\n' " }\n" " ]\n" " },\n" " {\n" ' "id": "I001",\n' ' "dimension": "transitions_internal",\n' ' "severity": "P1|P2|P3",\n' ' "title": "EP N: SETUP → ESCALATION — Short description",\n' ' "description": "What breaks the causal chain between Kill Box sections.",\n' ' "episodes": [N],\n' ' "connector": "AND THEN",\n' ' "kill_box_transition": "SETUP → ESCALATION",\n' ' "annotations": [\n' " {\n" ' "episode": N,\n' ' "line": 0,\n' ' "action": "REWRITE",\n' ' "selected_text": "First line of the weak section",\n' ' "note": "[I001/P2] Convenience. REPLACEMENT: \\"Motivated text.\\""\n' " }\n" " ]\n" " }\n" " ],\n" ' "transition_stats": {\n' ' "inter_episode": {\n' ' "total": 59,\n' ' "therefore_count": 0,\n' ' "but_count": 0,\n' ' "and_then_count": 0\n' " },\n" ' "intra_episode": {\n' ' "total_checked": 240,\n' ' "therefore_count": 0,\n' ' "but_count": 0,\n' ' "and_then_count": 0\n' " }\n" " },\n" ' "stats": {\n' ' "total_findings": 0,\n' ' "p1_count": 0,\n' ' "p2_count": 0,\n' ' "p3_count": 0,\n' ' "total_annotations": 0\n' " },\n" ' "workflow_observations": [\n' " {\n" ' "target": "generation_instructions|treatment.md",\n' ' "observation": "What should change upstream to prevent issues",\n' ' "rationale": "Why this is a systemic fix"\n' " }\n" " ]\n" "}\n" "```\n\n" ) # ─── SEVERITY ────────────────────────────────────────────────────── sections.append( "# SEVERITY GUIDE\n\n" "**Inter-episode:**\n" "- **P1**: Audience will be confused or disoriented. Teleportation, " "character materialization, off-screen cliffhanger resolution, emotional " "whiplash from dropped threads.\n" "- **P2**: Noticeable gap but inferrable. Location change without bridge, " "dropped emotional thread where the audience can fill in the gap.\n" "- **P3**: Minor. Works but could be tighter. Weak causal link.\n\n" "**Intra-episode:**\n" "- **P1**: A Kill Box section transition that breaks the episode's internal " "logic. The audience doesn't understand why we're in this new beat.\n" "- **P2**: A convenient plot device — an obstacle, location, or character " "that appears without setup to serve the story. The audience feels something " "is off even if they can't name it.\n" "- **P3**: Works but lazy. A stronger causal connector would improve flow.\n\n" "**Only report transitions that have issues.** Do NOT include transitions " "that work well. Keep output focused on actionable findings.\n\n" ) # ─── EPISODES ────────────────────────────────────────────────────── sections.append( f"{'='*60}\n" "# EPISODES\n" f"{'='*60}\n" ) for ep_num in sorted(corpus["episodes"].keys()): sections.append( f"\n{'='*60}\n" f"EPISODE {ep_num}\n" f"{'='*60}\n\n" f"{corpus['episodes'][ep_num]}\n" ) return "\n".join(sections) def format_verify_payload(corpus: dict, brief_path: Path) -> str: """Format verification payload (Prompt C): revised corpus + original brief.""" brief = json.loads(brief_path.read_text(encoding="utf-8")) sections = [] sections.append( "# VERIFICATION REQUEST\n\n" "You previously identified the following issues in this series. " "The series has been revised. Read the complete revised series below " "and verify each finding.\n\n" "## Original Findings\n\n" ) # Support both v1 (category) and v2 (dimension) briefs for f in brief.get("findings", []): dim = f.get("dimension", f.get("category", "unknown")) sections.append( f"- **{f['id']}** [{f['severity']}] [{dim}] {f['title']}: " f"{f['description'][:200]}...\n" ) # Add the full revised corpus sections.append("\n# REVISED EPISODE SCRIPTS\n") for ep_num in sorted(corpus["episodes"].keys()): sections.append( f"\n{'='*60}\n" f"EPISODE {ep_num}\n" f"{'='*60}\n\n" f"{corpus['episodes'][ep_num]}\n" ) sections.append( f"\n{'='*60}\n" "# VERIFICATION ANALYSIS\n" f"{'='*60}\n\n" "For each original finding, assess: RESOLVED / PARTIALLY_RESOLVED / UNRESOLVED\n" "Also check for NEW issues introduced by the revisions.\n\n" "Return ONLY valid JSON:\n" "```\n" "{\n" ' "verification_summary": "Overall assessment of revision quality",\n' ' "findings_status": [\n' " {\n" ' "id": "F001",\n' ' "status": "RESOLVED|PARTIALLY_RESOLVED|UNRESOLVED",\n' ' "notes": "What changed and whether it worked"\n' " }\n" " ],\n" ' "new_issues": [\n' " {\n" ' "id": "N001",\n' ' "dimension": "voice|pattern_fatigue|arc_earning|continuity|texture_tone_vitality|exposition_load|predictable_reversal|on_the_nose|integration|uncategorized",\n' ' "severity": "P1|P2|P3",\n' ' "title": "Short descriptive title",\n' ' "description": "What new issue was introduced",\n' ' "episodes": [4, 7, 14],\n' ' "annotations": [\n' ' {"episode": 4, "line": 0, "action": "REWRITE|DELETE|FLAG", ' '"selected_text": "Exact quote", "note": "[N001/P2] Fix guidance"}\n' " ]\n" " }\n" " ],\n" ' "overall_status": "PASS|NEEDS_WORK"\n' "}\n" "```\n" ) return "\n".join(sections) def format_deep_fix_payload( finding: dict, corpus: dict, project: str ) -> str: """ Format deep fix payload (Prompt B) for a structural P1 finding. Used for FLAG findings that need creative bridge content — unearned arcs, pacing restructuring, thread resolution. Takes the specific finding, relevant character data, and affected episode texts. """ finding_id = finding.get("id", "???") dimension = finding.get("dimension", finding.get("category", "unknown")) title = finding.get("title", "") description = finding.get("description", "") affected_episodes = finding.get("episodes", []) sections = [] # ─── ROLE ───────────────────────────────────────────────────────── sections.append( "# ROLE\n\n" "You are a veteran script doctor. You've already diagnosed a structural " "issue in this series. Now you need to FIX it — not just flag it, but " "write the actual bridge content, rewrite guidance, and continuity notes " "that will resolve the problem.\n\n" "Your output will be used by a revision agent to modify specific episodes. " "Be concrete: write actual scene drafts, not just suggestions.\n" ) # ─── THE FINDING ────────────────────────────────────────────────── sections.append( f"# THE FINDING: {finding_id}\n\n" f"**Dimension:** {dimension}\n" f"**Severity:** {finding.get('severity', '?')}\n" f"**Title:** {title}\n" f"**Description:** {description}\n" f"**Affected episodes:** {affected_episodes}\n" ) # Include annotations from the finding if present annotations = finding.get("annotations", []) if annotations: sections.append("\n**Diagnostic annotations:**\n") for ann in annotations: sections.append( f"- Episode {ann.get('episode')}: [{ann.get('action')}] " f"{ann.get('note', '')}\n" ) # ─── CONSTRAINTS ────────────────────────────────────────────────── sections.append( "\n# CONSTRAINTS\n\n" "- Each episode is 450-500 words. Any new content must fit this budget.\n" "- Dialogue ≤40%, max 8 exchanges per episode.\n" "- The 90-second Kill Box structure (Hook/Setup/Escalation/Turn/Cliffhanger) " "must be preserved.\n" "- New content must use the character's Behavioral DNA — their voice, tics, " "idioms, physical tells. Don't write generic dialogue.\n" "- Bridge scenes should feel organic, not inserted. They need to advance " "the plot while also serving the fix.\n" ) # ─── CHARACTER DATA ─────────────────────────────────────────────── if corpus["characters"]: sections.append( "\n# CHARACTER BIBLE\n\n" f"{corpus['characters']}\n" ) # ─── AFFECTED EPISODES ──────────────────────────────────────────── sections.append("\n# AFFECTED EPISODES\n") for ep_num in sorted(affected_episodes): if ep_num in corpus["episodes"]: sections.append( f"\n{'='*60}\n" f"EPISODE {ep_num}\n" f"{'='*60}\n\n" f"{corpus['episodes'][ep_num]}\n" ) # Include surrounding episodes for context (±2 from each affected) context_eps = set() for ep_num in affected_episodes: for offset in [-2, -1, 1, 2]: ctx_ep = ep_num + offset if ctx_ep > 0 and ctx_ep not in affected_episodes and ctx_ep in corpus["episodes"]: context_eps.add(ctx_ep) if context_eps: sections.append("\n# SURROUNDING EPISODES (for context)\n") for ep_num in sorted(context_eps): sections.append( f"\n{'='*60}\n" f"EPISODE {ep_num} (context)\n" f"{'='*60}\n\n" f"{corpus['episodes'][ep_num]}\n" ) # ─── DIMENSION-SPECIFIC CHECKLIST ───────────────────────────────── checklists = { "arc_earning": ( "## Arc Fix Checklist\n" "- What intermediate beat is missing between the current states?\n" "- What does the character's Behavioral DNA say about HOW they would transition?\n" "- What shared experience between characters would earn the relationship shift?\n" "- Is the fix using character logic or plot convenience?\n" "- After the fix, does the arc still feel surprising or does it telegraph?\n" ), "texture_tone_vitality": ( "## Pacing Fix Checklist\n" "- Which episodes need their energy level changed (high→valley or valley→peak)?\n" "- Where should breathing room be inserted?\n" "- What episode would benefit from a tonal shift (humor, tenderness, wonder)?\n" "- Does the fix preserve the cliffhanger variety rules?\n" "- After the fix, does the series rhythm have enough variation?\n" ), "continuity": ( "## Thread Fix Checklist\n" "- What was planted and where? What is the earliest payoff point?\n" "- Does the resolution use information already in the story world?\n" "- Will the resolution affect any other threads or character arcs?\n" "- Does the resolution feel earned or deus ex machina?\n" "- What downstream episodes need continuity adjustments?\n" ), "predictable_reversal": ( "## Reversal Fix Checklist\n" "- Is the reversal hitting a DIFFERENT value dimension than the previous one?\n" "- Does the reversal contain irony (character gets what they wanted but not needed)?\n" "- After the fix, can the audience predict the next reversal? If yes, try again.\n" "- Does the reversal emerge from character logic, not plot mechanics?\n" "- Is there at least one reversal in this sequence that surprises the CHARACTER too?\n" ), "on_the_nose": ( "## On-the-Nose Fix Checklist\n" "- Can the stated emotion be shown through a physical behavior instead?\n" "- Can the theme be expressed through action, prop, or environmental detail?\n" "- Does the character have a Behavioral DNA stress behavior that could replace the line?\n" "- Would silence + action communicate MORE than the dialogue?\n" "- If the line must stay, can it be made oblique (say one thing, mean another)?\n" ), "integration": ( "## Integration Fix Checklist\n" "- What lesson/skill/relationship from the B-story can resolve the A-story climax?\n" "- Where can earlier episodes plant the convergence seeds?\n" "- Does the fix create thematic echo (B-story exploring same theme from different angle)?\n" "- After the fix, does the finale feel inevitable rather than coincidental?\n" "- Is the convergence bidirectional (both stories affect each other)?\n" ), } checklist = checklists.get(dimension, "") if checklist: sections.append(f"\n# DIMENSION-SPECIFIC GUIDANCE\n\n{checklist}\n") # ─── R2: SEMANTIC ESCALATION CONTEXT ───────────────────────────── # Load escalation_log.json if it exists — forces tactic category shifts escalation_log_path = _project_paths_or_exit(project).state_dir / "escalation_log.json" if escalation_log_path.exists(): try: escalation_data = json.loads(escalation_log_path.read_text(encoding="utf-8")) if escalation_data: sections.append( "\n# SEMANTIC ESCALATION CONTEXT (R2)\n\n" "The following tactic bucket classifications were tracked during generation. " "When rewriting, ensure consecutive sequences do NOT repeat the same tactic " "bucket. If two adjacent episodes both use 'violence_force', the rewrite MUST " "shift one to a different bucket.\n\n" "**Tactic buckets:** violence_force, stealth_deception, negotiation_bargaining, " "technical_hack, self_sacrifice\n\n" "**Episode classifications:**\n" ) for entry in escalation_data: ep = entry.get("episode", "?") bucket = entry.get("bucket", "unknown") sections.append(f"- Episode {ep}: {bucket}\n") sections.append("\n") except Exception: pass # Graceful skip if log is malformed # ─── R3: COVER TEST FAILURE CONTEXT ────────────────────────────── # If voice convergence was flagged, add character-specific rewrite constraints if dimension == "voice" or dimension == "on_the_nose": cover_test_notes = [] for ann in annotations: note = ann.get("note", "") if "cover test" in note.lower() or "voice convergence" in note.lower(): cover_test_notes.append(ann) if cover_test_notes: sections.append( "\n# COVER TEST FAILURES (R3)\n\n" "The following dialogue was flagged by the Cover Test — an independent " "reader could NOT identify the speaker from the dialogue alone. Rewrites " "MUST incorporate character-specific idioms, verbal tics, and speech " "patterns from the character bible.\n\n" ) for ann in cover_test_notes: sections.append( f"- Episode {ann.get('episode', '?')}: " f"\"{ann.get('selected_text', '')[:100]}...\" — {ann.get('note', '')}\n" ) # ─── OUTPUT SCHEMA ──────────────────────────────────────────────── sections.append( "\n# OUTPUT SCHEMA\n\n" "Return ONLY valid JSON:\n\n" "```\n" "{\n" f' "finding_id": "{finding_id}",\n' ' "fix_strategy": "High-level description of the fix approach",\n' ' "bridge_content": [\n' " {\n" ' "episode": 49,\n' ' "insert_after": "Exact quote from episode to insert after",\n' ' "new_content": "150-200 word scene draft that fits the word budget",\n' ' "purpose": "Why this beat matters for the arc/thread/pacing"\n' " }\n" " ],\n" ' "rewrite_guidance": [\n' " {\n" ' "episode": 41,\n' ' "selected_text": "Exact text to replace",\n' ' "replacement": "New text that serves the fix",\n' ' "reason": "Why this rewrite is needed"\n' " }\n" " ],\n" ' "continuity_check": "What downstream episodes are affected and how",\n' ' "behavioral_dna_alignment": "How the fix uses character logic rather than plot convenience"\n' "}\n" "```\n" ) return "\n".join(sections) # --------------------------------------------------------------------------- # Tic Audit Pass (deterministic — no API call) # --------------------------------------------------------------------------- # Behavioral tic patterns to scan for. # Each entry: (character, tic_name, list_of_regex_patterns, max_per_series) TIC_PATTERNS = [ ("JINX", "finger_counting", [ r'(?:thumb\s+to\s+pinky|pinky\s+to\s+thumb)', r'(?:fingers?\s+count|counting\s+(?:on\s+)?(?:her\s+)?fingers?)', r'(?:her\s+fingers?\s+(?:tap|drum|work|calculate|run))', ], 8), ("JINX", "debt_counter_touch", [ r'(?:touches?\s+(?:her\s+)?debt\s+counter)', r'(?:hand\s+(?:drifts?\s+to|goes?\s+to|finds?)\s+(?:her\s+)?(?:wrist|counter|debt))', ], 6), ("KIAN", "head_tilt", [ r'(?:head\s+tilts?|tilts?\s+(?:his\s+)?head)', r'(?:slight\s+angle|the\s+tilt)', ], 6), ("KIAN", "grid_scan", [ r'(?:grid\s+pattern|left.to.right.*top.to.bottom)', r'(?:scans?\s+(?:the\s+)?(?:room|corridor|space)\s+in\s+grid)', r'(?:eyes?\s+(?:track|sweep|scan).*(?:left|grid|pattern))', ], 5), ("KIAN", "military_rest", [ r'(?:military\s+rest|parade\s+rest)', r'(?:hands?\s+(?:find|behind)\s+(?:his\s+)?back)', ], 5), ("VAREK", "cuff_adjusting", [ r'(?:adjusts?\s+(?:his\s+)?cuffs?)', r'(?:pulling\s+chrome\s+straight)', r'(?:tugs?\s+(?:at\s+)?(?:his\s+)?(?:chrome\s+)?cuffs?)', ], 4), ("VAREK", "chrome_cuffs_mention", [ r'(?:chrome\s+cuffs?\s+(?:catching|gleaming|reflecting|glinting))', ], 5), ("JINX", "stress_laugh", [ r'(?:laughs?\s+(?:at|between|when|before|involuntar))', r'(?:bark\s+of\s+laughter)', r'(?:broken\s+"?HA"?)', ], 6), ("KIAN", "query_prefix", [ r'(?:Query:\s)', ], None), # None = frequency managed by fade schedule, just count ] def run_tic_audit(corpus: dict, project: str) -> dict: """ Deterministic tic audit — greps all episodes for behavioral tic patterns. Returns a brief-compatible dict with findings and annotations. No API call needed — this is pure local text analysis. """ print(f"\n{'='*60}", file=sys.stderr) print("TIC AUDIT — Behavioral Frequency Analysis", file=sys.stderr) print(f"{'='*60}\n", file=sys.stderr) episodes = corpus.get("episodes", {}) findings = [] all_instances = {} # tic_name -> [(ep_num, matched_text, line_num)] # Scan every episode for every tic pattern for tic_char, tic_name, patterns, max_freq in TIC_PATTERNS: instances = [] for ep_num, ep_text in sorted(episodes.items(), key=lambda x: int(x[0])): lines = ep_text.split('\n') for line_num, line in enumerate(lines, 1): for pattern in patterns: match = re.search(pattern, line, re.IGNORECASE) if match: instances.append({ 'episode': int(ep_num), 'line': line_num, 'text': line.strip()[:100], 'match': match.group(0), }) break # Only count once per line per tic all_instances[f"{tic_char}:{tic_name}"] = instances # Report ep_list = sorted(set(i['episode'] for i in instances)) count = len(instances) print( f" {tic_char:8s} {tic_name:25s} — {count:2d} instances across {len(ep_list)} episodes", file=sys.stderr, ) # Generate finding if over budget if max_freq is not None and count > max_freq: over_by = count - max_freq severity = "P1" if over_by > max_freq * 0.5 else "P2" # Build annotations for the excess instances # Keep the first max_freq instances, annotate the rest as DELETE/REWRITE annotations = [] for idx, inst in enumerate(instances): if idx >= max_freq: annotations.append({ "episode": inst['episode'], "line": inst['line'], "action": "REWRITE", "selected_text": inst['text'], "note": ( f"[TIC_AUDIT/{severity}] {tic_char} '{tic_name}' instance #{idx+1} of {count} " f"(budget: {max_freq}). REWRITE with a scene-specific alternative that shows " f"what this character is FEELING in this moment, not their default tic." ), }) finding = { "id": f"TIC_{tic_char[0]}{tic_name[:3].upper()}", "dimension": "tic_audit", "severity": severity, "title": f"{tic_char}: '{tic_name}' over budget ({count}/{max_freq})", "description": ( f"{tic_char}'s '{tic_name}' behavior appears {count} times " f"(budget: {max_freq}). Over by {over_by}. " f"Episodes: {ep_list}. " f"Each excess instance should be rewritten with a behavior specific to " f"that scene's emotional context — not a generic replacement, but an action " f"that reveals what this character is feeling RIGHT HERE." ), "episodes": ep_list, "annotations": annotations, "pass": "tic_audit", "instance_count": count, "budget": max_freq, "over_by": over_by, } findings.append(finding) # Summary total_over = sum(1 for f in findings) total_annotations = sum(len(f.get("annotations", [])) for f in findings) print(f"\n Tic audit complete: {total_over} over-budget tics, {total_annotations} annotations", file=sys.stderr) # Save tic audit data state_dir = _project_paths_or_exit(project).state_dir tic_data = { "version": "1.0", "project": project, "generated": datetime.now(timezone.utc).isoformat(), "pass_type": "tic_audit", "all_instances": { k: [ {"episode": i["episode"], "text": i["text"], "match": i["match"]} for i in v ] for k, v in all_instances.items() }, "findings": findings, "stats": { "total_findings": len(findings), "total_annotations": total_annotations, "p1_count": sum(1 for f in findings if f["severity"] == "P1"), "p2_count": sum(1 for f in findings if f["severity"] == "P2"), }, } tic_path = state_dir / "script_doctor_tic_audit.json" tic_path.write_text( json.dumps(tic_data, indent=2, ensure_ascii=False), encoding="utf-8", ) print(f" → Saved to {tic_path.name}", file=sys.stderr) return { "findings": findings, "stats": tic_data["stats"], "all_instances": all_instances, } # --------------------------------------------------------------------------- # Close-Read Pass (batched scene-level analysis) # --------------------------------------------------------------------------- # 7 batches with 2-episode overlap — every consecutive pair appears in at least one batch CLOSE_READ_BATCHES = [ (1, 10), (9, 18), (17, 26), (25, 34), (33, 42), (41, 50), (49, 60), ] def format_close_read_payload( corpus: dict, batch_start: int, batch_end: int, batch_num: int, ) -> str: """ Format the close-read prompt for a single batch of episodes. Close-read focuses on scene-level logic rather than series-level patterns: spatial logic, motivation grounding, scene-to-scene transitions, physical consistency, and filmability gate. Args: corpus: Full bundled corpus dict (episodes, characters, treatment, etc.) batch_start: First episode number in this batch (inclusive) batch_end: Last episode number in this batch (inclusive) batch_num: Batch index (1-7) for finding IDs """ sections = [] # ─── ROLE ────────────────────────────────────────────────────────── sections.append( "# ROLE\n\n" "You are a continuity editor performing a CLOSE READ of a vertical microdrama " "series. You are reading a batch of episodes carefully — tracking spatial logic, " "character motivation, physical consistency, and filmability within and across " "consecutive episodes.\n\n" "This is NOT a series-level pattern analysis. The structural transition pass has " "already verified THEREFORE/BUT narrative logic. A separate series-level pass " "evaluates voice, arc earning, pattern fatigue, and other cross-series patterns.\n\n" "Your job: **close-read these episodes and catch scene-level logic issues** — " "things that require careful sequential reading of small groups of episodes, " "not pattern-matching across 60.\n" ) # ─── CLOSE-READ CHECKLIST ────────────────────────────────────────── sections.append( "# CLOSE-READ CHECKLIST\n\n" "Apply ALL six checks to every episode in this batch. Focus on what a viewer " "would actually see.\n\n" "## 1. Spatial Logic (P1 if violated)\n" "- Track ALL location headers (`INT./EXT.` lines) in sequence across episodes\n" "- Verify characters can physically travel between consecutive locations\n" "- Check deck/level numbers — going from Level -10 to Level -7 requires traveling UP\n" "- If a location change has no stated travel, flag it with the two locations and distance\n" "- Pay attention to CONTINUOUS vs scene breaks — CONTINUOUS means same time, so " "characters can't teleport\n\n" "## 2. Motivation Grounding (P1 if entirely missing, P2 if weak)\n" "- Every major character decision must have a visible/stated reason\n" "- The reason must pass the filmability gate (camera can see it, not internal narration)\n" "- \"Why did they do X instead of Y?\" must have an answer in the visible text\n" "- Flag the specific decision and what's missing\n" "- Note: a character acting impulsively is fine IF their behavioral pattern supports it\n\n" "## 3. Scene-to-Scene Transitions (P1 if impossible, P2 if unexplained)\n" "- How did characters get from ep N's final location to ep N+1's opening?\n" "- What happened to unresolved threats/antagonists from ep N's cliffhanger?\n" "- Are injuries/status effects carrying forward?\n" "- Flag the specific gap with both episode numbers\n" "- NOTE: The structural transition pass already checks THEREFORE/BUT causality. " "Focus here on SPATIAL and PHYSICAL continuity, not narrative causality.\n\n" "## 4. Physical Consistency (P2)\n" "- Objects established in one scene must persist or be explicitly removed\n" "- Injuries carry forward (check the character bible for known injuries/conditions)\n" "- Environment details within a scene are internally consistent\n" "- Characters cannot use limbs/abilities that were established as damaged/lost\n" "- Flag the specific inconsistency with both locations\n\n" "## 5. Filmability Gate (P2)\n" "- No internal narration (character thoughts/knowledge not visible to camera)\n" "- No authorial voice (\"we see...\", \"the audience realizes...\")\n" "- All prose passes \"would a camera operator write this?\"\n" "- Flag the specific line with a filmable alternative\n" "- Examples of violations:\n" " - \"She knows he's lying\" → not filmable (how would the camera show this?)\n" " - \"Admin-class — a key to every locked door\" → internal knowledge\n" " - \"The audience sees\" → authorial voice\n\n" "## 6. Cliffhanger-Hook Contract (P1 if broken, no exceptions)\n" "Every cliffhanger makes a PROMISE to the audience. The next episode's hook must HONOR it.\n\n" "Check EVERY episode boundary (ep N cliffhanger → ep N+1 hook):\n\n" "a. THREAT RESOLUTION: If ep N ends with an active threat (enemies incoming, weapon aimed,\n" " countdown running), ep N+1 MUST show the resolution on screen. Opening \"after\" the\n" " threat was handled = P1. The audience was promised danger — deliver it or don't set it up.\n\n" "b. QUANTIFIED STAKES: If the threat has numbers (\"six drones\", \"twelve fighters\", \"forty\n" " seconds\"), the resolution must account for the numbers. \"Drone wreckage\" without showing\n" " the fight = P1. \"They escaped\" without showing how = P1.\n\n" "c. CONSTRAINT HONORING: If ep N establishes constraints (\"one way out\", \"sealed corridor\",\n" " \"no exit\"), ep N+1 must show how that constraint was overcome before characters appear\n" " elsewhere. Characters at a new location without explanation = P1.\n\n" "d. EMOTIONAL STATE CARRY: If ep N ends with a character at emotional extremes (breakdown,\n" " revelation, collapse), ep N+1 must carry that state. Resetting to operational with no\n" " transition beat = P2.\n\n" "DO NOT rationalize gaps. \"The narrative makes it work\" is not an excuse. If the cliffhanger\n" "established specific stakes, the hook must resolve them visually. This is a filmability issue\n" "— you cannot cut away from a blade at someone's throat and open on them walking calmly.\n\n" ) # ─── OUTPUT SCHEMA ────────────────────────────────────────────────── sections.append( "# OUTPUT SCHEMA\n\n" "Return ONLY valid JSON (no markdown fences, no commentary outside the JSON).\n\n" f"Finding IDs use C-prefix with batch offset: C{(batch_num - 1) * 100 + 1:03d}, " f"C{(batch_num - 1) * 100 + 2:03d}, etc.\n\n" "```\n" "{\n" ' "version": "2.0",\n' f' "batch": {batch_num},\n' f' "episodes_range": [{batch_start}, {batch_end}],\n' ' "summary": "Brief assessment of close-read quality for this batch",\n' ' "findings": [\n' " {\n" f' "id": "C{(batch_num - 1) * 100 + 1:03d}",\n' ' "dimension": "close_read",\n' ' "sub_dimension": "spatial_logic|motivation|transition|physical_consistency|filmability|cliffhanger_contract",\n' ' "severity": "P1|P2|P3",\n' ' "title": "Short descriptive title",\n' ' "description": "What is wrong, which episodes, why it matters",\n' ' "episodes": [4],\n' f' "batch": {batch_num},\n' ' "annotations": [\n' " {\n" ' "episode": 4,\n' ' "line": 0,\n' ' "action": "REWRITE|FLAG",\n' ' "selected_text": "Exact text from the episode",\n' ' "note": "[C001/P1] Issue description. REPLACEMENT: \\"Fixed text.\\""\n' " }\n" " ]\n" " }\n" " ],\n" ' "stats": {\n' ' "total_findings": 0,\n' ' "p1_count": 0,\n' ' "p2_count": 0,\n' ' "p3_count": 0,\n' ' "total_annotations": 0\n' " }\n" "}\n" "```\n\n" "**Rules:**\n" "- Only report issues. Do NOT list things that are fine.\n" "- Every finding MUST include at least one annotation with exact selected_text.\n" "- For REWRITE annotations, include REPLACEMENT text in the note.\n" "- For spatial logic: include both locations in the description.\n" "- If this batch has no issues, return an empty findings array.\n" ) # ─── SEVERITY DEFINITIONS ─────────────────────────────────────────── sections.append( "# SEVERITY DEFINITIONS\n\n" "- **P1 — Must fix.** Spatial impossibility, missing motivation for major decision, " "impossible scene transition, injuries used that were established as lost, " "broken cliffhanger-hook contract (skipped threat resolution, unaccounted stakes, " "violated constraints).\n" "- **P2 — Should fix.** Weak motivation, unexplained location change (inferrable " "but not stated), minor physical inconsistency, filmability violation.\n" "- **P3 — Consider.** Very minor — cosmetic spatial note, slight filmability " "concern that doesn't really damage the read.\n" ) # ─── CHARACTER BIBLE ──────────────────────────────────────────────── if corpus["characters"]: sections.append( "\n# CHARACTER BIBLE\n\n" "Use this for injury tracking, ability tracking, and motivation validation:\n\n" f"{corpus['characters']}\n" ) # ─── TREATMENT EXCERPT ────────────────────────────────────────────── if corpus["treatment"]: # Extract treatment entries for just these episodes treatment_lines = corpus["treatment"].split("\n") relevant_lines = [] capturing = False for line in treatment_lines: # Look for episode headers like "## Episode 4" or "### Ep 4" ep_match = re.search(r"(?:episode|ep)\s*(\d+)", line, re.IGNORECASE) if ep_match: ep_n = int(ep_match.group(1)) capturing = batch_start <= ep_n <= batch_end if capturing: relevant_lines.append(line) if relevant_lines: sections.append( f"\n# TREATMENT EXCERPT (Episodes {batch_start}-{batch_end})\n\n" "Compare intended story vs actual execution:\n\n" + "\n".join(relevant_lines) + "\n" ) # ─── EPISODES ─────────────────────────────────────────────────────── sections.append( f"\n{'='*60}\n" f"# EPISODES {batch_start}-{batch_end}\n" f"{'='*60}\n" ) for ep_num in range(batch_start, batch_end + 1): if ep_num in corpus["episodes"]: sections.append( f"\n{'='*60}\n" f"EPISODE {ep_num}\n" f"{'='*60}\n\n" f"{corpus['episodes'][ep_num]}\n" ) return "\n".join(sections) def _deduplicate_close_read_findings(all_findings: list) -> list: """ Deduplicate findings from overlapping batches. If the same issue is found in two batches (same episode, same sub_dimension, similar title), keep the one from the earlier batch (it ran first). """ seen = {} # key: (episode_tuple, sub_dimension) → finding deduped = [] for f in all_findings: eps = tuple(sorted(f.get("episodes", []))) sub_dim = f.get("sub_dimension", "") key = (eps, sub_dim) if key in seen: # Duplicate — skip (keep the earlier one) continue seen[key] = f deduped.append(f) return deduped def run_close_read( args, corpus: dict, model_name: str, tracker=None ) -> dict: """ Run the close-read pass: 7 batches with 2-episode overlap. Returns the merged close-read brief dict. """ pp = _project_paths_or_exit(args.project) state_dir = pp.state_dir state_dir.mkdir(parents=True, exist_ok=True) genai = get_gemini_client() if tracker is None: tracker = CostTracker(pp.project_root) all_findings = [] batch_summaries = [] for batch_num, (start, end) in enumerate(CLOSE_READ_BATCHES, 1): # Skip batches with no episodes in corpus batch_eps = [e for e in range(start, end + 1) if e in corpus["episodes"]] if not batch_eps: continue print( f" Batch {batch_num}/7 — Episodes {start}-{end} " f"({len(batch_eps)} episodes)...", file=sys.stderr, ) payload = format_close_read_payload(corpus, start, end, batch_num) response_text, usage, elapsed_ms = call_gemini(genai, payload, model_name) _log_cost( tracker, model_name, usage, elapsed_ms, f"Script doctor close-read batch {batch_num} (eps {start}-{end})" ) batch_brief = parse_gemini_response(response_text) batch_findings = batch_brief.get("findings", []) # Tag each finding with pass marker for f in batch_findings: f["pass"] = "close_read" f["batch"] = batch_num all_findings.extend(batch_findings) batch_summaries.append(batch_brief.get("summary", "")) batch_stats = batch_brief.get("stats", {}) print( f" → {batch_stats.get('total_findings', len(batch_findings))} findings," f" {batch_stats.get('total_annotations', 0)} annotations", file=sys.stderr, ) # Deduplicate findings from overlap pre_dedup = len(all_findings) all_findings = _deduplicate_close_read_findings(all_findings) if pre_dedup != len(all_findings): print( f" Deduplication: {pre_dedup} → {len(all_findings)} findings " f"({pre_dedup - len(all_findings)} duplicates removed)", file=sys.stderr, ) # Renumber all findings with clean C-prefix sequence for i, f in enumerate(all_findings, 1): f["id"] = f"C{i:03d}" # Calculate stats p1 = sum(1 for f in all_findings if f.get("severity") == "P1") p2 = sum(1 for f in all_findings if f.get("severity") == "P2") p3 = sum(1 for f in all_findings if f.get("severity") == "P3") total_ann = sum(len(f.get("annotations", [])) for f in all_findings) close_read_brief = { "version": "2.0", "project": args.project, "generated": datetime.now(timezone.utc).isoformat(), "model": model_name, "pass_type": "close_read", "summary": " | ".join(s for s in batch_summaries if s), "findings": all_findings, "stats": { "total_findings": len(all_findings), "p1_count": p1, "p2_count": p2, "p3_count": p3, "total_annotations": total_ann, }, } # Save close-read results cr_path = state_dir / "script_doctor_close_read.json" cr_path.write_text( json.dumps(close_read_brief, indent=2, ensure_ascii=False), encoding="utf-8", ) print(f" → Close-read saved to {cr_path.name}", file=sys.stderr) return close_read_brief # --------------------------------------------------------------------------- # Gemini API Interaction # --------------------------------------------------------------------------- def get_gemini_client(): """Initialize and return Gemini client.""" try: import google.generativeai as genai except ImportError: print( "ERROR: google-generativeai package not installed.\n" "Run: pip install google-generativeai", file=sys.stderr, ) sys.exit(1) api_key = os.environ.get("GEMINI_API_KEY") if not api_key: print( "ERROR: GEMINI_API_KEY not set.\n" "Run: export GEMINI_API_KEY=\"your-key-here\"\n" "Get a key at: https://ai.google.dev/tutorials/setup", file=sys.stderr, ) sys.exit(1) genai.configure(api_key=api_key) return genai def call_gemini(genai, payload: str, model_name: str = DEFAULT_MODEL) -> tuple: """Send payload to Gemini and return (response_text, usage_metadata_dict, elapsed_ms).""" model = genai.GenerativeModel(model_name) print(f"Sending to {model_name}...", file=sys.stderr) print(f"Payload size: ~{len(payload.split())} words", file=sys.stderr) t0 = time.monotonic() response = model.generate_content( payload, generation_config={ "temperature": 0.3, # Low temp for analytical precision "max_output_tokens": 32000, # Room for detailed findings }, ) elapsed_ms = int((time.monotonic() - t0) * 1000) # Extract token counts from usage_metadata if available usage = {} um = getattr(response, "usage_metadata", None) if um: usage["prompt_token_count"] = getattr(um, "prompt_token_count", 0) or 0 usage["candidates_token_count"] = getattr(um, "candidates_token_count", 0) or 0 return response.text, usage, elapsed_ms def _log_cost(tracker, model_name, usage, elapsed_ms, detail): """Log a Gemini call to the cost tracker (no-op if tracker is None).""" if tracker is None: return tracker.log( category="analysis", provider="gemini", model=model_name, tokens_in=usage.get("prompt_token_count", 0), tokens_out=usage.get("candidates_token_count", 0), detail=detail, success=True, duration_ms=elapsed_ms, ) def parse_gemini_response(response_text: str) -> dict: """Extract JSON from Gemini's response, handling markdown fences.""" # Strip markdown code fences if present text = response_text.strip() if text.startswith("```"): # Remove opening fence (possibly with language tag) text = re.sub(r"^```\w*\n?", "", text) if text.endswith("```"): text = text[:-3] text = text.strip() try: return json.loads(text) except json.JSONDecodeError as e: # Try to find JSON object in the response match = re.search(r"\{[\s\S]*\}", text) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass print( f"ERROR: Could not parse Gemini response as JSON.\n" f"Parse error: {e}\n" f"Response preview: {text[:500]}...", file=sys.stderr, ) sys.exit(1) # --------------------------------------------------------------------------- # Brief Management # --------------------------------------------------------------------------- def save_brief(brief: dict, project: str, pass_type: str = "diagnostic") -> Path: """Save the structured brief to the project state directory.""" state_dir = _project_paths_or_exit(project).state_dir state_dir.mkdir(parents=True, exist_ok=True) brief_with_meta = { "version": "2.0", "project": project, "generated": datetime.now(timezone.utc).isoformat(), "model": DEFAULT_MODEL, "pass_type": pass_type, **brief, } if pass_type == "diagnostic": out_path = state_dir / "script_doctor_brief.json" else: out_path = state_dir / "script_doctor_verify.json" out_path.write_text( json.dumps(brief_with_meta, indent=2, ensure_ascii=False), encoding="utf-8", ) return out_path def print_summary(brief: dict, pass_type: str = "diagnostic"): """Print a human-readable summary of the brief.""" stats = brief.get("stats", {}) findings = brief.get("findings", []) if pass_type == "diagnostic": print(f"\n{'='*60}") print("SCRIPT DOCTOR — DIAGNOSTIC COMPLETE") print(f"{'='*60}") print(f"\nSummary: {brief.get('summary', 'N/A')}\n") # Transition stats if present (supports both old flat and new nested format) t_stats = brief.get("transition_stats", {}) if t_stats: # Check for nested format (inter_episode / intra_episode) inter = t_stats.get("inter_episode", {}) intra = t_stats.get("intra_episode", {}) if inter: print("Inter-Episode Transitions:") print(f" THEREFORE: {inter.get('therefore_count', '?')}") print(f" BUT: {inter.get('but_count', '?')}") print(f" AND THEN: {inter.get('and_then_count', '?')}") if intra: print("Intra-Episode Transitions (Kill Box):") print(f" THEREFORE: {intra.get('therefore_count', '?')}") print(f" BUT: {intra.get('but_count', '?')}") print(f" AND THEN: {intra.get('and_then_count', '?')}") # Fallback: old flat format if not inter and not intra: print("Transitions:") print(f" THEREFORE: {t_stats.get('therefore_count', '?')}") print(f" BUT: {t_stats.get('but_count', '?')}") print(f" AND THEN: {t_stats.get('and_then_count', '?')}") print() print(f"Total findings: {stats.get('total_findings', len(findings))}") print(f" P1 (must fix): {stats.get('p1_count', 0)}") print(f" P2 (should fix): {stats.get('p2_count', 0)}") print(f" P3 (consider): {stats.get('p3_count', 0)}") # Separate structural, close-read, and series findings for display structural_dims = {"transitions", "transitions_internal"} structural = [f for f in findings if f.get("pass") == "structural" or f.get("dimension") in structural_dims] close_read = [f for f in findings if f.get("pass") == "close_read" or f.get("dimension") == "close_read"] series = [f for f in findings if f not in structural and f not in close_read] # Further split structural into inter (T-prefix) and intra (I-prefix) inter_findings = [f for f in structural if f.get("id", "").startswith("T")] intra_findings = [f for f in structural if f.get("id", "").startswith("I")] # Print inter-episode P1s p1_inter = [f for f in inter_findings if f.get("severity") == "P1"] if p1_inter: print("\nP1 INTER-EPISODE (transitions):") for f in p1_inter: eps = f.get("episodes", []) ep_str = f"{len(eps)} episodes" if len(eps) > 5 else str(eps) ann_count = len(f.get("annotations", [])) ann_str = f", {ann_count} annotations" if ann_count else "" connector = f" [{f['connector']}]" if "connector" in f else "" print(f" {f['id']} {f['title']}{connector} ({ep_str}{ann_str})") # Print intra-episode findings (P1/P2 Kill Box only; P3 intra not surfaced) if intra_findings: p1_intra = [f for f in intra_findings if f.get("severity") == "P1"] p2_intra = [f for f in intra_findings if f.get("severity") == "P2"] if p1_intra: print("\nP1 INTRA-EPISODE (Kill Box):") for f in p1_intra: kb = f.get("kill_box_transition", "") ann_count = len(f.get("annotations", [])) ann_str = f", {ann_count} annotations" if ann_count else "" print(f" {f['id']} {f['title']} [{kb}] ({ann_str})") if p2_intra: print("\nP2 INTRA-EPISODE (Kill Box):") for f in p2_intra: kb = f.get("kill_box_transition", "") ann_count = len(f.get("annotations", [])) ann_str = f", {ann_count} annotations" if ann_count else "" print(f" {f['id']} {f['title']} [{kb}] ({ann_str})") # Close-read findings if close_read: p1_cr = [f for f in close_read if f.get("severity") == "P1"] p2_cr = [f for f in close_read if f.get("severity") == "P2"] if p1_cr: print("\nP1 CLOSE-READ:") for f in p1_cr: sub = f.get("sub_dimension", "") ann_count = len(f.get("annotations", [])) ann_str = f", {ann_count} annotations" if ann_count else "" print(f" {f['id']} [{sub}] {f.get('title', '')} ({ann_str})") if p2_cr: print("\nP2 CLOSE-READ:") for f in p2_cr: sub = f.get("sub_dimension", "") ann_count = len(f.get("annotations", [])) ann_str = f", {ann_count} annotations" if ann_count else "" print(f" {f['id']} [{sub}] {f.get('title', '')} ({ann_str})") # Then series P1s p1_series = [f for f in series if f.get("severity") == "P1"] if p1_series: print("\nP1 SERIES-LEVEL:") for f in p1_series: eps = f.get("episodes", []) ep_str = f"{len(eps)} episodes" if len(eps) > 5 else str(eps) dim = f.get("dimension", f.get("category", "?")) ann_count = len(f.get("annotations", [])) ann_str = f", {ann_count} annotations" if ann_count else "" print(f" {f['id']} [{dim}] {f['title']} ({ep_str}{ann_str})") # Character grades grades = brief.get("character_grades", {}) if grades: print("\nCHARACTER GRADES:") for name, info in grades.items(): print(f" {name}: {info.get('grade', '?')} — {info.get('arc_summary', '')}") else: # verification print(f"\n{'='*60}") print("SCRIPT DOCTOR — VERIFICATION") print(f"{'='*60}") print(f"\n{brief.get('verification_summary', 'N/A')}\n") for fs in brief.get("findings_status", []): status_icon = {"RESOLVED": "+", "PARTIALLY_RESOLVED": "~", "UNRESOLVED": "!"} icon = status_icon.get(fs["status"], "?") print(f" [{icon}] {fs['id']}: {fs['status']} — {fs.get('notes', '')}") new_issues = brief.get("new_issues", []) if new_issues: print(f"\nNEW ISSUES: {len(new_issues)}") for ni in new_issues: print(f" {ni['id']} [{ni['severity']}] {ni['title']}") print(f"\nOVERALL: {brief.get('overall_status', 'UNKNOWN')}") print(f"\n{'='*60}") # --------------------------------------------------------------------------- # Brief → Annotations Extraction (for /revise) # --------------------------------------------------------------------------- def extract_annotations(brief_path: Path, project: str) -> Path: """ Extract annotations directly from a v2 brief's findings. Since Gemini now produces annotations directly with exact episode quotes, extraction is just flattening the nested structure — no searching episode files for quote matches. For v1 briefs (no annotations array in findings), falls back to legacy conversion. """ brief = json.loads(brief_path.read_text(encoding="utf-8")) # Detect v1 brief (has evidence array but no annotations in findings) is_v1 = False findings = brief.get("findings", []) if findings and "annotations" not in findings[0] and "evidence" in findings[0]: is_v1 = True if is_v1: print( "Detected v1 brief (evidence-based). Using legacy conversion.", file=sys.stderr, ) return _legacy_brief_to_annotations(brief_path, project) # v2 extraction: flatten annotations from findings all_annotations = [] for finding in findings: fid = finding.get("id", "?") severity = finding.get("severity", "P3") dimension = finding.get("dimension", finding.get("category", "unknown")) for ann in finding.get("annotations", []): all_annotations.append({ "episode": ann.get("episode"), "line": ann.get("line", 0), "action": ann.get("action", "FLAG"), "selected_text": ann.get("selected_text", ""), "note": ann.get("note", f"[{fid}/{severity}]"), "finding_id": fid, "severity": severity, "dimension": dimension, }) # Sort by episode, then line number all_annotations.sort(key=lambda a: (a.get("episode", 0), a.get("line", 0))) # Build the output state_dir = _project_paths_or_exit(project).state_dir state_dir.mkdir(parents=True, exist_ok=True) output = { "source": f"script_doctor_brief ({brief.get('generated', 'unknown')})", "project": project, "generated": datetime.now(timezone.utc).isoformat(), "brief_version": brief.get("version", "2.0"), "annotations": all_annotations, "edits": [], # No formatting edits from script doctor "stats": { "total_annotations": len(all_annotations), "rewrite_count": sum(1 for a in all_annotations if a["action"] == "REWRITE"), "flag_count": sum(1 for a in all_annotations if a["action"] == "FLAG"), "delete_count": sum(1 for a in all_annotations if a["action"] == "DELETE"), "episodes_affected": len(set(a["episode"] for a in all_annotations)), }, } out_path = state_dir / "script_doctor_annotations.json" out_path.write_text( json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8", ) return out_path def _legacy_brief_to_annotations(brief_path: Path, project: str) -> Path: """ Legacy conversion for v1 briefs (evidence-based, no inline annotations). Substitution findings (repetition, voice, exposition) → REWRITE annotations Structural findings (arc, pacing, threads) → FLAG annotations Each evidence quote is matched against the actual episode file to get the correct line number. This is the brittle path — v2 briefs avoid it. """ brief = json.loads(brief_path.read_text(encoding="utf-8")) pp = _project_paths_or_exit(project) episodes_dir = pp.episodes_dir if not episodes_dir.is_dir(): print(f"ERROR: Episodes directory not found: {episodes_dir}", file=sys.stderr) sys.exit(1) if not list(episodes_dir.glob("ep_*.md")): print(f"ERROR: No episode files found in {episodes_dir}", file=sys.stderr) sys.exit(1) annotations = [] skipped = 0 substitution_categories = {"repetition", "voice", "exposition"} # (structural categories — arc/pacing/threads — are not branched on here) for finding in brief.get("findings", []): fid = finding.get("id", "?") category = finding.get("category", finding.get("dimension", "")) severity = finding.get("severity", "P3") title = finding.get("title", "") recommendation = finding.get("recommendation", "") suggested_cuts = set(finding.get("suggested_cuts", [])) for ev in finding.get("evidence", []): ep_num = ev.get("episode") quote = ev.get("quote", "") if not ep_num or not quote: continue suggested_keeps = set(finding.get("suggested_keeps", [])) if ep_num in suggested_keeps: continue if suggested_cuts and ep_num not in suggested_cuts: continue ep_file = episodes_dir / f"ep_{ep_num:03d}.md" if not ep_file.exists(): skipped += 1 continue ep_text = ep_file.read_text(encoding="utf-8") ep_lines = ep_text.split("\n") line_num = None quote_clean = " ".join(quote.split()).lower() for i, line in enumerate(ep_lines, 1): line_clean = " ".join(line.split()).lower() if quote_clean in line_clean or line_clean in quote_clean: line_num = i break if line_num is None: partial = quote_clean[:30] for i, line in enumerate(ep_lines, 1): if partial in " ".join(line.split()).lower(): line_num = i break if line_num is None: line_num = 0 skipped += 1 if category in substitution_categories: action = "REWRITE" else: action = "FLAG" note = f"[{fid}/{severity}] {title}. {recommendation}" annotations.append({ "episode": ep_num, "line": line_num, "action": action, "selected_text": quote, "note": note, "finding_id": fid, "severity": severity, }) annotations.sort(key=lambda a: (a["episode"], a["line"])) state_dir = pp.state_dir state_dir.mkdir(parents=True, exist_ok=True) output = { "source": f"script_doctor_brief ({brief.get('generated', 'unknown')})", "project": project, "generated": datetime.now(timezone.utc).isoformat(), "brief_version": brief.get("version", "1.0"), "annotations": annotations, "edits": [], "stats": { "total_annotations": len(annotations), "rewrite_count": sum(1 for a in annotations if a["action"] == "REWRITE"), "flag_count": sum(1 for a in annotations if a["action"] == "FLAG"), "delete_count": sum(1 for a in annotations if a["action"] == "DELETE"), "episodes_affected": len(set(a["episode"] for a in annotations)), "quotes_not_found": skipped, }, } out_path = state_dir / "script_doctor_annotations.json" out_path.write_text( json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8", ) return out_path # --------------------------------------------------------------------------- # Full Diagnostic Pipeline # --------------------------------------------------------------------------- ALL_DIMENSIONS = [ "transitions", # Pass 1: structural (runs first) "voice", "pattern_fatigue", "arc_earning", "continuity", "texture_tone_vitality", "exposition_load", "predictable_reversal", "on_the_nose", "integration", # R1, R4, R5 refinement gates ] # Dimensions handled by Pass 1 (structural) vs Pass 2 (series-level) STRUCTURAL_DIMENSIONS = ["transitions", "transitions_internal"] SERIES_DIMENSIONS = [d for d in ALL_DIMENSIONS if d not in STRUCTURAL_DIMENSIONS] def merge_briefs( broad: dict, focus_list: list, structural_brief: dict = None, close_read_brief: dict = None, tic_audit_brief: dict = None, ) -> dict: """ Merge all pass results into a single brief. Ordering: structural (T/I-prefixed) → close-read (C-prefixed) → tic-audit (TIC-prefixed) → series-level (F-prefixed). """ # Start with structural findings (Pass 1) — they sort first structural_findings = [] if structural_brief: for f in structural_brief.get("findings", []): structural_findings.append(f) # Close-read findings (Pass 4) — sort between structural and series cr_findings = [] if close_read_brief: cr_findings = list(close_read_brief.get("findings", [])) # Tic audit findings (deterministic) — sort between close-read and series tic_findings = [] if tic_audit_brief: tic_findings = list(tic_audit_brief.get("findings", [])) # Then series-level findings (Pass 2 broad + Pass 3 focused) series_findings = list(broad.get("findings", [])) # Track existing finding IDs to renumber focus findings max_id = 0 for f in series_findings: match = re.search(r"F(\d+)", f.get("id", "")) if match: max_id = max(max_id, int(match.group(1))) for focus_brief in focus_list: for f in focus_brief.get("findings", []): max_id += 1 f["id"] = f"F{max_id:03d}" series_findings.append(f) # Combined: structural first, then close-read, then tic audit, then series all_findings = structural_findings + cr_findings + tic_findings + series_findings # Merge character grades (broad has full context, focus may add specifics) merged_grades = dict(broad.get("character_grades", {})) for focus_brief in focus_list: for char, grades in focus_brief.get("character_grades", {}).items(): if char in merged_grades: existing = merged_grades[char] focus_vd = grades.get("voice_diversification", "") if focus_vd and focus_vd != existing.get("voice_diversification", ""): existing["voice_diversification"] = ( existing.get("voice_diversification", "") + " " + focus_vd ).strip() else: merged_grades[char] = grades # Merge workflow observations from all passes all_observations = [] if structural_brief: all_observations.extend(structural_brief.get("workflow_observations", [])) all_observations.extend(broad.get("workflow_observations", [])) for focus_brief in focus_list: all_observations.extend(focus_brief.get("workflow_observations", [])) # Recalculate stats p1 = sum(1 for f in all_findings if f.get("severity") == "P1") p2 = sum(1 for f in all_findings if f.get("severity") == "P2") p3 = sum(1 for f in all_findings if f.get("severity") == "P3") total_ann = sum(len(f.get("annotations", [])) for f in all_findings) # Build summary structural_summary = "" if structural_brief: t_stats = structural_brief.get("transition_stats", {}) # Support nested format (inter_episode/intra_episode) or flat fallback inter = t_stats.get("inter_episode", {}) intra = t_stats.get("intra_episode", {}) if inter: inter_at = inter.get("and_then_count", 0) inter_t = inter.get("therefore_count", 0) inter_b = inter.get("but_count", 0) intra_at = intra.get("and_then_count", 0) if intra else 0 structural_summary = ( f"[Structural pass: inter-episode {inter_at} AND THEN / " f"{inter_t} THEREFORE / {inter_b} BUT" ) if intra: intra_t = intra.get("therefore_count", 0) intra_b = intra.get("but_count", 0) structural_summary += ( f"; intra-episode {intra_at} AND THEN / " f"{intra_t} THEREFORE / {intra_b} BUT" ) structural_summary += "] " else: # Flat fallback for old briefs structural_summary = ( f"[Structural pass: {t_stats.get('and_then_count', 0)} AND THEN transitions, " f"{t_stats.get('therefore_count', 0)} THEREFORE, " f"{t_stats.get('but_count', 0)} BUT] " ) close_read_summary = "" if close_read_brief: cr_stats = close_read_brief.get("stats", {}) close_read_summary = ( f"[Close-read: {cr_stats.get('total_findings', 0)} findings, " f"{cr_stats.get('p1_count', 0)} P1] " ) merged_summary = structural_summary + close_read_summary + broad.get("summary", "") if focus_list: focus_dims = [] for fb in focus_list: for f in fb.get("findings", []): dim = f.get("dimension", "") if dim and dim not in focus_dims: focus_dims.append(dim) if focus_dims: merged_summary += ( f" [Extended with focused analysis on: {', '.join(focus_dims)}]" ) result = { "summary": merged_summary, "findings": all_findings, "character_grades": merged_grades, "workflow_observations": all_observations, "stats": { "total_findings": len(all_findings), "p1_count": p1, "p2_count": p2, "p3_count": p3, "total_annotations": total_ann, }, } # Preserve transition stats in merged brief if structural_brief and "transition_stats" in structural_brief: result["transition_stats"] = structural_brief["transition_stats"] return result def run_full_diagnostic( args, corpus: dict, model_name: str ) -> Path: """ Run complete automated diagnostic pipeline: Phase 1 — Structural pass (transitions: THEREFORE/BUT vs AND THEN) Phase 2 — Broad diagnostic (9 series-level dimensions) Phase 3 — Focused passes on undercovered dimensions Phase 4 — Close-read pass (batched scene-level analysis) Phase 5 — Deep fix for P1 FLAG findings Final — Merge everything into one canonical brief """ pp = _project_paths_or_exit(args.project) state_dir = pp.state_dir state_dir.mkdir(parents=True, exist_ok=True) genai = get_gemini_client() tracker = CostTracker(pp.project_root) # ── PHASE 1: Structural pass (transitions) ───────────────────────── print(f"\n{'='*60}", file=sys.stderr) print("PHASE 1/5 — STRUCTURAL PASS (transitions)", file=sys.stderr) print(f"{'='*60}\n", file=sys.stderr) print(" Checking inter-episode causality: THEREFORE/BUT vs AND THEN", file=sys.stderr) print(" Checking spatial continuity, character positioning, emotional carry-over\n", file=sys.stderr) transition_payload = format_transition_payload(corpus) transition_response, transition_usage, transition_elapsed = call_gemini( genai, transition_payload, model_name ) _log_cost( tracker, model_name, transition_usage, transition_elapsed, "Script doctor Pass 1: structural transitions" ) transition_brief = parse_gemini_response(transition_response) # Tag all transition findings with pass marker for f in transition_brief.get("findings", []): f["pass"] = "structural" # Save structural results separately structural_path = state_dir / "script_doctor_structural.json" structural_with_meta = { "version": "2.0", "project": args.project, "generated": datetime.now(timezone.utc).isoformat(), "model": model_name, "pass_type": "structural", **transition_brief, } structural_path.write_text( json.dumps(structural_with_meta, indent=2, ensure_ascii=False), encoding="utf-8", ) # Print transition summary t_stats = transition_brief.get("stats", {}) t_trans = transition_brief.get("transition_stats", {}) print("\n Transition analysis:", file=sys.stderr) print(f" THEREFORE: {t_trans.get('therefore_count', '?')}", file=sys.stderr) print(f" BUT: {t_trans.get('but_count', '?')}", file=sys.stderr) print(f" AND THEN: {t_trans.get('and_then_count', '?')}", file=sys.stderr) print( f" → {t_stats.get('total_findings', 0)} findings," f" {t_stats.get('total_annotations', 0)} annotations" f" → {structural_path.name}", file=sys.stderr, ) # ── PHASE 2: Broad diagnostic (9 series-level dimensions) ────────── print(f"\n{'='*60}", file=sys.stderr) print("PHASE 2/5 — BROAD DIAGNOSTIC (9 series-level dimensions)", file=sys.stderr) print(f"{'='*60}\n", file=sys.stderr) payload = format_corpus_payload(corpus, focus=None) response_text, usage, elapsed_ms = call_gemini(genai, payload, model_name) _log_cost(tracker, model_name, usage, elapsed_ms, "Script doctor Pass 2: broad diagnostic") broad_brief = parse_gemini_response(response_text) # Tag series-level findings for f in broad_brief.get("findings", []): f["pass"] = "series" # Save broad results separately broad_path = state_dir / "script_doctor_broad.json" broad_with_meta = { "version": "2.0", "project": args.project, "generated": datetime.now(timezone.utc).isoformat(), "model": model_name, "pass_type": "broad", **broad_brief, } broad_path.write_text( json.dumps(broad_with_meta, indent=2, ensure_ascii=False), encoding="utf-8", ) print_summary(broad_brief) broad_stats = broad_brief.get("stats", {}) print( f"\n Saved broad results → {broad_path.name}" f" ({broad_stats.get('total_findings', 0)} findings," f" {broad_stats.get('total_annotations', 0)} annotations)", file=sys.stderr, ) # ── Identify gap dimensions (series-level only) ──────────────────── covered_dims = set() for f in broad_brief.get("findings", []): dim = f.get("dimension", f.get("category", "")) if dim: covered_dims.add(dim) gap_dims = [d for d in SERIES_DIMENSIONS if d not in covered_dims] # ── PHASE 3: Focused passes on gaps ──────────────────────────────── focus_briefs = [] if gap_dims: print(f"\n{'='*60}", file=sys.stderr) print( f"PHASE 3/5 — FOCUSED PASSES ({len(gap_dims)} uncovered dimensions)", file=sys.stderr, ) print(f"{'='*60}", file=sys.stderr) print(f" Broad pass covered: {sorted(covered_dims)}", file=sys.stderr) print(f" Running focused on: {gap_dims}\n", file=sys.stderr) for dim in gap_dims: print(f" ── Focus: {dim} ──", file=sys.stderr) focus_payload = format_corpus_payload(corpus, focus=[dim]) focus_response, focus_usage, focus_elapsed = call_gemini(genai, focus_payload, model_name) _log_cost(tracker, model_name, focus_usage, focus_elapsed, f"Script doctor focused pass: {dim}") focus_brief = parse_gemini_response(focus_response) # Tag focused findings for f in focus_brief.get("findings", []): f["pass"] = "series" # Save individual focus result focus_path = state_dir / f"script_doctor_focus_{dim}.json" focus_with_meta = { "version": "2.0", "project": args.project, "generated": datetime.now(timezone.utc).isoformat(), "model": model_name, "pass_type": f"focus_{dim}", **focus_brief, } focus_path.write_text( json.dumps(focus_with_meta, indent=2, ensure_ascii=False), encoding="utf-8", ) fs = focus_brief.get("stats", {}) print( f" → {fs.get('total_findings', 0)} findings," f" {fs.get('total_annotations', 0)} annotations" f" → {focus_path.name}", file=sys.stderr, ) focus_briefs.append(focus_brief) else: print( "\n All 6 series dimensions covered in broad pass. Skipping focused passes.", file=sys.stderr, ) # ── PHASE 4: Close-read pass (batched scene-level analysis) ──────── print(f"\n{'='*60}", file=sys.stderr) print("PHASE 4/5 — CLOSE READ (7 batches, 2-ep overlap)", file=sys.stderr) print(f"{'='*60}\n", file=sys.stderr) print(" Checking: spatial logic, motivation, transitions, physical consistency, filmability", file=sys.stderr) close_read_brief = run_close_read(args, corpus, model_name, tracker=tracker) cr_stats = close_read_brief.get("stats", {}) print( f"\n Close-read complete: {cr_stats.get('total_findings', 0)} findings," f" {cr_stats.get('p1_count', 0)} P1," f" {cr_stats.get('total_annotations', 0)} annotations", file=sys.stderr, ) # ── TIC AUDIT (deterministic, no API) ────────────────────────────── tic_audit_result = run_tic_audit(corpus, args.project) # ── MERGE (structural + close-read + series-level + tic audit) ─── print(f"\n{'='*60}", file=sys.stderr) print("MERGING RESULTS (structural + close-read + series-level + tic audit)", file=sys.stderr) print(f"{'='*60}", file=sys.stderr) merged = merge_briefs( broad_brief, focus_briefs, structural_brief=transition_brief, close_read_brief=close_read_brief, tic_audit_brief=tic_audit_result, ) merged_stats = merged.get("stats", {}) print( f" Combined: {merged_stats.get('total_findings', 0)} findings," f" {merged_stats.get('total_annotations', 0)} annotations", file=sys.stderr, ) # Set overall_status based on close-read P1 count if cr_stats.get("p1_count", 0) > 0: merged["overall_status"] = "NEEDS_WORK" print( f"\n CLOSE-READ GATE: FAIL — {cr_stats['p1_count']} P1 issues require fixes", file=sys.stderr, ) # ── PHASE 5: Deep Fix for P1 FLAGs ──────────────────────────────── p1_flags = [ f for f in merged.get("findings", []) if f.get("severity") == "P1" and any(a.get("action") == "FLAG" for a in f.get("annotations", [])) ] if p1_flags: print(f"\n{'='*60}", file=sys.stderr) print( f"PHASE 5/5 — DEEP FIX ({len(p1_flags)} P1 FLAG findings)", file=sys.stderr, ) print(f"{'='*60}\n", file=sys.stderr) if getattr(args, 'deep_fix_engine', 'claude') == "gemini": # Gemini path (backward compat) for finding in p1_flags: fid = finding.get("id", "???") print( f" ── Deep fix: {fid} — {finding.get('title', '')} ──", file=sys.stderr, ) fix_payload = format_deep_fix_payload(finding, corpus, args.project) fix_response, fix_usage, fix_elapsed = call_gemini(genai, fix_payload, model_name) _log_cost(tracker, model_name, fix_usage, fix_elapsed, f"Script doctor deep fix: {fid}") fix_data = parse_gemini_response(fix_response) save_deep_fix(fix_data, fid, args.project) print_deep_fix_summary(fix_data) else: # Claude path (default): save payloads for Claude Code reasoning print(f"\n Saving {len(p1_flags)} deep fix payloads for Claude reasoning...", file=sys.stderr) for finding in p1_flags: fid = finding.get("id", "???") fix_payload = format_deep_fix_payload(finding, corpus, args.project) payload_path = state_dir / f"deep_fix_{fid}_payload.txt" payload_path.write_text(fix_payload, encoding="utf-8") print(f" ── {fid}: {finding.get('title', '')} → {payload_path.name}", file=sys.stderr) print("\n Deep fix payloads saved. Run /script-doctor --deep-fix [ID] to process with Claude.", file=sys.stderr) else: print( "\n No P1 FLAG findings requiring deep fix. Skipping Phase 5.", file=sys.stderr, ) # ── FINAL: Save merged brief as canonical ────────────────────────── out_path = save_brief(merged, args.project, pass_type="diagnostic") print(f"\n{'='*60}") print("SCRIPT DOCTOR — FULL DIAGNOSTIC COMPLETE") print(f"{'='*60}") print_summary(merged) print(f"\nCanonical brief saved to: {out_path}") if focus_briefs: print(f"Individual focus files in: {state_dir}/") print(f"{'='*60}") return out_path # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def save_deep_fix(fix_data: dict, finding_id: str, project: str) -> Path: """Save deep fix output to the project state directory.""" state_dir = _project_paths_or_exit(project).state_dir state_dir.mkdir(parents=True, exist_ok=True) fix_with_meta = { "generated": datetime.now(timezone.utc).isoformat(), "model": DEFAULT_MODEL, "project": project, **fix_data, } out_path = state_dir / f"deep_fix_{finding_id}.json" out_path.write_text( json.dumps(fix_with_meta, indent=2, ensure_ascii=False), encoding="utf-8", ) return out_path def print_deep_fix_summary(fix_data: dict): """Print a human-readable summary of a deep fix.""" print(f"\n{'='*60}") print(f"SCRIPT DOCTOR — DEEP FIX: {fix_data.get('finding_id', '?')}") print(f"{'='*60}") print(f"\nStrategy: {fix_data.get('fix_strategy', 'N/A')}") bridges = fix_data.get("bridge_content", []) if bridges: print(f"\nBridge scenes: {len(bridges)}") for b in bridges: print(f" Episode {b.get('episode')}: {b.get('purpose', '')[:80]}") rewrites = fix_data.get("rewrite_guidance", []) if rewrites: print(f"\nRewrites: {len(rewrites)}") for r in rewrites: print(f" Episode {r.get('episode')}: {r.get('reason', '')[:80]}") cont = fix_data.get("continuity_check", "") if cont: print(f"\nContinuity: {cont[:200]}") dna = fix_data.get("behavioral_dna_alignment", "") if dna: print(f"\nDNA alignment: {dna[:200]}") print(f"\n{'='*60}") def main(): parser = argparse.ArgumentParser( description="Script Doctor — Full-series diagnostic via Gemini" ) parser.add_argument("project", help="Project folder name") parser.add_argument( "--bundle", action="store_true", help="Preview corpus payload (don't send to Gemini)" ) parser.add_argument( "--diagnose", action="store_true", help="Run full diagnostic" ) parser.add_argument( "--verify", action="store_true", help="Run verification pass (requires existing brief)" ) parser.add_argument( "--focus", type=str, default=None, help="Comma-separated focus areas: transitions,voice,pattern_fatigue," "arc_earning,continuity,texture_tone_vitality,exposition_load" ) parser.add_argument( "--model", type=str, default=DEFAULT_MODEL, help=f"Gemini model to use (default: {DEFAULT_MODEL})" ) parser.add_argument( "--to-annotations", action="store_true", help="Extract annotations from existing brief to /revise format" ) parser.add_argument( "--deep-fix", type=str, default=None, metavar="FINDING_ID", help="Generate creative fix for a structural finding (e.g., --deep-fix F002)" ) parser.add_argument( "--deep-fix-engine", type=str, default="claude", choices=["claude", "gemini"], help="Engine for deep fix reasoning (default: claude)" ) parser.add_argument( "--close-read-engine", type=str, default="claude", choices=["claude", "gemini"], help="Engine for close-read analysis (default: claude). " "claude: saves payloads for Opus sub-agent processing. " "gemini: sends directly to Gemini API." ) parser.add_argument( "--full", action="store_true", help="Run full automated pipeline: structural (Pass 1) → broad (Pass 2) → " "focused (Pass 3) → close-read (Pass 4) → deep-fix (Pass 5) → merged brief" ) parser.add_argument( "--close-read", action="store_true", help="Run close-read pass only (7 batches with 2-ep overlap, scene-level analysis)" ) parser.add_argument( "--dry-run", action="store_true", help="Generate payload and save to file instead of calling API" ) parser.add_argument( "--parse-response", type=str, default=None, metavar="FILE", help="Parse a manually-obtained Gemini response file (pairs with --dry-run)" ) parser.add_argument( "--parse-type", type=str, default="diagnostic", choices=["diagnostic", "verify", "deep-fix"], help="Type of response being parsed (default: diagnostic)" ) args = parser.parse_args() focus = args.focus.split(",") if args.focus else None # --- Bundle --- print("Bundling corpus...", file=sys.stderr) corpus = bundle_corpus(args.project) print( f" Episodes: {corpus['episode_count']}\n" f" Words: {corpus['word_count']:,}\n" f" Estimated tokens: {corpus['estimated_tokens']:,}", file=sys.stderr, ) # Check context window fit (skip for deep-fix which uses subset) if not args.deep_fix: model_limit = CONTEXT_LIMITS.get(args.model, 1_000_000) if corpus["estimated_tokens"] > model_limit * 0.85: print( f"\nWARNING: Corpus ({corpus['estimated_tokens']:,} tokens) is " f">{85}% of {args.model} context window ({model_limit:,}).\n" f"Consider using --model {LARGE_CONTEXT_MODEL} or --focus to limit scope.", file=sys.stderr, ) # --- Full Automated Pipeline --- if args.full: out_path = run_full_diagnostic(args, corpus, args.model) return # --- Close-Read Pass (standalone) --- if args.close_read: engine = getattr(args, "close_read_engine", "claude") print(f"\n{'='*60}", file=sys.stderr) print(f"CLOSE-READ PASS (standalone) — engine: {engine}", file=sys.stderr) print(f"{'='*60}\n", file=sys.stderr) if args.dry_run or engine == "claude": # Save all 7 batch payloads for Opus sub-agent processing (or dry-run inspection) state_dir = _project_paths_or_exit(args.project).state_dir state_dir.mkdir(parents=True, exist_ok=True) for batch_num, (start, end) in enumerate(CLOSE_READ_BATCHES, 1): payload = format_close_read_payload(corpus, start, end, batch_num) dry_path = state_dir / f"close_read_batch{batch_num}_payload.txt" dry_path.write_text(payload, encoding="utf-8") print(f" Batch {batch_num} payload saved to: {dry_path.name}", file=sys.stderr) if engine == "claude": print( "\n Payloads saved for Opus sub-agent processing.\n" " Process each batch with Claude Opus, then merge results.", file=sys.stderr, ) return # engine == "gemini" — existing Gemini API path cr_brief = run_close_read(args, corpus, args.model) cr_stats = cr_brief.get("stats", {}) print(f"\n{'='*60}") print("SCRIPT DOCTOR — CLOSE-READ COMPLETE") print(f"{'='*60}") print(f"\nFindings: {cr_stats.get('total_findings', 0)}") print(f" P1 (must fix): {cr_stats.get('p1_count', 0)}") print(f" P2 (should fix): {cr_stats.get('p2_count', 0)}") print(f" P3 (consider): {cr_stats.get('p3_count', 0)}") print(f" Annotations: {cr_stats.get('total_annotations', 0)}") # Print P1 findings p1s = [f for f in cr_brief.get("findings", []) if f.get("severity") == "P1"] if p1s: print("\nP1 CLOSE-READ FINDINGS:") for f in p1s: sub = f.get("sub_dimension", "") print(f" {f['id']} [{sub}] {f.get('title', '')}") print( f"\nSaved to: " f"{_project_paths_or_exit(args.project).state_dir / 'script_doctor_close_read.json'}" ) print(f"{'='*60}") # Exit code: 0 = pass, 1 = P1 failures, 2 = P2 review p1_count = cr_stats.get("p1_count", 0) p2_count = cr_stats.get("p2_count", 0) if p1_count > 0: sys.exit(1) elif p2_count > 0: sys.exit(2) return # --- Parse Response (manual Gemini workflow) --- if args.parse_response: response_path = Path(args.parse_response) if not response_path.is_absolute(): # Try relative to project state dir, then cwd state_candidate = _project_paths_or_exit(args.project).state_dir / args.parse_response if state_candidate.exists(): response_path = state_candidate else: response_path = Path.cwd() / args.parse_response if not response_path.exists(): print( f"ERROR: Response file not found: {response_path}", file=sys.stderr, ) sys.exit(1) response_text = response_path.read_text(encoding="utf-8") parsed = parse_gemini_response(response_text) parse_type = args.parse_type if parse_type == "diagnostic": out_path = save_brief(parsed, args.project, pass_type="diagnostic") print_summary(parsed, pass_type="diagnostic") print(f"\nBrief saved to: {out_path}") print(f"(Parsed from manual response: {response_path.name})") elif parse_type == "verify": out_path = save_brief(parsed, args.project, pass_type="verify") print_summary(parsed, pass_type="verify") print(f"\nVerification saved to: {out_path}") elif parse_type == "deep-fix": finding_id = parsed.get("finding_id", "unknown") out_path = save_deep_fix(parsed, finding_id, args.project) print_deep_fix_summary(parsed) print(f"\nDeep fix saved to: {out_path}") return # --- Deep Fix --- if args.deep_fix: state_dir = _project_paths_or_exit(args.project).state_dir brief_path = state_dir / "script_doctor_brief.json" if not brief_path.exists(): print( "ERROR: No brief found. Run --diagnose first.", file=sys.stderr, ) sys.exit(1) brief = json.loads(brief_path.read_text(encoding="utf-8")) finding = None for f in brief.get("findings", []): if f.get("id") == args.deep_fix: finding = f break if finding is None: print( f"ERROR: Finding '{args.deep_fix}' not found in brief.\n" f"Available findings: {[f['id'] for f in brief.get('findings', [])]}", file=sys.stderr, ) sys.exit(1) payload = format_deep_fix_payload(finding, corpus, args.project) if args.dry_run: dry_path = ( state_dir / f"script_doctor_deepfix_{args.deep_fix}_payload.txt" ) dry_path.parent.mkdir(parents=True, exist_ok=True) dry_path.write_text(payload, encoding="utf-8") print(f"Dry run payload saved to: {dry_path}") return if args.deep_fix_engine == "gemini": # Gemini path (backward compat) genai = get_gemini_client() tracker = CostTracker(_project_paths_or_exit(args.project).project_root) response_text, usage, elapsed_ms = call_gemini(genai, payload, args.model) _log_cost(tracker, args.model, usage, elapsed_ms, f"Script doctor deep fix: {args.deep_fix}") fix_data = parse_gemini_response(response_text) out_path = save_deep_fix(fix_data, args.deep_fix, args.project) print_deep_fix_summary(fix_data) print(f"\nDeep fix saved to: {out_path}") else: # Claude path (default): save payload for Claude Code reasoning state_dir.mkdir(parents=True, exist_ok=True) payload_path = state_dir / f"deep_fix_{args.deep_fix}_payload.txt" payload_path.write_text(payload, encoding="utf-8") print(f"Deep fix payload saved to: {payload_path}", file=sys.stderr) print(f"Use Claude Code to reason about this fix and generate deep_fix_{args.deep_fix}.json", file=sys.stderr) return # --- Extract Annotations --- if args.to_annotations: brief_path = _project_paths_or_exit(args.project).state_dir / "script_doctor_brief.json" if not brief_path.exists(): print( "ERROR: No brief found. Run --diagnose first.", file=sys.stderr, ) sys.exit(1) out_path = extract_annotations(brief_path, args.project) ann = json.loads(out_path.read_text(encoding="utf-8")) stats = ann["stats"] print(f"\n{'='*60}") print("SCRIPT DOCTOR — ANNOTATIONS EXTRACTED") print(f"{'='*60}") print(f"\nBrief version: {ann.get('brief_version', '?')}") print(f"Total annotations: {stats['total_annotations']}") print(f" REWRITE: {stats['rewrite_count']}") print(f" FLAG: {stats['flag_count']}") print(f" DELETE: {stats['delete_count']}") print(f" Episodes affected: {stats['episodes_affected']}") if stats.get("quotes_not_found", 0) > 0: print(f" Quotes not located: {stats['quotes_not_found']} (line 0, legacy)") print(f"\nSaved to: {out_path}") print("\nNext step:") print(f" /revise {args.project} {out_path.name}") print(f"{'='*60}") return # --- Bundle Preview --- if args.bundle: payload = format_corpus_payload(corpus, focus) preview_path = _project_paths_or_exit(args.project).state_dir / "script_doctor_payload_preview.txt" preview_path.parent.mkdir(parents=True, exist_ok=True) preview_path.write_text(payload, encoding="utf-8") print(f"\nPayload preview saved to: {preview_path}") print(f"Payload length: {len(payload):,} chars, ~{len(payload.split()):,} words") return # --- Verification Pass --- if args.verify: brief_path = _project_paths_or_exit(args.project).state_dir / "script_doctor_brief.json" if not brief_path.exists(): print( "ERROR: No existing brief found. Run --diagnose first.", file=sys.stderr, ) sys.exit(1) payload = format_verify_payload(corpus, brief_path) if args.dry_run: dry_path = _project_paths_or_exit(args.project).state_dir / "script_doctor_verify_payload.txt" dry_path.parent.mkdir(parents=True, exist_ok=True) dry_path.write_text(payload, encoding="utf-8") print(f"Dry run payload saved to: {dry_path}") return genai = get_gemini_client() tracker = CostTracker(_project_paths_or_exit(args.project).project_root) response_text, usage, elapsed_ms = call_gemini(genai, payload, args.model) _log_cost(tracker, args.model, usage, elapsed_ms, "Script doctor verification pass") brief = parse_gemini_response(response_text) out_path = save_brief(brief, args.project, pass_type="verify") print_summary(brief, pass_type="verify") print(f"\nVerification saved to: {out_path}") return # --- Diagnostic Pass --- if args.diagnose: # Handle --focus transitions separately (uses transition prompt) if focus and "transitions" in focus: payload = format_transition_payload(corpus) if args.dry_run: dry_path = _project_paths_or_exit(args.project).state_dir / "script_doctor_transitions_payload.txt" dry_path.parent.mkdir(parents=True, exist_ok=True) dry_path.write_text(payload, encoding="utf-8") print(f"Dry run payload saved to: {dry_path}") return genai = get_gemini_client() tracker = CostTracker(_project_paths_or_exit(args.project).project_root) response_text, usage, elapsed_ms = call_gemini(genai, payload, args.model) _log_cost(tracker, args.model, usage, elapsed_ms, "Script doctor transitions pass") brief = parse_gemini_response(response_text) out_path = save_brief(brief, args.project, pass_type="diagnostic") print_summary(brief, pass_type="diagnostic") # Print transition-specific stats t_stats = brief.get("transition_stats", {}) if t_stats: print("\nTransition stats:") print(f" THEREFORE: {t_stats.get('therefore_count', '?')}") print(f" BUT: {t_stats.get('but_count', '?')}") print(f" AND THEN: {t_stats.get('and_then_count', '?')}") print(f"\nBrief saved to: {out_path}") return # Standard series-level diagnostic (with optional focus filter) # Filter out 'transitions' from focus if mixed with series dimensions series_focus = [f for f in (focus or []) if f != "transitions"] or None payload = format_corpus_payload(corpus, series_focus) if args.dry_run: dry_path = _project_paths_or_exit(args.project).state_dir / "script_doctor_diagnostic_payload.txt" dry_path.parent.mkdir(parents=True, exist_ok=True) dry_path.write_text(payload, encoding="utf-8") print(f"Dry run payload saved to: {dry_path}") return genai = get_gemini_client() tracker = CostTracker(_project_paths_or_exit(args.project).project_root) response_text, usage, elapsed_ms = call_gemini(genai, payload, args.model) _log_cost(tracker, args.model, usage, elapsed_ms, "Script doctor diagnostic pass") brief = parse_gemini_response(response_text) out_path = save_brief(brief, args.project, pass_type="diagnostic") print_summary(brief, pass_type="diagnostic") print(f"\nBrief saved to: {out_path}") return # Default: show usage parser.print_help() if __name__ == "__main__": main()