feat: Built ScoreRunner with 5-dimension LLM-as-judge scoring rubric, C…

- "backend/pipeline/quality/scorer.py"
- "backend/pipeline/quality/__main__.py"
- "backend/pipeline/quality/fixtures/sample_moments.json"
- "backend/pipeline/quality/fixtures/__init__.py"

GSD-Task: S02/T01

backend/pipeline/quality/__main__.py

@@ -1,16 +1,22 @@
-"""FYN-LLM fitness test suite.
+"""FYN-LLM quality assurance toolkit.
 
-Run with: python -m pipeline.quality fitness
+Subcommands:
+  fitness  — Run LLM fitness tests across four categories
+  score    — Score a Stage 5 technique page across 5 quality dimensions
+
+Run with: python -m pipeline.quality <command>
 """
 from __future__ import annotations
 
 import argparse
+import json
 import sys
 
 from config import get_settings
 from pipeline.llm_client import LLMClient
 
 from .fitness import FitnessRunner
+from .scorer import ScoreRunner
 
 
 def main() -> int:
@@ -23,6 +29,29 @@ def main() -> int:
     # -- fitness subcommand --
     sub.add_parser("fitness", help="Run LLM fitness tests across four categories")
 
+    # -- score subcommand --
+    score_parser = sub.add_parser(
+        "score",
+        help="Score a Stage 5 technique page across 5 quality dimensions",
+    )
+    source_group = score_parser.add_mutually_exclusive_group(required=True)
+    source_group.add_argument(
+        "--file",
+        type=str,
+        help="Path to a moments JSON file (creator_name, moments array)",
+    )
+    source_group.add_argument(
+        "--slug",
+        type=str,
+        help="Technique slug to load from the database",
+    )
+    score_parser.add_argument(
+        "--voice-level",
+        type=float,
+        default=None,
+        help="Voice preservation dial (0.0=clinical, 1.0=maximum voice). Triggers re-synthesis before scoring.",
+    )
+
     args = parser.parse_args()
 
     if args.command is None:
@@ -35,6 +64,66 @@ def main() -> int:
         runner = FitnessRunner(client)
         return runner.run_all()
 
+    if args.command == "score":
+        return _run_score(args)
+
+    return 0
+
+
+def _run_score(args: argparse.Namespace) -> int:
+    """Execute the score subcommand."""
+    # -- Load source data --
+    if args.slug:
+        print("DB loading not yet implemented", file=sys.stderr)
+        return 1
+
+    try:
+        with open(args.file) as f:
+            data = json.load(f)
+    except FileNotFoundError:
+        print(f"File not found: {args.file}", file=sys.stderr)
+        return 1
+    except json.JSONDecodeError as exc:
+        print(f"Invalid JSON in {args.file}: {exc}", file=sys.stderr)
+        return 1
+
+    moments = data.get("moments", [])
+    creator_name = data.get("creator_name", "Unknown")
+
+    if not moments:
+        print("No moments found in input file", file=sys.stderr)
+        return 1
+
+    # -- Build page stub from moments for scoring --
+    # When --voice-level is set, T02 will re-synthesize. For now, build a
+    # minimal page representation from the moments so the scorer has
+    # something to evaluate.
+    page_json = {
+        "title": f"{creator_name} — Technique Page",
+        "creator_name": creator_name,
+        "summary": f"Technique page synthesized from {len(moments)} key moments.",
+        "body_sections": [
+            {
+                "heading": m.get("topic_tags", ["Technique"])[0] if m.get("topic_tags") else "Technique",
+                "content": m.get("summary", "") + "\n\n" + m.get("transcript_excerpt", ""),
+            }
+            for m in moments
+        ],
+    }
+
+    settings = get_settings()
+    client = LLMClient(settings)
+    runner = ScoreRunner(client)
+
+    print(f"\nScoring page for '{creator_name}' ({len(moments)} moments)...")
+
+    result = runner.score_page(page_json, moments)
+
+    if result.error:
+        runner.print_report(result)
+        return 1
+
+    runner.print_report(result)
     return 0
 
 

backend/pipeline/quality/fixtures/sample_moments.json

@@ -0,0 +1,54 @@
+{
+  "creator_name": "KOAN Sound",
+  "topic_category": "Sound design",
+  "moments": [
+    {
+      "summary": "Layering snare transients by combining a high-frequency click from a Popcorn Snare with a mid-body from a pitched-down 808 rim shot, blending at -6dB relative offset.",
+      "transcript_excerpt": "So what I'll do is take the Popcorn Snare — that's got this really sharp click at like 4k — and then I layer underneath it a rim shot pitched down maybe 3 semitones. You blend those together and suddenly you've got this snare that cuts through everything but still has weight.",
+      "topic_tags": ["snare layering", "transient design", "sample stacking"],
+      "topic_category": "Sound design",
+      "start_time": 124.5,
+      "end_time": 158.2
+    },
+    {
+      "summary": "Using Serum's noise oscillator with the 'Analog_Crackle' wavetable at 12% mix to add organic texture to bass patches, followed by OTT at 30% depth for glue.",
+      "transcript_excerpt": "One trick I always come back to is Serum's noise osc with Analog_Crackle. You don't want it loud — like 12 percent mix — just enough that the bass feels alive. Then slap OTT on there at maybe 30 percent depth and it glues the whole thing together without squashing it.",
+      "topic_tags": ["bass design", "Serum", "OTT", "texture"],
+      "topic_category": "Sound design",
+      "start_time": 203.1,
+      "end_time": 241.7
+    },
+    {
+      "summary": "Resampling technique: bounce a bass patch to audio, chop the best 2 bars, then re-pitch in Simpler with warp off for tighter timing and consistent tone.",
+      "transcript_excerpt": "I'll resample everything. Bounce it down, find the two bars that sound best, throw it in Simpler with warp completely off. Now you've got this tight, consistent thing where every hit is exactly the same energy. The pitch tracking is way more predictable too.",
+      "topic_tags": ["resampling", "Ableton", "Simpler", "bass production"],
+      "topic_category": "Sound design",
+      "start_time": 312.0,
+      "end_time": 349.8
+    },
+    {
+      "summary": "Parallel compression chain for drums using Ableton's Drum Buss at 40% drive into a return track with Valhalla Room at 1.2s decay, mixed at -12dB.",
+      "transcript_excerpt": "The parallel chain is dead simple — Drum Buss, crank the drive to about 40 percent, send that to a return with Valhalla Room. Keep the decay short, like 1.2 seconds. Mix it in at minus 12 and your drums just... breathe. They've got this room sound without getting washy.",
+      "topic_tags": ["parallel compression", "drum processing", "Valhalla Room", "Drum Buss"],
+      "topic_category": "Sound design",
+      "start_time": 421.3,
+      "end_time": 462.1
+    },
+    {
+      "summary": "Frequency-specific sidechain using Trackspacer plugin instead of volume ducking, targeting only 100-300Hz so the bass ducks under the kick without losing high-end presence.",
+      "transcript_excerpt": "Everyone does volume sidechain but honestly Trackspacer changed everything for me. You set it to only affect 100 to 300 Hz so when the kick hits, the bass ducks just in that low-mid range. The top end of the bass stays right there — you keep all the character and harmonics, you just clear the mud.",
+      "topic_tags": ["sidechaining", "Trackspacer", "frequency ducking", "mixing"],
+      "topic_category": "Sound design",
+      "start_time": 498.7,
+      "end_time": 534.2
+    },
+    {
+      "summary": "Using Ableton's Utility plugin to check mono compatibility at every stage, specifically toggling mono on the sub bus to catch phase cancellation from layered bass patches.",
+      "transcript_excerpt": "I'm almost paranoid about mono. I've got Utility on the sub bus and I'm flipping to mono constantly. If your layered bass sounds thin in mono you've got phase issues — doesn't matter how fat it sounds in stereo, it'll collapse on a club system.",
+      "topic_tags": ["mono compatibility", "phase checking", "club mixing", "Utility"],
+      "topic_category": "Sound design",
+      "start_time": 567.0,
+      "end_time": 598.4
+    }
+  ]
+}

backend/pipeline/quality/scorer.py

@@ -0,0 +1,263 @@
+"""Stage 5 quality scorer — LLM-as-judge evaluation across 5 dimensions.
+
+Evaluates a synthesized technique page against source moments on:
+1. Structural quality — section naming, count, paragraph depth
+2. Content specificity — concrete details vs vague generalities
+3. Voice preservation — direct quotes, attributed opinions, personality
+4. Readability / flow — synthesis quality, logical ordering, no redundancy
+5. Factual fidelity — no hallucinated specifics, grounded in source moments
+
+Run via: python -m pipeline.quality score --file <path>
+"""
+from __future__ import annotations
+
+import json
+import logging
+import time
+from dataclasses import dataclass, field
+
+import openai
+from pydantic import BaseModel
+
+from pipeline.llm_client import LLMClient
+
+logger = logging.getLogger(__name__)
+
+
+# ── Scoring rubric (hardcoded for iteration speed) ───────────────────────────
+
+SCORING_RUBRIC = """\
+You are an expert evaluator of synthesized technique articles for music production education.
+
+You will be given:
+1. A synthesized technique page (JSON with title, summary, body_sections)
+2. The source key moments (transcript excerpts, summaries, tags) used to create it
+
+Evaluate the page across these 5 dimensions, scoring each 0.0 to 1.0:
+
+**structural** — Section naming and organization
+- 0.9-1.0: Well-named specific sections (not generic "Overview"/"Tips"), appropriate count (3-6), 2-5 paragraphs per section
+- 0.5-0.7: Acceptable structure but some generic section names or uneven depth
+- 0.0-0.3: Poor structure — too few/many sections, generic names, single-paragraph sections
+
+**content_specificity** — Concrete technical details
+- 0.9-1.0: Rich in frequencies (Hz), time values (ms), ratios, plugin names, specific settings, dB values
+- 0.5-0.7: Some specific details but padded with vague statements ("adjust to taste", "experiment with settings")
+- 0.0-0.3: Mostly vague generalities with few concrete values from the source material
+
+**voice_preservation** — Creator's authentic voice
+- 0.9-1.0: Direct quotes preserved, opinions attributed to creator by name, personality and strong views retained
+- 0.5-0.7: Some paraphrased references to creator's views but few direct quotes
+- 0.0-0.3: Encyclopedia style — creator's voice completely smoothed out, no attribution
+
+**readability** — Synthesis quality and flow
+- 0.9-1.0: Reads as a cohesive article, related info merged, logical flow, no redundancy or contradiction
+- 0.5-0.7: Generally readable but some awkward transitions or minor repetition
+- 0.0-0.3: Feels like concatenated bullet points, disjointed, redundant passages
+
+**factual_fidelity** — Grounded in source material
+- 0.9-1.0: Every claim traceable to source moments, no invented plugin names/settings/techniques
+- 0.5-0.7: Mostly grounded but 1-2 details seem embellished or not directly from sources
+- 0.0-0.3: Contains hallucinated specifics — plugin names, settings, or techniques not in sources
+
+Return ONLY a JSON object with this exact structure:
+{
+  "structural": <float 0.0-1.0>,
+  "content_specificity": <float 0.0-1.0>,
+  "voice_preservation": <float 0.0-1.0>,
+  "readability": <float 0.0-1.0>,
+  "factual_fidelity": <float 0.0-1.0>,
+  "justifications": {
+    "structural": "<1-2 sentence justification>",
+    "content_specificity": "<1-2 sentence justification>",
+    "voice_preservation": "<1-2 sentence justification>",
+    "readability": "<1-2 sentence justification>",
+    "factual_fidelity": "<1-2 sentence justification>"
+  }
+}
+"""
+
+DIMENSIONS = [
+    "structural",
+    "content_specificity",
+    "voice_preservation",
+    "readability",
+    "factual_fidelity",
+]
+
+
+# ── Result type ──────────────────────────────────────────────────────────────
+
+@dataclass
+class ScoreResult:
+    """Outcome of scoring a technique page across 5 quality dimensions."""
+
+    structural: float = 0.0
+    content_specificity: float = 0.0
+    voice_preservation: float = 0.0
+    readability: float = 0.0
+    factual_fidelity: float = 0.0
+    composite: float = 0.0
+    justifications: dict[str, str] = field(default_factory=dict)
+    elapsed_seconds: float = 0.0
+    error: str | None = None
+
+
+# ── Runner ───────────────────────────────────────────────────────────────────
+
+class ScoreRunner:
+    """Scores a Stage 5 technique page using LLM-as-judge evaluation."""
+
+    def __init__(self, client: LLMClient) -> None:
+        self.client = client
+
+    def score_page(
+        self,
+        page_json: dict,
+        moments: list[dict],
+    ) -> ScoreResult:
+        """Evaluate a technique page against source moments.
+
+        Parameters
+        ----------
+        page_json:
+            Synthesized page dict (title, summary, body_sections).
+        moments:
+            Source key moments with transcript_excerpt, summary, etc.
+
+        Returns
+        -------
+        ScoreResult with per-dimension scores and justifications.
+        """
+        # Build the user prompt with the page and source moments
+        user_prompt = (
+            "## Synthesized Technique Page\n\n"
+            f"```json\n{json.dumps(page_json, indent=2)}\n```\n\n"
+            "## Source Key Moments\n\n"
+            f"```json\n{json.dumps(moments, indent=2)}\n```\n\n"
+            "Score this page across all 5 dimensions."
+        )
+
+        t0 = time.monotonic()
+        try:
+            resp = self.client.complete(
+                system_prompt=SCORING_RUBRIC,
+                user_prompt=user_prompt,
+                response_model=BaseModel,  # triggers JSON mode
+                modality="chat",
+            )
+            elapsed = round(time.monotonic() - t0, 2)
+        except (openai.APIConnectionError, openai.APITimeoutError) as exc:
+            elapsed = round(time.monotonic() - t0, 2)
+            url = self.client.settings.llm_api_url
+            fallback = self.client.settings.llm_fallback_url
+            return ScoreResult(
+                elapsed_seconds=elapsed,
+                error=(
+                    f"Cannot reach LLM endpoint at {url} (fallback {fallback}). "
+                    f"Error: {exc}"
+                ),
+            )
+
+        # Parse the LLM judge response
+        raw_text = str(resp).strip()
+        try:
+            parsed = json.loads(raw_text)
+        except json.JSONDecodeError:
+            logger.error("Malformed judge response (not JSON): %.300s", raw_text)
+            return ScoreResult(
+                elapsed_seconds=elapsed,
+                error=f"Malformed judge response (not valid JSON). Raw excerpt: {raw_text[:200]}",
+            )
+
+        return self._parse_scores(parsed, elapsed)
+
+    def _parse_scores(self, parsed: dict, elapsed: float) -> ScoreResult:
+        """Extract and validate scores from parsed JSON response."""
+        scores: dict[str, float] = {}
+        justifications: dict[str, str] = {}
+
+        raw_justifications = parsed.get("justifications", {})
+        if not isinstance(raw_justifications, dict):
+            raw_justifications = {}
+
+        for dim in DIMENSIONS:
+            raw = parsed.get(dim)
+            if raw is None:
+                logger.warning("Missing dimension '%s' in judge response", dim)
+                scores[dim] = 0.0
+                justifications[dim] = "(missing from judge response)"
+                continue
+
+            try:
+                val = float(raw)
+                scores[dim] = max(0.0, min(1.0, val))  # clamp
+            except (TypeError, ValueError):
+                logger.warning("Invalid value for '%s': %r", dim, raw)
+                scores[dim] = 0.0
+                justifications[dim] = f"(invalid value: {raw!r})"
+                continue
+
+            justifications[dim] = str(raw_justifications.get(dim, ""))
+
+        composite = sum(scores.values()) / len(DIMENSIONS)
+
+        return ScoreResult(
+            structural=scores["structural"],
+            content_specificity=scores["content_specificity"],
+            voice_preservation=scores["voice_preservation"],
+            readability=scores["readability"],
+            factual_fidelity=scores["factual_fidelity"],
+            composite=round(composite, 3),
+            justifications=justifications,
+            elapsed_seconds=elapsed,
+        )
+
+    def print_report(self, result: ScoreResult) -> None:
+        """Print a formatted scoring report to stdout."""
+        print("\n" + "=" * 60)
+        print("  STAGE 5 QUALITY SCORE REPORT")
+        print("=" * 60)
+
+        if result.error:
+            print(f"\n  ✗ Error: {result.error}\n")
+            print("=" * 60 + "\n")
+            return
+
+        for dim in DIMENSIONS:
+            score = getattr(result, dim)
+            bar = self._score_bar(score)
+            justification = result.justifications.get(dim, "")
+            print(f"\n  {dim.replace('_', ' ').title()}")
+            print(f"    Score: {score:.2f}  {bar}")
+            if justification:
+                # Wrap justification at ~60 chars
+                for line in self._wrap(justification, 56):
+                    print(f"    {line}")
+
+        print("\n" + "-" * 60)
+        print(f"  Composite: {result.composite:.3f}")
+        print(f"  Time: {result.elapsed_seconds}s")
+        print("=" * 60 + "\n")
+
+    @staticmethod
+    def _score_bar(score: float, width: int = 20) -> str:
+        """Render a visual bar for a 0-1 score."""
+        filled = int(score * width)
+        return "█" * filled + "░" * (width - filled)
+
+    @staticmethod
+    def _wrap(text: str, width: int) -> list[str]:
+        """Simple word wrap."""
+        words = text.split()
+        lines: list[str] = []
+        current = ""
+        for word in words:
+            if current and len(current) + len(word) + 1 > width:
+                lines.append(current)
+                current = word
+            else:
+                current = f"{current} {word}" if current else word
+        if current:
+            lines.append(current)
+        return lines