Detecting Prompt Regression and Quality Drops in Production

from dataclasses import dataclass
from typing import Dict, List
import json
from datetime import datetime
from pathlib import Path

@dataclass
class Baseline:
    """Snapshot of system quality at a point in time"""
    version: str  # e.g., "v1.2.0" or git SHA
    timestamp: str
    prompt_hash: str  # Hash of prompt template
    model_name: str
    test_results: Dict[str, any]  # Full eval results
    summary_metrics: Dict[str, float]  # Aggregate scores
    per_category_metrics: Dict[str, Dict[str, float]]  # Broken down by category
    metadata: Dict[str, any]

class BaselineManager:
    """Manage quality baselines"""
    
    def __init__(self, baseline_dir: Path):
        self.baseline_dir = baseline_dir
        self.baseline_dir.mkdir(parents=True, exist_ok=True)
    
    def save_baseline(
        self,
        baseline: Baseline,
        name: str = "production"
    ) -> None:
        """
        Save baseline to disk
        
        Args:
            baseline: Baseline object to save
            name: Baseline identifier (e.g., "production", "staging", "v1.2.0")
        """
        baseline_file = self.baseline_dir / f"{name}.json"
        
        data = {
            'version': baseline.version,
            'timestamp': baseline.timestamp,
            'prompt_hash': baseline.prompt_hash,
            'model_name': baseline.model_name,
            'summary_metrics': baseline.summary_metrics,
            'per_category_metrics': baseline.per_category_metrics,
            'metadata': baseline.metadata
        }
        
        with open(baseline_file, 'w') as f:
            json.dump(data, f, indent=2)
        
        # Also save full test results separately (can be large)
        results_file = self.baseline_dir / f"{name}_results.json"
        with open(results_file, 'w') as f:
            json.dump(baseline.test_results, f, indent=2)
    
    def load_baseline(self, name: str = "production") -> Baseline:
        """Load baseline from disk"""
        baseline_file = self.baseline_dir / f"{name}.json"
        
        if not baseline_file.exists():
            raise FileNotFoundError(f"Baseline '{name}' not found")
        
        with open(baseline_file, 'r') as f:
            data = json.load(f)
        
        # Load full results if available
        results_file = self.baseline_dir / f"{name}_results.json"
        if results_file.exists():
            with open(results_file, 'r') as f:
                test_results = json.load(f)
        else:
            test_results = {}
        
        return Baseline(
            version=data['version'],
            timestamp=data['timestamp'],
            prompt_hash=data['prompt_hash'],
            model_name=data['model_name'],
            test_results=test_results,
            summary_metrics=data['summary_metrics'],
            per_category_metrics=data['per_category_metrics'],
            metadata=data.get('metadata', {})
        )
    
    def list_baselines(self) -> List[str]:
        """List all saved baselines"""
        return [
            f.stem for f in self.baseline_dir.glob("*.json")
            if not f.stem.endswith("_results")
        ]

def create_production_baseline(
    eval_dataset: Dataset,
    current_system: AISystem
) -> Baseline:
    """Create baseline from current production system"""
    
    # Run full evaluation
    results = run_evaluation(
        dataset=eval_dataset,
        system=current_system
    )
    
    # Compute summary metrics
    summary_metrics = {
        'pass_rate': results.pass_rate,
        'avg_score': results.avg_score,
        'avg_latency_ms': results.avg_latency_ms,
        'p95_latency_ms': results.p95_latency_ms,
        'p99_latency_ms': results.p99_latency_ms
    }
    
    # Compute per-category metrics
    per_category = {}
    for category in set(case.category for case in eval_dataset.cases):
        category_results = [
            r for r in results.results
            if r.category == category
        ]
        per_category[category] = {
            'pass_rate': sum(1 for r in category_results if r.passed) / len(category_results),
            'avg_score': sum(r.score for r in category_results) / len(category_results),
            'count': len(category_results)
        }
    
    # Create baseline
    baseline = Baseline(
        version=current_system.version,
        timestamp=datetime.utcnow().isoformat(),
        prompt_hash=hash_prompt(current_system.prompt),
        model_name=current_system.model_name,
        test_results=results.to_dict(),
        summary_metrics=summary_metrics,
        per_category_metrics=per_category,
        metadata={
            'git_sha': get_git_sha(),
            'deployment_env': 'production'
        }
    )
    
    # Save baseline
    manager = BaselineManager(Path("baselines"))
    manager.save_baseline(baseline, name="production")
    
    return baseline

def hash_prompt(prompt_template: str) -> str:
    """Generate hash of prompt for change detection"""
    import hashlib
    return hashlib.sha256(prompt_template.encode()).hexdigest()[:12]

# regression/change_detection.py
import numpy as np
from scipy import stats
from typing import Dict, List, Tuple

class ChangeDetector:
    """Statistical change detection for quality metrics"""
    
    def detect_regression(
        self,
        baseline_scores: List[float],
        current_scores: List[float],
        alpha: float = 0.05  # Significance level
    ) -> Dict[str, any]:
        """
        Detect if current scores represent a regression
        
        Uses multiple statistical tests for robustness
        
        Returns:
            regression_detected: bool
            p_value: float
            effect_size: float
            test_used: str
        """
        # Test 1: Two-sample t-test
        t_stat, t_pvalue = stats.ttest_ind(baseline_scores, current_scores)
        
        # Test 2: Mann-Whitney U (non-parametric alternative)
        u_stat, u_pvalue = stats.mannwhitneyu(
            baseline_scores,
            current_scores,
            alternative='greater'  # Baseline > current = regression
        )
        
        # Effect size (Cohen's d)
        effect_size = self._cohens_d(baseline_scores, current_scores)
        
        # Decision logic
        # Use t-test if data is roughly normal, otherwise Mann-Whitney
        normality_baseline = self._is_normal(baseline_scores)
        normality_current = self._is_normal(current_scores)
        
        if normality_baseline and normality_current:
            p_value = t_pvalue
            test_used = "t-test"
        else:
            p_value = u_pvalue
            test_used = "Mann-Whitney U"
        
        # Regression detected if:
        # 1. p-value < alpha (statistically significant)
        # 2. Mean decreased (not just different)
        # 3. Effect size is meaningful (|d| > 0.2)
        mean_baseline = np.mean(baseline_scores)
        mean_current = np.mean(current_scores)
        
        regression_detected = (
            p_value < alpha and
            mean_current < mean_baseline and
            abs(effect_size) > 0.2
        )
        
        return {
            'regression_detected': regression_detected,
            'p_value': p_value,
            'effect_size': effect_size,
            'test_used': test_used,
            'baseline_mean': mean_baseline,
            'current_mean': mean_current,
            'difference': mean_current - mean_baseline
        }
    
    def _cohens_d(self, group1: List[float], group2: List[float]) -> float:
        """Calculate Cohen's d effect size"""
        mean1, mean2 = np.mean(group1), np.mean(group2)
        std1, std2 = np.std(group1, ddof=1), np.std(group2, ddof=1)
        
        # Pooled standard deviation
        n1, n2 = len(group1), len(group2)
        pooled_std = np.sqrt(((n1-1)*std1**2 + (n2-1)*std2**2) / (n1+n2-2))
        
        return (mean1 - mean2) / pooled_std
    
    def _is_normal(self, data: List[float], alpha: float = 0.05) -> bool:
        """Test if data is normally distributed"""
        if len(data) < 8:
            return True  # Assume normal for small samples
        
        _, p_value = stats.shapiro(data)
        return p_value >= alpha

# Usage
detector = ChangeDetector()

baseline_scores = [0.89, 0.91, 0.88, 0.90, 0.92, 0.87, 0.91, 0.89]
current_scores = [0.82, 0.79, 0.81, 0.80, 0.83, 0.78, 0.81, 0.79]

result = detector.detect_regression(baseline_scores, current_scores)

if result['regression_detected']:
    print(f"⚠️  Regression detected!")
    print(f"Baseline mean: {result['baseline_mean']:.2f}")
    print(f"Current mean: {result['current_mean']:.2f}")
    print(f"Drop: {result['difference']:.2f}")
    print(f"Effect size: {result['effect_size']:.2f}")

class CUSUMDetector:
    """
    CUSUM (Cumulative Sum) for online change detection
    
    Detects when metric drifts below baseline
    """
    
    def __init__(
        self,
        baseline_mean: float,
        baseline_std: float,
        threshold: float = 5.0
    ):
        self.baseline_mean = baseline_mean
        self.baseline_std = baseline_std
        self.threshold = threshold
        self.cusum = 0.0
    
    def update(self, new_value: float) -> bool:
        """
        Update CUSUM with new observation
        
        Returns True if change detected
        """
        # Standardize value
        z_score = (self.baseline_mean - new_value) / self.baseline_std
        
        # Update CUSUM
        self.cusum = max(0, self.cusum + z_score - 0.5)
        
        # Check threshold
        return self.cusum > self.threshold
    
    def reset(self):
        """Reset CUSUM after addressing change"""
        self.cusum = 0.0

# Usage: Online monitoring
cusum = CUSUMDetector(
    baseline_mean=0.90,
    baseline_std=0.03,
    threshold=5.0
)

for new_score in production_scores:
    if cusum.update(new_score):
        alert("Regression detected via CUSUM!")
        cusum.reset()

# tests/test_regression.py
import pytest
from eval_suite import run_evaluation, Dataset
from baseline import BaselineManager

class TestRegressionSuite:
    """Automated regression tests"""
    
    @pytest.fixture
    def baseline(self):
        """Load production baseline"""
        manager = BaselineManager(Path("baselines"))
        return manager.load_baseline("production")
    
    @pytest.fixture
    def eval_dataset(self):
        """Load evaluation dataset"""
        dataset = Dataset(name="regression_suite", version="v1.0", base_path=Path("evals"))
        dataset.load()
        return dataset
    
    def test_overall_pass_rate_no_regression(self, baseline, eval_dataset):
        """Overall pass rate must not decrease by >3%"""
        
        # Run evaluation with current system
        current_results = run_evaluation(
            dataset=eval_dataset,
            system=get_current_system()
        )
        
        baseline_pass_rate = baseline.summary_metrics['pass_rate']
        current_pass_rate = current_results.pass_rate
        
        drop = baseline_pass_rate - current_pass_rate
        
        assert drop <= 0.03, (
            f"Pass rate regression detected: "
            f"{baseline_pass_rate:.1%} → {current_pass_rate:.1%} "
            f"(drop of {drop:.1%})"
        )
    
    def test_category_pass_rates_no_regression(self, baseline, eval_dataset):
        """No category should drop >5%"""
        
        current_results = run_evaluation(
            dataset=eval_dataset,
            system=get_current_system()
        )
        
        # Compute current per-category pass rates
        current_by_category = compute_category_metrics(current_results)
        
        regressions = []
        for category, baseline_metrics in baseline.per_category_metrics.items():
            baseline_rate = baseline_metrics['pass_rate']
            current_rate = current_by_category.get(category, {}).get('pass_rate', 0.0)
            
            drop = baseline_rate - current_rate
            if drop > 0.05:
                regressions.append({
                    'category': category,
                    'baseline': baseline_rate,
                    'current': current_rate,
                    'drop': drop
                })
        
        if regressions:
            report = "\n".join([
                f"  {r['category']}: {r['baseline']:.1%} → {r['current']:.1%} (-{r['drop']:.1%})"
                for r in regressions
            ])
            pytest.fail(f"Category regressions detected:\n{report}")
    
    def test_latency_no_regression(self, baseline, eval_dataset):
        """P95 latency must not increase >20%"""
        
        current_results = run_evaluation(
            dataset=eval_dataset,
            system=get_current_system()
        )
        
        baseline_p95 = baseline.summary_metrics['p95_latency_ms']
        current_p95 = current_results.p95_latency_ms
        
        increase_pct = (current_p95 - baseline_p95) / baseline_p95
        
        assert increase_pct <= 0.20, (
            f"Latency regression: "
            f"P95 {baseline_p95:.0f}ms → {current_p95:.0f}ms "
            f"(+{increase_pct:.1%})"
        )
    
    @pytest.mark.parametrize("critical_test_id", [
        "refund_policy_basic",
        "refund_policy_edge_case_30_days",
        "json_output_format",
        "multi_turn_context_tracking"
    ])
    def test_critical_cases_no_regression(
        self,
        critical_test_id: str,
        baseline,
        eval_dataset
    ):
        """Critical test cases must always pass"""
        
        current_results = run_evaluation(
            dataset=eval_dataset,
            system=get_current_system()
        )
        
        result = next(
            (r for r in current_results.results if r.test_id == critical_test_id),
            None
        )
        
        assert result is not None, f"Critical test {critical_test_id} not found"
        assert result.passed, (
            f"Critical test {critical_test_id} failed:\n"
            f"{result.error_message}"
        )

# .github/workflows/regression_tests.yml
name: Regression Tests

on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'agents/**'

jobs:
  regression:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run regression test suite
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          pytest tests/test_regression.py \
            --maxfail=1 \
            --verbose \
            --tb=short \
            --junit-xml=regression_results.xml
      
      - name: Upload results
        if: always()
        uses: actions/upload-artifact@v3
        with:
          name: regression-results
          path: regression_results.xml
      
      - name: Comment PR with results
        if: failure()
        uses: actions/github-script@v6
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: '⚠️ Regression tests failed. Review results before merging.'
            });

# monitoring/rolling_metrics.py
from collections import deque
from typing import Deque, Optional
import time

class RollingMetrics:
    """Track metrics over sliding time window"""
    
    def __init__(self, window_seconds: int = 3600):  # 1 hour default
        self.window_seconds = window_seconds
        self.data: Deque[tuple[float, float]] = deque()  # (timestamp, value)
    
    def add(self, value: float, timestamp: Optional[float] = None):
        """Add new data point"""
        if timestamp is None:
            timestamp = time.time()
        
        self.data.append((timestamp, value))
        self._prune_old()
    
    def _prune_old(self):
        """Remove data outside window"""
        cutoff = time.time() - self.window_seconds
        while self.data and self.data[0][0] < cutoff:
            self.data.popleft()
    
    def mean(self) -> float:
        """Current mean over window"""
        if not self.data:
            return 0.0
        return sum(v for _, v in self.data) / len(self.data)
    
    def count(self) -> int:
        """Number of points in window"""
        return len(self.data)
    
    def percentile(self, p: float) -> float:
        """Compute percentile (e.g., 0.95 for P95)"""
        if not self.data:
            return 0.0
        values = sorted([v for _, v in self.data])
        index = int(len(values) * p)
        return values[min(index, len(values)-1)]

# monitoring/monitor.py
import numpy as np
from dataclasses import dataclass
from typing import Dict

@dataclass
class QualitySnapshot:
    """Current quality metrics"""
    timestamp: float
    pass_rate_1h: float
    pass_rate_24h: float
    avg_score_1h: float
    p95_latency_1h: float
    request_count_1h: int

class ProductionMonitor:
    """Real-time quality monitoring"""
    
    def __init__(self, baseline: Baseline):
        self.baseline = baseline
        
        # Rolling windows
        self.scores_1h = RollingMetrics(window_seconds=3600)
        self.scores_24h = RollingMetrics(window_seconds=86400)
        self.latencies_1h = RollingMetrics(window_seconds=3600)
        
        # Change detector
        self.change_detector = CUSUMDetector(
            baseline_mean=baseline.summary_metrics['avg_score'],
            baseline_std=0.05,  # Estimate from baseline variance
            threshold=5.0
        )
    
    def record_request(
        self,
        score: float,
        latency_ms: float,
        timestamp: Optional[float] = None
    ):
        """Record production request metrics"""
        self.scores_1h.add(score, timestamp)
        self.scores_24h.add(score, timestamp)
        self.latencies_1h.add(latency_ms, timestamp)
        
        # Check for regression
        if self.change_detector.update(score):
            self._trigger_regression_alert()
    
    def get_snapshot(self) -> QualitySnapshot:
        """Get current quality snapshot"""
        return QualitySnapshot(
            timestamp=time.time(),
            pass_rate_1h=self._compute_pass_rate(self.scores_1h),
            pass_rate_24h=self._compute_pass_rate(self.scores_24h),
            avg_score_1h=self.scores_1h.mean(),
            p95_latency_1h=self.latencies_1h.percentile(0.95),
            request_count_1h=self.scores_1h.count()
        )
    
    def check_health(self) -> Dict[str, any]:
        """Check if current quality is healthy"""
        snapshot = self.get_snapshot()
        baseline_pass_rate = self.baseline.summary_metrics['pass_rate']
        baseline_latency = self.baseline.summary_metrics['p95_latency_ms']
        
        issues = []
        
        # Check pass rate
        if snapshot.pass_rate_1h < baseline_pass_rate - 0.05:
            issues.append({
                'type': 'pass_rate_drop',
                'severity': 'high',
                'message': f'Pass rate {snapshot.pass_rate_1h:.1%} vs baseline {baseline_pass_rate:.1%}'
            })
        
        # Check latency
        if snapshot.p95_latency_1h > baseline_latency * 1.5:
            issues.append({
                'type': 'latency_increase',
                'severity': 'medium',
                'message': f'P95 latency {snapshot.p95_latency_1h:.0f}ms vs baseline {baseline_latency:.0f}ms'
            })
        
        # Check sample size
        if snapshot.request_count_1h < 100:
            issues.append({
                'type': 'low_traffic',
                'severity': 'low',
                'message': f'Only {snapshot.request_count_1h} requests in last hour'
            })
        
        return {
            'healthy': len([i for i in issues if i['severity'] == 'high']) == 0,
            'issues': issues,
            'snapshot': snapshot
        }
    
    def _compute_pass_rate(self, rolling_metrics: RollingMetrics) -> float:
        """Compute pass rate (score >= 0.7)"""
        if rolling_metrics.count() == 0:
            return 0.0
        passed = sum(1 for _, score in rolling_metrics.data if score >= 0.7)
        return passed / rolling_metrics.count()
    
    def _trigger_regression_alert(self):
        """Trigger regression alert (implement with your alerting system)"""
        snapshot = self.get_snapshot()
        alert_message = f"""
🚨 Quality regression detected

Current 1h pass rate: {snapshot.pass_rate_1h:.1%}
Baseline: {self.baseline.summary_metrics['pass_rate']:.1%}

Investigate prompt/model changes from last deployment.
"""
        # Send to Slack, PagerDuty, etc.
        send_alert(alert_message)

# monitoring/alerts.py
from enum import Enum
from typing import List, Dict

class AlertSeverity(Enum):
    CRITICAL = "critical"  # Page immediately
    HIGH = "high"  # Alert in 5-10 minutes
    MEDIUM = "medium"  # Alert in 30-60 minutes
    LOW = "low"  # Log only, no alert

@dataclass
class AlertRule:
    """Alert rule configuration"""
    name: str
    metric: str
    condition: str  # e.g., "< baseline - 0.05"
    severity: AlertSeverity
    cooldown_minutes: int  # Min time between alerts
    
class AlertManager:
    """Manage quality alerts"""
    
    def __init__(self):
        self.rules = self._define_rules()
        self.last_alert_times: Dict[str, float] = {}
    
    def _define_rules(self) -> List[AlertRule]:
        """Define alert rules"""
        return [
            AlertRule(
                name="critical_pass_rate_drop",
                metric="pass_rate_1h",
                condition="< baseline - 0.10",
                severity=AlertSeverity.CRITICAL,
                cooldown_minutes=15
            ),
            AlertRule(
                name="high_pass_rate_drop",
                metric="pass_rate_1h",
                condition="< baseline - 0.05",
                severity=AlertSeverity.HIGH,
                cooldown_minutes=30
            ),
            AlertRule(
                name="latency_spike",
                metric="p95_latency_1h",
                condition="> baseline * 2.0",
                severity=AlertSeverity.HIGH,
                cooldown_minutes=30
            ),
            AlertRule(
                name="category_regression",
                metric="category_pass_rate",
                condition="< baseline - 0.15",
                severity=AlertSeverity.HIGH,
                cooldown_minutes=60
            )
        ]
    
    def check_alerts(
        self,
        snapshot: QualitySnapshot,
        baseline: Baseline
    ) -> List[Dict]:
        """Check all alert rules"""
        triggered_alerts = []
        
        for rule in self.rules:
            if self._should_alert(rule):
                if self._evaluate_condition(rule, snapshot, baseline):
                    triggered_alerts.append({
                        'rule': rule.name,
                        'severity': rule.severity.value,
                        'message': self._format_alert_message(rule, snapshot, baseline)
                    })
                    self._record_alert(rule.name)
        
        return triggered_alerts
    
    def _should_alert(self, rule: AlertRule) -> bool:
        """Check if cooldown period has passed"""
        if rule.name not in self.last_alert_times:
            return True
        
        time_since_last = time.time() - self.last_alert_times[rule.name]
        cooldown_seconds = rule.cooldown_minutes * 60
        
        return time_since_last >= cooldown_seconds
    
    def _evaluate_condition(
        self,
        rule: AlertRule,
        snapshot: QualitySnapshot,
        baseline: Baseline
    ) -> bool:
        """Evaluate alert condition"""
        # Parse condition
        if "<" in rule.condition:
            threshold_expr = rule.condition.split("<")[1].strip()
            if "baseline" in threshold_expr:
                # e.g., "baseline - 0.05"
                baseline_value = baseline.summary_metrics.get(rule.metric.replace("_1h", ""), 0)
                offset = float(threshold_expr.split("-")[1].strip())
                threshold = baseline_value - offset
            else:
                threshold = float(threshold_expr)
            
            current_value = getattr(snapshot, rule.metric)
            return current_value < threshold
        
        elif ">" in rule.condition:
            threshold_expr = rule.condition.split(">")[1].strip()
            if "baseline" in threshold_expr:
                baseline_value = baseline.summary_metrics.get(rule.metric.replace("_1h", ""), 0)
                multiplier = float(threshold_expr.split("*")[1].strip())
                threshold = baseline_value * multiplier
            else:
                threshold = float(threshold_expr)
            
            current_value = getattr(snapshot, rule.metric)
            return current_value > threshold
        
        return False
    
    def _format_alert_message(
        self,
        rule: AlertRule,
        snapshot: QualitySnapshot,
        baseline: Baseline
    ) -> str:
        """Format alert message"""
        current_value = getattr(snapshot, rule.metric)
        baseline_value = baseline.summary_metrics.get(rule.metric.replace("_1h", ""), 0)
        
        return f"""
Alert: {rule.name} ({rule.severity.value})

Metric: {rule.metric}
Current: {current_value:.2f}
Baseline: {baseline_value:.2f}
Condition: {rule.condition}

Investigate recent prompt/model changes.
"""
    
    def _record_alert(self, rule_name: str):
        """Record alert timestamp"""
        self.last_alert_times[rule_name] = time.time()

# regression/bisect.py
from typing import List, Optional
import subprocess

def find_regression_commit(
    good_commit: str,
    bad_commit: str,
    test_command: str
) -> str:
    """
    Binary search to find commit that introduced regression
    
    Args:
        good_commit: Known good commit SHA
        bad_commit: Known bad commit SHA (current)
        test_command: Command that fails on bad commit
        
    Returns:
        First bad commit SHA
    """
    # Get commit range
    result = subprocess.run(
        ["git", "log", "--oneline", f"{good_commit}..{bad_commit}"],
        capture_output=True,
        text=True
    )
    commits = [line.split()[0] for line in result.stdout.strip().split("\n")]
    
    if not commits:
        return bad_commit
    
    # Binary search
    left, right = 0, len(commits) - 1
    first_bad = None
    
    while left <= right:
        mid = (left + right) // 2
        commit = commits[mid]
        
        # Checkout commit
        subprocess.run(["git", "checkout", commit], check=True)
        
        # Run test
        test_result = subprocess.run(test_command, shell=True)
        
        if test_result.returncode == 0:
            # Test passed - regression is after this commit
            left = mid + 1
        else:
            # Test failed - regression is at or before this commit
            first_bad = commit
            right = mid - 1
    
    # Return to original commit
    subprocess.run(["git", "checkout", bad_commit], check=True)
    
    return first_bad or bad_commit

# Usage
bad_commit_sha = find_regression_commit(
    good_commit="abc123def",  # Last known good
    bad_commit="HEAD",
    test_command="pytest tests/test_regression.py"
)

print(f"Regression introduced in commit: {bad_commit_sha}")

def generate_regression_root_cause_report(
    baseline: Baseline,
    current_results: EvalResults,
    git_sha: str
) -> str:
    """Generate root cause analysis report"""
    
    # Find failing tests
    failures = [r for r in current_results.results if not r.passed]
    
    # Group by category
    failures_by_category = {}
    for failure in failures:
        cat = failure.category
        if cat not in failures_by_category:
            failures_by_category[cat] = []
        failures_by_category[cat].append(failure)
    
    # Get commit info
    commit_info = subprocess.run(
        ["git", "show", "--stat", git_sha],
        capture_output=True,
        text=True
    ).stdout
    
    # Build report
    report = f"""# Regression Root Cause Analysis

## Summary
- **Commit**: {git_sha}
- **Baseline**: {baseline.version}
- **Pass Rate**: {baseline.summary_metrics['pass_rate']:.1%} → {current_results.pass_rate:.1%}
- **Failures**: {len(failures)} ({len(failures)/len(current_results.results):.1%})

## Affected Categories
"""
    
    for cat, fails in failures_by_category.items():
        baseline_cat_rate = baseline.per_category_metrics.get(cat, {}).get('pass_rate', 1.0)
        current_cat_rate = sum(1 for r in current_results.results if r.category == cat and r.passed) / len([r for r in current_results.results if r.category == cat])
        
        report += f"\n### {cat}\n"
        report += f"- Pass rate: {baseline_cat_rate:.1%} → {current_cat_rate:.1%}\n"
        report += f"- Failures: {len(fails)}\n"
        report += f"- Example failures:\n"
        for fail in fails[:3]:
            report += f"  - `{fail.test_id}`: {fail.error_message[:100]}\n"
    
    report += f"\n## Commit Details\n```\n{commit_info}\n```\n"
    
    report += f"\n## Recommended Actions\n"
    report += f"1. Review prompt changes in commit {git_sha}\n"
    report += f"2. Run regression suite on previous commit to confirm\n"
    report += f"3. Either revert or fix prompt to address failures\n"
    
    return report

# Prompt Change Checklist

Before committing prompt changes:

- [ ] Run full regression test suite locally
- [ ] Check pass rate hasn't dropped >3%
- [ ] Verify critical test cases still pass
- [ ] Test on edge cases from previous incidents
- [ ] Review changes with team member
- [ ] Update baseline after merge

def gradual_prompt_rollout(
    new_prompt: str,
    traffic_percentage: float,
    monitoring_duration_hours: int = 24
):
    """
    Roll out new prompt to small % of traffic first
    
    Args:
        new_prompt: New prompt to test
        traffic_percentage: % of traffic to route to new prompt (e.g., 0.10)
        monitoring_duration_hours: How long to monitor before full rollout
    """
    # Deploy with traffic split
    deploy_with_traffic_split(
        prompt_a=current_production_prompt,
        prompt_b=new_prompt,
        traffic_b=traffic_percentage
    )
    
    # Monitor for duration
    time.sleep(monitoring_duration_hours * 3600)
    
    # Compare metrics
    metrics_a = get_metrics(variant="a")
    metrics_b = get_metrics(variant="b")
    
    # Decision
    if metrics_b['pass_rate'] >= metrics_a['pass_rate'] - 0.02:
        # New prompt is acceptable - full rollout
        deploy_full(new_prompt)
        return True
    else:
        # Regression detected - rollback
        rollback_to_variant_a()
        alert(f"Prompt rollout aborted due to {metrics_b['pass_rate']:.1%} vs {metrics_a['pass_rate']:.1%}")
        return False

def auto_rollback_on_regression(
    deployment_id: str,
    grace_period_minutes: int = 30,
    regression_threshold: float = 0.05
):
    """
    Automatically rollback if regression detected
    
    Monitors quality for grace period after deployment.
    Rolls back if pass rate drops below threshold.
    """
    deployment_time = time.time()
    monitor = ProductionMonitor(baseline=load_baseline("production"))
    
    while time.time() - deployment_time < grace_period_minutes * 60:
        health = monitor.check_health()
        
        if not health['healthy']:
            critical_issues = [i for i in health['issues'] if i['severity'] == 'high']
            
            if critical_issues:
                # Trigger rollback
                trigger_rollback(deployment_id)
                alert(f"Auto-rollback triggered for {deployment_id}: {critical_issues}")
                return False
        
        time.sleep(60)  # Check every minute
    
    # Grace period passed without issues
    return True