VD10
/

PatchJudge

code-evaluation

swe-bench

llm-judge

code-quality

merge-score

Model card Files Files and versions

xet

Community

PatchJudge / patchjudge /validation.py

VD10

Upload patchjudge/validation.py with huggingface_hub

d73c869 verified 7 months ago

Raw History Blame Contribute Delete

21.1 kB

"""Validation module for PatchJudge.
Validates that PatchJudge scores correlate with actual code quality:
1. METR alignment: ~50% of test-passing patches should score below 50
2. Known-bad pattern detection: deliberately bad patches should score low
3. Score distribution analysis
4. Resolved vs unresolved separation
"""

import json
import logging
import statistics
from collections import defaultdict
from typing import Optional

from patchjudge.models import (
    PatchExample, PatchFeatures, JudgeResult, ValidationResult
)

logger = logging.getLogger(__name__)


# ============================================================================
# Known-Bad Patch Generator
# ============================================================================

class KnownBadPatchGenerator:
    """Generate deliberately bad patches that pass tests but are low quality."""
    
    @staticmethod
    def generate_all(gold_examples: list[dict]) -> list[PatchExample]:
        """Generate known-bad variants for a set of gold examples.
        
        Args:
            gold_examples: List of dicts with keys:
                instance_id, repo, problem_statement, gold_patch, base_commit
                
        Returns:
            List of PatchExample with known-bad patches.
        """
        bad_patches = []
        generator = KnownBadPatchGenerator()
        
        for ex in gold_examples[:50]:  # Generate from up to 50 examples
            variants = generator._generate_variants(ex)
            bad_patches.extend(variants)
        
        logger.info(f"Generated {len(bad_patches)} known-bad patch variants")
        return bad_patches
    
    def _generate_variants(self, ex: dict) -> list[PatchExample]:
        """Generate known-bad variants of a gold patch."""
        variants = []
        gold = ex.get("gold_patch", "")
        
        if not gold:
            return variants
        
        # Variant 1: Hardcoded return values
        hardcoded = self._make_hardcoded_variant(ex)
        if hardcoded:
            variants.append(hardcoded)
        
        # Variant 2: Pass statement (minimal no-op)
        noop = self._make_noop_variant(ex)
        if noop:
            variants.append(noop)
        
        # Variant 3: Comment-only patch
        comment_only = self._make_comment_only_variant(ex)
        if comment_only:
            variants.append(comment_only)
        
        # Variant 4: Overly broad try/except
        broad_except = self._make_broad_except_variant(ex)
        if broad_except:
            variants.append(broad_except)
        
        # Variant 5: Test-disabling patch  
        test_disable = self._make_test_disable_variant(ex)
        if test_disable:
            variants.append(test_disable)
        
        return variants
    
    def _make_hardcoded_variant(self, ex: dict) -> Optional[PatchExample]:
        """Create a patch with hardcoded return values."""
        gold = ex["gold_patch"]
        lines = gold.split('\n')
        
        # Find added lines with return statements and hardcode them
        new_lines = []
        modified = False
        for line in lines:
            if line.startswith('+') and not line.startswith('+++'):
                content = line[1:]
                if 'return' in content and not modified:
                    # Replace with hardcoded value
                    indent = len(content) - len(content.lstrip())
                    new_lines.append('+' + ' ' * indent + 'return True  # HARDCODED')
                    modified = True
                    continue
            new_lines.append(line)
        
        if not modified:
            return None
        
        return PatchExample(
            instance_id=ex["instance_id"] + "__hardcoded",
            repo=ex["repo"],
            problem_statement=ex["problem_statement"],
            gold_patch=ex["gold_patch"],
            agent_patch='\n'.join(new_lines),
            agent_name="known-bad:hardcoded",
            test_passed=True,  # Assume it passes (test oracle weakness)
            base_commit=ex.get("base_commit", ""),
            difficulty=ex.get("difficulty", ""),
        )
    
    def _make_noop_variant(self, ex: dict) -> Optional[PatchExample]:
        """Create a minimal no-op patch (just adds 'pass')."""
        gold = ex["gold_patch"]
        lines = gold.split('\n')
        
        # Find the first hunk and replace all added lines with 'pass'
        new_lines = []
        in_hunk = False
        added_pass = False
        
        for line in lines:
            if line.startswith('@@'):
                in_hunk = True
                new_lines.append(line)
                continue
            
            if in_hunk:
                if line.startswith('+') and not line.startswith('+++'):
                    if not added_pass:
                        content = line[1:]
                        indent = len(content) - len(content.lstrip())
                        new_lines.append('+' + ' ' * indent + 'pass  # TODO: implement')
                        added_pass = True
                    # Skip other added lines
                    continue
                elif line.startswith('-') and not line.startswith('---'):
                    new_lines.append(line)
                else:
                    new_lines.append(line)
            else:
                new_lines.append(line)
        
        if not added_pass:
            return None
        
        return PatchExample(
            instance_id=ex["instance_id"] + "__noop",
            repo=ex["repo"],
            problem_statement=ex["problem_statement"],
            gold_patch=ex["gold_patch"],
            agent_patch='\n'.join(new_lines),
            agent_name="known-bad:noop",
            test_passed=False,
            base_commit=ex.get("base_commit", ""),
            difficulty=ex.get("difficulty", ""),
        )
    
    def _make_comment_only_variant(self, ex: dict) -> Optional[PatchExample]:
        """Create a patch that only adds comments, no real code changes."""
        gold = ex["gold_patch"]
        lines = gold.split('\n')
        
        new_lines = []
        modified = False
        
        for line in lines:
            if line.startswith('+') and not line.startswith('+++'):
                content = line[1:]
                indent = len(content) - len(content.lstrip())
                # Replace real code with a comment
                new_lines.append('+' + ' ' * indent + '# Fixed: ' + content.strip())
                modified = True
            elif line.startswith('-') and not line.startswith('---'):
                # Keep the removal but don't add real replacement
                new_lines.append(line)
            else:
                new_lines.append(line)
        
        if not modified:
            return None
        
        return PatchExample(
            instance_id=ex["instance_id"] + "__comment_only",
            repo=ex["repo"],
            problem_statement=ex["problem_statement"],
            gold_patch=ex["gold_patch"],
            agent_patch='\n'.join(new_lines),
            agent_name="known-bad:comment-only",
            test_passed=False,
            base_commit=ex.get("base_commit", ""),
            difficulty=ex.get("difficulty", ""),
        )
    
    def _make_broad_except_variant(self, ex: dict) -> Optional[PatchExample]:
        """Create a patch that wraps everything in a broad try/except."""
        gold = ex["gold_patch"]
        lines = gold.split('\n')
        
        # Find added lines and wrap them in try/except
        new_lines = []
        added_lines = []
        min_indent = 999
        
        for line in lines:
            if line.startswith('+') and not line.startswith('+++'):
                content = line[1:]
                if content.strip():
                    indent = len(content) - len(content.lstrip())
                    min_indent = min(min_indent, indent)
                    added_lines.append(content)
        
        if not added_lines or min_indent == 999:
            return None
        
        indent_str = ' ' * min_indent
        
        # Reconstruct diff with try/except wrapper
        for line in lines:
            if line.startswith('+') and not line.startswith('+++'):
                continue  # Skip original added lines
            elif line.startswith('@@') and added_lines:
                new_lines.append(line)
                # Add our try/except version
                new_lines.append(f'+{indent_str}try:')
                for al in added_lines:
                    new_lines.append(f'+    {al}')
                new_lines.append(f'+{indent_str}except Exception:')
                new_lines.append(f'+{indent_str}    pass  # Silently ignore all errors')
                added_lines = []  # Only do this once
                continue
            new_lines.append(line)
        
        return PatchExample(
            instance_id=ex["instance_id"] + "__broad_except",
            repo=ex["repo"],
            problem_statement=ex["problem_statement"],
            gold_patch=ex["gold_patch"],
            agent_patch='\n'.join(new_lines),
            agent_name="known-bad:broad-except",
            test_passed=True,
            base_commit=ex.get("base_commit", ""),
            difficulty=ex.get("difficulty", ""),
        )
    
    def _make_test_disable_variant(self, ex: dict) -> Optional[PatchExample]:
        """Create a patch that 'fixes' the issue by modifying tests."""
        instance_id = ex["instance_id"]
        repo = ex["repo"]
        
        # Create a patch that adds pytest.mark.skip to failing tests
        patch = f"""diff --git a/tests/test_fix.py b/tests/test_fix.py
--- a/tests/test_fix.py
+++ b/tests/test_fix.py
@@ -1,5 +1,7 @@
+import pytest
+
 class TestFix:
-    def test_issue(self):
+    @pytest.mark.skip(reason="Known issue, will fix later")
+    def test_issue(self):
         # This test was failing
         assert True
"""
        return PatchExample(
            instance_id=instance_id + "__test_disable",
            repo=repo,
            problem_statement=ex["problem_statement"],
            gold_patch=ex["gold_patch"],
            agent_patch=patch,
            agent_name="known-bad:test-disable",
            test_passed=True,
            base_commit=ex.get("base_commit", ""),
            difficulty=ex.get("difficulty", ""),
        )


# ============================================================================
# Validator
# ============================================================================

class PatchJudgeValidator:
    """Validates PatchJudge scoring against ground truth."""
    
    def __init__(self, merge_threshold: float = 50.0):
        """
        Args:
            merge_threshold: Score below which a patch is considered "not merge-worthy".
        """
        self.merge_threshold = merge_threshold
    
    def validate(
        self,
        examples: list[PatchExample],
        results: list[JudgeResult],
        known_bad_results: Optional[list[tuple[PatchExample, JudgeResult]]] = None,
    ) -> ValidationResult:
        """Run full validation suite.
        
        Args:
            examples: The patch examples that were judged.
            results: The corresponding judge results.
            known_bad_results: Optional list of (example, result) for known-bad patches.
            
        Returns:
            ValidationResult with all metrics.
        """
        assert len(examples) == len(results), "examples and results must match"
        
        vr = ValidationResult(total_examples=len(examples))
        
        scores = [r.merge_score for r in results]
        
        # --- Score distribution ---
        if scores:
            vr.score_mean = statistics.mean(scores)
            vr.score_std = statistics.stdev(scores) if len(scores) > 1 else 0.0
            vr.score_median = statistics.median(scores)
        
        # --- METR alignment ---
        # Among test-passing patches, what fraction scores below threshold?
        passed_scores = [
            r.merge_score
            for ex, r in zip(examples, results)
            if ex.test_passed
        ]
        if passed_scores:
            below_threshold = sum(1 for s in passed_scores if s < self.merge_threshold)
            vr.test_passing_below_50_pct = below_threshold / len(passed_scores)
        
        # --- Resolved vs Unresolved separation ---
        resolved_scores = [
            r.merge_score
            for ex, r in zip(examples, results)
            if ex.test_passed
        ]
        unresolved_scores = [
            r.merge_score
            for ex, r in zip(examples, results)
            if not ex.test_passed
        ]
        
        if resolved_scores:
            vr.mean_score_resolved = statistics.mean(resolved_scores)
        if unresolved_scores:
            vr.mean_score_unresolved = statistics.mean(unresolved_scores)
        
        # Basic correlation: difference between resolved and unresolved means
        if resolved_scores and unresolved_scores:
            # Point-biserial-ish: just use the difference normalized
            diff = vr.mean_score_resolved - vr.mean_score_unresolved
            combined_std = statistics.stdev(scores) if len(scores) > 1 else 1.0
            vr.score_resolved_correlation = min(1.0, max(-1.0, diff / max(combined_std, 0.01)))
        
        # --- Known-bad detection ---
        if known_bad_results:
            vr.known_bad_total = len(known_bad_results)
            vr.known_bad_detected = sum(
                1 for _, r in known_bad_results
                if r.merge_score < self.merge_threshold
            )
            vr.known_bad_detection_rate = (
                vr.known_bad_detected / vr.known_bad_total
                if vr.known_bad_total > 0 else 0.0
            )
        
        # --- Per-dimension stats ---
        dim_scores = defaultdict(list)
        for r in results:
            for dim, data in r.dimension_scores.items():
                dim_scores[dim].append(data.get("score", 0))
        
        for dim, ds in dim_scores.items():
            if ds:
                vr.dimension_stats[dim] = {
                    "mean": round(statistics.mean(ds), 2),
                    "std": round(statistics.stdev(ds) if len(ds) > 1 else 0.0, 2),
                    "median": statistics.median(ds),
                    "min": min(ds),
                    "max": max(ds),
                }
        
        return vr
    
    def print_report(
        self,
        vr: ValidationResult,
        examples: list[PatchExample],
        results: list[JudgeResult],
    ) -> str:
        """Generate a human-readable validation report."""
        lines = []
        lines.append("=" * 70)
        lines.append("  PatchJudge Validation Report")
        lines.append("=" * 70)
        
        lines.append(f"\n📊 Dataset: {vr.total_examples} examples")
        
        # Score distribution
        lines.append(f"\n📈 Score Distribution:")
        lines.append(f"  Mean:   {vr.score_mean:.1f}")
        lines.append(f"  Median: {vr.score_median:.1f}")
        lines.append(f"  Std:    {vr.score_std:.1f}")
        
        # Score histogram
        bins = [0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
        scores = [r.merge_score for r in results]
        hist = defaultdict(int)
        for s in scores:
            for i in range(len(bins) - 1):
                if bins[i] <= s < bins[i+1]:
                    hist[f"{bins[i]}-{bins[i+1]}"] += 1
                    break
            else:
                hist[f"90-100"] += 1
        
        lines.append(f"\n  Score Distribution:")
        for label in [f"{bins[i]}-{bins[i+1]}" for i in range(len(bins)-1)]:
            count = hist.get(label, 0)
            bar = "█" * count
            lines.append(f"    {label:>7}: {bar} ({count})")
        
        # METR alignment
        lines.append(f"\n🎯 METR Alignment:")
        lines.append(
            f"  Test-passing patches below {self.merge_threshold}: "
            f"{vr.test_passing_below_50_pct:.1%}"
        )
        metr_target = 0.50
        if abs(vr.test_passing_below_50_pct - metr_target) < 0.15:
            lines.append(f"  ✅ ALIGNED with METR finding (~50% not merge-worthy)")
        elif vr.test_passing_below_50_pct < metr_target - 0.15:
            lines.append(f"  ⚠️ Too lenient — scoring too many patches as merge-worthy")
        else:
            lines.append(f"  ⚠️ Too harsh — scoring too many patches as not merge-worthy")
        
        # Resolved vs Unresolved
        lines.append(f"\n🔀 Resolved vs Unresolved Separation:")
        lines.append(f"  Mean score (resolved):   {vr.mean_score_resolved:.1f}")
        lines.append(f"  Mean score (unresolved): {vr.mean_score_unresolved:.1f}")
        lines.append(f"  Separation:              {vr.mean_score_resolved - vr.mean_score_unresolved:+.1f}")
        lines.append(f"  Correlation:             {vr.score_resolved_correlation:.3f}")
        
        # Known-bad detection
        if vr.known_bad_total > 0:
            lines.append(f"\n🚨 Known-Bad Pattern Detection:")
            lines.append(
                f"  Detected: {vr.known_bad_detected}/{vr.known_bad_total} "
                f"({vr.known_bad_detection_rate:.1%})"
            )
            if vr.known_bad_detection_rate >= 0.80:
                lines.append(f"  ✅ Good detection rate")
            else:
                lines.append(f"  ⚠️ Detection rate below 80% — judge may be too lenient")
        
        # Per-dimension stats
        lines.append(f"\n📐 Per-Dimension Scores:")
        for dim in ["correctness", "completeness", "code_quality", 
                     "non_regression_risk", "merge_readiness"]:
            stats = vr.dimension_stats.get(dim, {})
            if stats:
                lines.append(
                    f"  {dim:>25}: "
                    f"mean={stats['mean']:.1f}  "
                    f"std={stats['std']:.1f}  "
                    f"[{stats['min']}-{stats['max']}]"
                )
        
        # Top flags
        all_flags = defaultdict(int)
        for r in results:
            for dim, data in r.dimension_scores.items():
                for flag in data.get("flags", []):
                    if flag and flag != "JUDGE_ERROR":
                        all_flags[flag] += 1
        
        if all_flags:
            lines.append(f"\n🏴 Most Common Flags:")
            for flag, count in sorted(all_flags.items(), key=lambda x: -x[1])[:10]:
                lines.append(f"  {count:>4}x  {flag}")
        
        # Example best/worst
        scored = list(zip(examples, results))
        scored.sort(key=lambda x: x[1].merge_score, reverse=True)
        
        if len(scored) >= 3:
            lines.append(f"\n⭐ Top 3 Patches:")
            for ex, r in scored[:3]:
                lines.append(
                    f"  {r.merge_score:5.1f}  {ex.instance_id}  "
                    f"({ex.agent_name}, {'PASS' if ex.test_passed else 'FAIL'})"
                )
            
            lines.append(f"\n💀 Bottom 3 Patches:")
            for ex, r in scored[-3:]:
                lines.append(
                    f"  {r.merge_score:5.1f}  {ex.instance_id}  "
                    f"({ex.agent_name}, {'PASS' if ex.test_passed else 'FAIL'})"
                )
        
        lines.append("\n" + "=" * 70)
        
        report = '\n'.join(lines)
        return report


def run_full_validation(
    examples: list[PatchExample],
    results: list[JudgeResult],
    gold_data: Optional[list[dict]] = None,
    judge=None,
) -> tuple[ValidationResult, str]:
    """Run the complete validation pipeline.
    
    Args:
        examples: Judged patch examples.
        results: Judge results for those examples.
        gold_data: Gold standard data for generating known-bad patches.
        judge: PatchJudge instance (needed if judging known-bad patches).
        
    Returns:
        (ValidationResult, report_string)
    """
    known_bad_results = None
    
    # Generate and judge known-bad patches if we have gold data and a judge
    if gold_data and judge:
        logger.info("Generating known-bad patches...")
        bad_patches = KnownBadPatchGenerator.generate_all(gold_data)
        
        if bad_patches:
            logger.info(f"Judging {len(bad_patches)} known-bad patches...")
            bad_judge_results = judge.judge_batch(bad_patches, show_progress=True)
            known_bad_results = list(zip(bad_patches, bad_judge_results))
            
            # Print known-bad summary
            for bp, br in known_bad_results[:5]:
                logger.info(
                    f"  Known-bad [{bp.agent_name}] "
                    f"{bp.instance_id}: {br.merge_score:.1f}/100"
                )
    
    # Run validation
    validator = PatchJudgeValidator()
    vr = validator.validate(examples, results, known_bad_results)
    report = validator.print_report(vr, examples, results)
    
    return vr, report

	"""Validation module for PatchJudge.

	Validates that PatchJudge scores correlate with actual code quality:
	1. METR alignment: ~50% of test-passing patches should score below 50
	2. Known-bad pattern detection: deliberately bad patches should score low
	3. Score distribution analysis
	4. Resolved vs unresolved separation
	"""

	import json
	import logging
	import statistics
	from collections import defaultdict
	from typing import Optional

	from patchjudge.models import (
	PatchExample, PatchFeatures, JudgeResult, ValidationResult
	)

	logger = logging.getLogger(__name__)


	# ============================================================================
	# Known-Bad Patch Generator
	# ============================================================================

	class KnownBadPatchGenerator:
	"""Generate deliberately bad patches that pass tests but are low quality."""

	@staticmethod
	def generate_all(gold_examples: list[dict]) -> list[PatchExample]:
	"""Generate known-bad variants for a set of gold examples.

	Args:
	gold_examples: List of dicts with keys:
	instance_id, repo, problem_statement, gold_patch, base_commit

	Returns:
	List of PatchExample with known-bad patches.
	"""
	bad_patches = []
	generator = KnownBadPatchGenerator()

	for ex in gold_examples[:50]: # Generate from up to 50 examples
	variants = generator._generate_variants(ex)
	bad_patches.extend(variants)

	logger.info(f"Generated {len(bad_patches)} known-bad patch variants")
	return bad_patches

	def _generate_variants(self, ex: dict) -> list[PatchExample]:
	"""Generate known-bad variants of a gold patch."""
	variants = []
	gold = ex.get("gold_patch", "")

	if not gold:
	return variants

	# Variant 1: Hardcoded return values
	hardcoded = self._make_hardcoded_variant(ex)
	if hardcoded:
	variants.append(hardcoded)

	# Variant 2: Pass statement (minimal no-op)
	noop = self._make_noop_variant(ex)
	if noop:
	variants.append(noop)

	# Variant 3: Comment-only patch
	comment_only = self._make_comment_only_variant(ex)
	if comment_only:
	variants.append(comment_only)

	# Variant 4: Overly broad try/except
	broad_except = self._make_broad_except_variant(ex)
	if broad_except:
	variants.append(broad_except)

	# Variant 5: Test-disabling patch
	test_disable = self._make_test_disable_variant(ex)
	if test_disable:
	variants.append(test_disable)

	return variants

	def _make_hardcoded_variant(self, ex: dict) -> Optional[PatchExample]:
	"""Create a patch with hardcoded return values."""
	gold = ex["gold_patch"]
	lines = gold.split('\n')

	# Find added lines with return statements and hardcode them
	new_lines = []
	modified = False
	for line in lines:
	if line.startswith('+') and not line.startswith('+++'):
	content = line[1:]
	if 'return' in content and not modified:
	# Replace with hardcoded value
	indent = len(content) - len(content.lstrip())
	new_lines.append('+' + ' ' * indent + 'return True # HARDCODED')
	modified = True
	continue
	new_lines.append(line)

	if not modified:
	return None

	return PatchExample(
	instance_id=ex["instance_id"] + "__hardcoded",
	repo=ex["repo"],
	problem_statement=ex["problem_statement"],
	gold_patch=ex["gold_patch"],
	agent_patch='\n'.join(new_lines),
	agent_name="known-bad:hardcoded",
	test_passed=True, # Assume it passes (test oracle weakness)
	base_commit=ex.get("base_commit", ""),
	difficulty=ex.get("difficulty", ""),
	)

	def _make_noop_variant(self, ex: dict) -> Optional[PatchExample]:
	"""Create a minimal no-op patch (just adds 'pass')."""
	gold = ex["gold_patch"]
	lines = gold.split('\n')

	# Find the first hunk and replace all added lines with 'pass'
	new_lines = []
	in_hunk = False
	added_pass = False

	for line in lines:
	if line.startswith('@@'):
	in_hunk = True
	new_lines.append(line)
	continue

	if in_hunk:
	if line.startswith('+') and not line.startswith('+++'):
	if not added_pass:
	content = line[1:]
	indent = len(content) - len(content.lstrip())
	new_lines.append('+' + ' ' * indent + 'pass # TODO: implement')
	added_pass = True
	# Skip other added lines
	continue
	elif line.startswith('-') and not line.startswith('---'):
	new_lines.append(line)
	else:
	new_lines.append(line)
	else:
	new_lines.append(line)

	if not added_pass:
	return None

	return PatchExample(
	instance_id=ex["instance_id"] + "__noop",
	repo=ex["repo"],
	problem_statement=ex["problem_statement"],
	gold_patch=ex["gold_patch"],
	agent_patch='\n'.join(new_lines),
	agent_name="known-bad:noop",
	test_passed=False,
	base_commit=ex.get("base_commit", ""),
	difficulty=ex.get("difficulty", ""),
	)

	def _make_comment_only_variant(self, ex: dict) -> Optional[PatchExample]:
	"""Create a patch that only adds comments, no real code changes."""
	gold = ex["gold_patch"]
	lines = gold.split('\n')

	new_lines = []
	modified = False

	for line in lines:
	if line.startswith('+') and not line.startswith('+++'):
	content = line[1:]
	indent = len(content) - len(content.lstrip())
	# Replace real code with a comment
	new_lines.append('+' + ' ' * indent + '# Fixed: ' + content.strip())
	modified = True
	elif line.startswith('-') and not line.startswith('---'):
	# Keep the removal but don't add real replacement
	new_lines.append(line)
	else:
	new_lines.append(line)

	if not modified:
	return None

	return PatchExample(
	instance_id=ex["instance_id"] + "__comment_only",
	repo=ex["repo"],
	problem_statement=ex["problem_statement"],
	gold_patch=ex["gold_patch"],
	agent_patch='\n'.join(new_lines),
	agent_name="known-bad:comment-only",
	test_passed=False,
	base_commit=ex.get("base_commit", ""),
	difficulty=ex.get("difficulty", ""),
	)

	def _make_broad_except_variant(self, ex: dict) -> Optional[PatchExample]:
	"""Create a patch that wraps everything in a broad try/except."""
	gold = ex["gold_patch"]
	lines = gold.split('\n')

	# Find added lines and wrap them in try/except
	new_lines = []
	added_lines = []
	min_indent = 999

	for line in lines:
	if line.startswith('+') and not line.startswith('+++'):
	content = line[1:]
	if content.strip():
	indent = len(content) - len(content.lstrip())
	min_indent = min(min_indent, indent)
	added_lines.append(content)

	if not added_lines or min_indent == 999:
	return None

	indent_str = ' ' * min_indent

	# Reconstruct diff with try/except wrapper
	for line in lines:
	if line.startswith('+') and not line.startswith('+++'):
	continue # Skip original added lines
	elif line.startswith('@@') and added_lines:
	new_lines.append(line)
	# Add our try/except version
	new_lines.append(f'+{indent_str}try:')
	for al in added_lines:
	new_lines.append(f'+ {al}')
	new_lines.append(f'+{indent_str}except Exception:')
	new_lines.append(f'+{indent_str} pass # Silently ignore all errors')
	added_lines = [] # Only do this once
	continue
	new_lines.append(line)

	return PatchExample(
	instance_id=ex["instance_id"] + "__broad_except",
	repo=ex["repo"],
	problem_statement=ex["problem_statement"],
	gold_patch=ex["gold_patch"],
	agent_patch='\n'.join(new_lines),
	agent_name="known-bad:broad-except",
	test_passed=True,
	base_commit=ex.get("base_commit", ""),
	difficulty=ex.get("difficulty", ""),
	)

	def _make_test_disable_variant(self, ex: dict) -> Optional[PatchExample]:
	"""Create a patch that 'fixes' the issue by modifying tests."""
	instance_id = ex["instance_id"]
	repo = ex["repo"]

	# Create a patch that adds pytest.mark.skip to failing tests
	patch = f"""diff --git a/tests/test_fix.py b/tests/test_fix.py
	--- a/tests/test_fix.py
	+++ b/tests/test_fix.py
	@@ -1,5 +1,7 @@
	+import pytest
	+
	class TestFix:
	- def test_issue(self):
	+ @pytest.mark.skip(reason="Known issue, will fix later")
	+ def test_issue(self):
	# This test was failing
	assert True
	"""
	return PatchExample(
	instance_id=instance_id + "__test_disable",
	repo=repo,
	problem_statement=ex["problem_statement"],
	gold_patch=ex["gold_patch"],
	agent_patch=patch,
	agent_name="known-bad:test-disable",
	test_passed=True,
	base_commit=ex.get("base_commit", ""),
	difficulty=ex.get("difficulty", ""),
	)


	# ============================================================================
	# Validator
	# ============================================================================

	class PatchJudgeValidator:
	"""Validates PatchJudge scoring against ground truth."""

	def __init__(self, merge_threshold: float = 50.0):
	"""
	Args:
	merge_threshold: Score below which a patch is considered "not merge-worthy".
	"""
	self.merge_threshold = merge_threshold

	def validate(
	self,
	examples: list[PatchExample],
	results: list[JudgeResult],
	known_bad_results: Optional[list[tuple[PatchExample, JudgeResult]]] = None,
	) -> ValidationResult:
	"""Run full validation suite.

	Args:
	examples: The patch examples that were judged.
	results: The corresponding judge results.
	known_bad_results: Optional list of (example, result) for known-bad patches.

	Returns:
	ValidationResult with all metrics.
	"""
	assert len(examples) == len(results), "examples and results must match"

	vr = ValidationResult(total_examples=len(examples))

	scores = [r.merge_score for r in results]

	# --- Score distribution ---
	if scores:
	vr.score_mean = statistics.mean(scores)
	vr.score_std = statistics.stdev(scores) if len(scores) > 1 else 0.0
	vr.score_median = statistics.median(scores)

	# --- METR alignment ---
	# Among test-passing patches, what fraction scores below threshold?
	passed_scores = [
	r.merge_score
	for ex, r in zip(examples, results)
	if ex.test_passed
	]
	if passed_scores:
	below_threshold = sum(1 for s in passed_scores if s < self.merge_threshold)
	vr.test_passing_below_50_pct = below_threshold / len(passed_scores)

	# --- Resolved vs Unresolved separation ---
	resolved_scores = [
	r.merge_score
	for ex, r in zip(examples, results)
	if ex.test_passed
	]
	unresolved_scores = [
	r.merge_score
	for ex, r in zip(examples, results)
	if not ex.test_passed
	]

	if resolved_scores:
	vr.mean_score_resolved = statistics.mean(resolved_scores)
	if unresolved_scores:
	vr.mean_score_unresolved = statistics.mean(unresolved_scores)

	# Basic correlation: difference between resolved and unresolved means
	if resolved_scores and unresolved_scores:
	# Point-biserial-ish: just use the difference normalized
	diff = vr.mean_score_resolved - vr.mean_score_unresolved
	combined_std = statistics.stdev(scores) if len(scores) > 1 else 1.0
	vr.score_resolved_correlation = min(1.0, max(-1.0, diff / max(combined_std, 0.01)))

	# --- Known-bad detection ---
	if known_bad_results:
	vr.known_bad_total = len(known_bad_results)
	vr.known_bad_detected = sum(
	1 for _, r in known_bad_results
	if r.merge_score < self.merge_threshold
	)
	vr.known_bad_detection_rate = (
	vr.known_bad_detected / vr.known_bad_total
	if vr.known_bad_total > 0 else 0.0
	)

	# --- Per-dimension stats ---
	dim_scores = defaultdict(list)
	for r in results:
	for dim, data in r.dimension_scores.items():
	dim_scores[dim].append(data.get("score", 0))

	for dim, ds in dim_scores.items():
	if ds:
	vr.dimension_stats[dim] = {
	"mean": round(statistics.mean(ds), 2),
	"std": round(statistics.stdev(ds) if len(ds) > 1 else 0.0, 2),
	"median": statistics.median(ds),
	"min": min(ds),
	"max": max(ds),
	}

	return vr

	def print_report(
	self,
	vr: ValidationResult,
	examples: list[PatchExample],
	results: list[JudgeResult],
	) -> str:
	"""Generate a human-readable validation report."""
	lines = []
	lines.append("=" * 70)
	lines.append(" PatchJudge Validation Report")
	lines.append("=" * 70)

	lines.append(f"\n📊 Dataset: {vr.total_examples} examples")

	# Score distribution
	lines.append(f"\n📈 Score Distribution:")
	lines.append(f" Mean: {vr.score_mean:.1f}")
	lines.append(f" Median: {vr.score_median:.1f}")
	lines.append(f" Std: {vr.score_std:.1f}")

	# Score histogram
	bins = [0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
	scores = [r.merge_score for r in results]
	hist = defaultdict(int)
	for s in scores:
	for i in range(len(bins) - 1):
	if bins[i] <= s < bins[i+1]:
	hist[f"{bins[i]}-{bins[i+1]}"] += 1
	break
	else:
	hist[f"90-100"] += 1

	lines.append(f"\n Score Distribution:")
	for label in [f"{bins[i]}-{bins[i+1]}" for i in range(len(bins)-1)]:
	count = hist.get(label, 0)
	bar = "█" * count
	lines.append(f" {label:>7}: {bar} ({count})")

	# METR alignment
	lines.append(f"\n🎯 METR Alignment:")
	lines.append(
	f" Test-passing patches below {self.merge_threshold}: "
	f"{vr.test_passing_below_50_pct:.1%}"
	)
	metr_target = 0.50
	if abs(vr.test_passing_below_50_pct - metr_target) < 0.15:
	lines.append(f" ✅ ALIGNED with METR finding (~50% not merge-worthy)")
	elif vr.test_passing_below_50_pct < metr_target - 0.15:
	lines.append(f" ⚠️ Too lenient — scoring too many patches as merge-worthy")
	else:
	lines.append(f" ⚠️ Too harsh — scoring too many patches as not merge-worthy")

	# Resolved vs Unresolved
	lines.append(f"\n🔀 Resolved vs Unresolved Separation:")
	lines.append(f" Mean score (resolved): {vr.mean_score_resolved:.1f}")
	lines.append(f" Mean score (unresolved): {vr.mean_score_unresolved:.1f}")
	lines.append(f" Separation: {vr.mean_score_resolved - vr.mean_score_unresolved:+.1f}")
	lines.append(f" Correlation: {vr.score_resolved_correlation:.3f}")

	# Known-bad detection
	if vr.known_bad_total > 0:
	lines.append(f"\n🚨 Known-Bad Pattern Detection:")
	lines.append(
	f" Detected: {vr.known_bad_detected}/{vr.known_bad_total} "
	f"({vr.known_bad_detection_rate:.1%})"
	)
	if vr.known_bad_detection_rate >= 0.80:
	lines.append(f" ✅ Good detection rate")
	else:
	lines.append(f" ⚠️ Detection rate below 80% — judge may be too lenient")

	# Per-dimension stats
	lines.append(f"\n📐 Per-Dimension Scores:")
	for dim in ["correctness", "completeness", "code_quality",
	"non_regression_risk", "merge_readiness"]:
	stats = vr.dimension_stats.get(dim, {})
	if stats:
	lines.append(
	f" {dim:>25}: "
	f"mean={stats['mean']:.1f} "
	f"std={stats['std']:.1f} "
	f"[{stats['min']}-{stats['max']}]"
	)

	# Top flags
	all_flags = defaultdict(int)
	for r in results:
	for dim, data in r.dimension_scores.items():
	for flag in data.get("flags", []):
	if flag and flag != "JUDGE_ERROR":
	all_flags[flag] += 1

	if all_flags:
	lines.append(f"\n🏴 Most Common Flags:")
	for flag, count in sorted(all_flags.items(), key=lambda x: -x[1])[:10]:
	lines.append(f" {count:>4}x {flag}")

	# Example best/worst
	scored = list(zip(examples, results))
	scored.sort(key=lambda x: x[1].merge_score, reverse=True)

	if len(scored) >= 3:
	lines.append(f"\n⭐ Top 3 Patches:")
	for ex, r in scored[:3]:
	lines.append(
	f" {r.merge_score:5.1f} {ex.instance_id} "
	f"({ex.agent_name}, {'PASS' if ex.test_passed else 'FAIL'})"
	)

	lines.append(f"\n💀 Bottom 3 Patches:")
	for ex, r in scored[-3:]:
	lines.append(
	f" {r.merge_score:5.1f} {ex.instance_id} "
	f"({ex.agent_name}, {'PASS' if ex.test_passed else 'FAIL'})"
	)

	lines.append("\n" + "=" * 70)

	report = '\n'.join(lines)
	return report


	def run_full_validation(
	examples: list[PatchExample],
	results: list[JudgeResult],
	gold_data: Optional[list[dict]] = None,
	judge=None,
	) -> tuple[ValidationResult, str]:
	"""Run the complete validation pipeline.

	Args:
	examples: Judged patch examples.
	results: Judge results for those examples.
	gold_data: Gold standard data for generating known-bad patches.
	judge: PatchJudge instance (needed if judging known-bad patches).

	Returns:
	(ValidationResult, report_string)
	"""
	known_bad_results = None

	# Generate and judge known-bad patches if we have gold data and a judge
	if gold_data and judge:
	logger.info("Generating known-bad patches...")
	bad_patches = KnownBadPatchGenerator.generate_all(gold_data)

	if bad_patches:
	logger.info(f"Judging {len(bad_patches)} known-bad patches...")
	bad_judge_results = judge.judge_batch(bad_patches, show_progress=True)
	known_bad_results = list(zip(bad_patches, bad_judge_results))

	# Print known-bad summary
	for bp, br in known_bad_results[:5]:
	logger.info(
	f" Known-bad [{bp.agent_name}] "
	f"{bp.instance_id}: {br.merge_score:.1f}/100"
	)

	# Run validation
	validator = PatchJudgeValidator()
	vr = validator.validate(examples, results, known_bad_results)
	report = validator.print_report(vr, examples, results)

	return vr, report