#!/usr/bin/env python3 """Boardability Doctor: canon-aware script readiness for storyboards.""" from __future__ import annotations import argparse import json import logging import sys from dataclasses import asdict, dataclass from pathlib import Path from typing import Any from recoil.core.atomic_write import atomic_write_json from recoil.core.claude_cli import claude_cli_call, claude_transport from recoil.core.paths import ProjectPaths logger = logging.getLogger(__name__) @dataclass class BoardabilityFinding: id: str category: str severity: str line: int quoted_text: str why: str suggested_fix: str suggested_mechanism: str CATEGORIES = { "under_specified_staging", "unrenderable_authorial", "mechanic_legibility", "spatial_ambiguity", "reference_inconsistency", } SEVERITIES = {"P1", "P2", "P3"} # Intentional boardability subset of the SYNTHESIS D4 mechanism enum. MECHANISMS = { "patch_script", "patch_board_prompt", "patch_blocking", "patch_plan", "patch_ref", "edit_bible", "ask_clarify", } FINDING_FIELDS = ( "id", "category", "severity", "line", "quoted_text", "why", "suggested_fix", "suggested_mechanism", ) SEVERITY_RUBRIC = """Severity rubric (PINNED — the `--gate` depends on P1, so this is concrete and prompt-tested): - **P1** = the script as written will produce a WRONG or INCONSISTENT board, OR asks the board to depict something un-renderable (a thought/impossibility). Blocks boarding. - **P2** = ambiguity that risks cross-board drift but a competent boarder can resolve. - **P3** = minor / polish.""" def resolve_inputs(project: str, episode: int) -> tuple[Path, str, list[Path], str]: """Resolve script and canon inputs through the ProjectPaths SSOT.""" pp = ProjectPaths.for_project(project) script_path = pp.episodes_dir / f"ep_{episode:03d}.md" if not script_path.exists(): raise FileNotFoundError(script_path) script_text = script_path.read_text(encoding="utf-8") bible_paths = sorted(pp.bible_dir.glob("*.md")) if not bible_paths: logger.warning("No bible markdown files found under %s", pp.bible_dir) return script_path, script_text, [], "" bible_parts = [] for path in bible_paths: bible_parts.append(f"## {path.name}\n\n{path.read_text(encoding='utf-8')}") return script_path, script_text, bible_paths, "\n\n".join(bible_parts) def build_system_prompt(bible_text: str) -> str: """Build the boardability lens prompt with canon embedded.""" return f"""You are Boardability Doctor, a script-readiness reviewer for storyboards. Flag only issues that affect whether the SCRIPT can be boarded clearly and consistently. Return strict JSON only. Do not wrap the response in Markdown. CATEGORIES (the boardability lens): 1. under_specified_staging Definition: vague action that cannot be rendered without inventing staging. Positive example to flag: "She does something with the panel." Negative example not to flag: "She flips the red breaker switch on the left edge of the panel." 2. unrenderable_authorial Definition: a thought, impossibility, or judgment a storyboard cannot show. Positive example to flag: "Her shoulders are wider than the pod should allow." Negative example not to flag: "Her shoulders press against both sides of the pod." 3. mechanic_legibility Definition: a load-bearing causal mechanic that will not read in one board. Positive example to flag: "She holds her breath and the counter-freezes." Negative example not to flag: "She holds her breath; frost blooms over the counter display, stopping the digits." 4. spatial_ambiguity Definition: unclear geometry or motion the renderer cannot place. Positive example to flag: "She grabs the cable as the platform rises to meet her." Negative example not to flag: "She grabs the black tether hanging from the ceiling hook as the floor lift rises beneath her boots." 5. reference_inconsistency Definition: an element drifts across boards without an anchor. Positive example to flag: "The flare is suddenly back in her hand after she dropped it." Negative example not to flag: "She retrieves the flare from the deck before raising it again." {SEVERITY_RUBRIC} Allowed suggested_mechanism values: {", ".join(sorted(MECHANISMS))} CANON (do NOT flag intentional choices established here): {bible_text} Worked canon rule: A `JADE (V.O.)` open is canon if the bible's signature-tell/voice section calls for it — DO NOT flag it as internal narration. """ def build_user_prompt(script_text: str) -> str: """Build a line-numbered script prompt with the strict JSON contract.""" numbered = "\n".join( f"{line_no:04d}: {line}" for line_no, line in enumerate(script_text.splitlines(), start=1) ) contract = { "findings": [ { "id": "BD001", "category": "under_specified_staging", "severity": "P2", "line": 1, "quoted_text": "exact text from that line", "why": "why this blocks or risks boarding", "suggested_fix": "concrete script or canon fix", "suggested_mechanism": "patch_script", } ], "summary": "brief summary string", } return ( "Review this line-numbered script for boardability only.\n\n" "SCRIPT:\n" f"{numbered}\n\n" "Return ONLY strict JSON matching this contract. Use an empty findings list if clean:\n" f"{json.dumps(contract, indent=2)}" ) def _strip_code_fence(raw_text: str) -> str: text = raw_text.strip() if not text.startswith("```"): return text lines = text.splitlines() if lines and lines[0].strip().startswith("```"): lines = lines[1:] if lines and lines[-1].strip() == "```": lines = lines[:-1] return "\n".join(lines).strip() def parse_response(raw_text: str) -> dict[str, Any]: """Parse Claude JSON and fail loudly on malformed top-level shape.""" try: parsed = json.loads(_strip_code_fence(raw_text)) except json.JSONDecodeError as exc: raise ValueError(f"Invalid JSON response: {exc}") from exc if not isinstance(parsed, dict): raise ValueError(f"Response must be a JSON object, got {type(parsed).__name__}") if not isinstance(parsed.get("summary"), str): raise ValueError("Response field 'summary' must be a str") if not isinstance(parsed.get("findings"), list): raise ValueError("Response field 'findings' must be a list") return parsed def validate_findings(raw_findings: list[dict[str, Any]]) -> list[BoardabilityFinding]: """Validate raw findings against the boardability schema.""" findings: list[BoardabilityFinding] = [] for idx, item in enumerate(raw_findings): if not isinstance(item, dict): raise ValueError(f"findings[{idx}] must be a dict, got {type(item).__name__}") for field in FINDING_FIELDS: if field not in item: raise ValueError(f"findings[{idx}] missing field {field!r}") category = item["category"] if category not in CATEGORIES: raise ValueError(f"findings[{idx}].category invalid value {category!r}") severity = item["severity"] if severity not in SEVERITIES: raise ValueError(f"findings[{idx}].severity invalid value {severity!r}") mechanism = item["suggested_mechanism"] if mechanism not in MECHANISMS: raise ValueError( f"findings[{idx}].suggested_mechanism invalid value {mechanism!r}" ) line = item["line"] if not isinstance(line, int) or isinstance(line, bool): raise ValueError(f"findings[{idx}].line must be an int, got {line!r}") findings.append(BoardabilityFinding(**{field: item[field] for field in FINDING_FIELDS})) return findings def _brief_stats(findings: list[BoardabilityFinding]) -> dict[str, Any]: by_category = {category: 0 for category in sorted(CATEGORIES)} for finding in findings: by_category[finding.category] += 1 return { "total": len(findings), "p1": sum(1 for finding in findings if finding.severity == "P1"), "p2": sum(1 for finding in findings if finding.severity == "P2"), "p3": sum(1 for finding in findings if finding.severity == "P3"), "by_category": by_category, } def run_boardability(project: str, episode: int, model: str | None = None) -> dict[str, Any]: """Run the boardability pass and return the brief dict.""" if not claude_transport() == "cli": raise RuntimeError("boardability_doctor requires claude_transport() == 'cli'") transport = claude_transport() _script_path, script_text, bible_paths, bible_text = resolve_inputs(project, episode) system_prompt = build_system_prompt(bible_text) user_prompt = build_user_prompt(script_text) out = claude_cli_call(user_prompt, system_prompt=system_prompt, model=model) data = parse_response(out) findings = validate_findings(data["findings"]) return { "version": "1.0", "tool": "boardability_doctor", "project": project, "episode": episode, "transport": transport, "model": model, "canon_sources": [path.name for path in bible_paths], "summary": data["summary"], "findings": [asdict(finding) for finding in findings], "stats": _brief_stats(findings), } def write_brief(brief: dict[str, Any], project: str, episode: int) -> Path: """Write the boardability brief atomically and return its path.""" pp = ProjectPaths.for_project(project) target = ( pp.project_root / "_pipeline" / "state" / f"boardability_brief_ep_{episode:03d}.json" ) target.parent.mkdir(parents=True, exist_ok=True) atomic_write_json(target, brief, indent=2) return target def _print_human_summary(brief: dict[str, Any], path: Path, gate: bool) -> None: stats = brief["stats"] print( f"Boardability Doctor ep_{brief['episode']:03d}: " f"{stats['total']} findings ({stats['p1']} P1, {stats['p2']} P2, {stats['p3']} P3)" ) print(f"Summary: {brief['summary']}") if gate and stats["p1"]: print("Blocking P1 findings:") for finding in brief["findings"]: if finding["severity"] == "P1": print( f"- {finding['id']} line {finding['line']} " f"[{finding['category']}]: {finding['quoted_text']}" ) print(f"Wrote brief: {path}") def main(argv: list[str]) -> int: parser = argparse.ArgumentParser(description="Run the canon-aware Boardability Doctor.") parser.add_argument("--project", required=True) parser.add_argument("--episode", required=True, type=int) parser.add_argument("--model") parser.add_argument("--gate", action="store_true") parser.add_argument("--json", action="store_true", dest="json_out") args = parser.parse_args(argv) brief = run_boardability(args.project, args.episode, model=args.model) path = write_brief(brief, args.project, args.episode) exit_code = 2 if args.gate and brief["stats"]["p1"] else 0 if args.json_out: print(json.dumps(brief, indent=2, ensure_ascii=False)) else: _print_human_summary(brief, path, args.gate) return exit_code if __name__ == "__main__": sys.exit(main(sys.argv[1:]))