Spaces:
Sleeping
Sleeping
File size: 11,498 Bytes
ee933ab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 | # Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
"""
Pillar 4 — Performance Regression Sentinel.
After the agent applies a deep architectural fix, this module checks whether
the fix introduced a meaningful performance slowdown by running a small,
deterministic benchmark and comparing its timing distribution against a
pre-fix baseline using the Mann-Whitney U statistical test.
Design rationale:
A global threading.Lock on the DB connection will stop all flakiness, but
serialise every write and make the app 10x slower. FlakeForge v1 cannot
detect this — it only cares about pass/fail. The Sentinel closes that gap.
Reference:
Mostafa et al., "Risk-Aware Batch Testing for Performance Regression
Detection", arXiv 2025 — statistical Change-Point Detection on timing
distributions for CI without full benchmark suites.
"""
from __future__ import annotations
import logging
import math
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
# Optional scipy import — degrade gracefully when not installed
try:
from scipy import stats as _scipy_stats
_SCIPY_AVAILABLE = True
except ImportError:
_scipy_stats = None # type: ignore
_SCIPY_AVAILABLE = False
logger.warning(
"perf_sentinel: scipy not available — "
"falling back to median-ratio-only regression check."
)
try:
import numpy as _np
_NUMPY_AVAILABLE = True
except ImportError:
_np = None # type: ignore
_NUMPY_AVAILABLE = False
# ── Data classes ──────────────────────────────────────────────────────────────
@dataclass
class PerformanceBaseline:
"""Timing distribution captured during reset() before any patches."""
benchmark_test_id: str
timing_distribution_ms: List[float]
p50_ms: float
p95_ms: float
sample_count: int
@classmethod
def from_timings(cls, test_id: str, timings: List[float]) -> "PerformanceBaseline":
if not timings:
return cls(test_id, [], 0.0, 0.0, 0)
sorted_t = sorted(timings)
n = len(sorted_t)
p50 = sorted_t[n // 2]
p95 = sorted_t[int(n * 0.95)] if n > 1 else sorted_t[-1]
return cls(
benchmark_test_id=test_id,
timing_distribution_ms=timings,
p50_ms=p50,
p95_ms=p95,
sample_count=n,
)
@dataclass
class SentinelResult:
"""Result of a single regression check."""
is_regression: bool
median_ratio: float # post_median / baseline_median (1.0 = no change)
p_value: Optional[float] # Mann-Whitney p-value (None if scipy unavailable)
post_p50_ms: float
baseline_p50_ms: float
message: str
def to_dict(self) -> Dict[str, Any]:
return {
"regression": self.is_regression,
"median_ratio": round(self.median_ratio, 3),
"p_value": round(self.p_value, 4) if self.p_value is not None else None,
"post_p50_ms": round(self.post_p50_ms, 1),
"baseline_p50_ms": round(self.baseline_p50_ms, 1),
"message": self.message,
}
def penalty(self) -> float:
"""
Reward penalty for a detected regression.
Formula: 10 × log(median_ratio), capped at 25.
A 2× slowdown → 6.93 penalty.
A 10× slowdown → 23.03 penalty (worse than the 15-pt regression penalty).
"""
if not self.is_regression:
return 0.0
return min(25.0, 10.0 * math.log(max(1.0, self.median_ratio)))
# ── Main sentinel class ────────────────────────────────────────────────────────
class PerformanceSentinel:
"""
Captures a timing baseline and checks for regressions after patches.
Parameters
----------
slowdown_threshold:
Minimum median_ratio to consider a regression (default 1.5 = 50% slower).
p_value_cutoff:
Statistical significance level for the Mann-Whitney U test (default 0.05).
n_benchmark_runs:
Number of benchmark runs for both baseline and post-fix checks.
"""
def __init__(
self,
slowdown_threshold: float = 1.5,
p_value_cutoff: float = 0.05,
n_benchmark_runs: int = 20, # raised from 10 — Mann-Whitney U needs ≥20 per group
max_adaptive_samples: int = 50, # hard cap when re-sampling in inconclusive p-value zone
) -> None:
self.slowdown_threshold = slowdown_threshold
self.p_value_cutoff = p_value_cutoff
self.n_benchmark_runs = n_benchmark_runs
self.max_adaptive_samples = max_adaptive_samples
self._baseline: Optional[PerformanceBaseline] = None
@property
def has_baseline(self) -> bool:
return self._baseline is not None
def capture_baseline(self, runner: Any, benchmark_test_id: str) -> PerformanceBaseline:
"""
Run the benchmark test n times via `runner` and store the timing baseline.
``runner`` must expose a ``run_test_n_times(test_id, n)`` method that
returns a list of RunRecord objects with a ``duration_ms`` field.
"""
records = runner.run_test_n_times(benchmark_test_id, n=self.n_benchmark_runs)
timings = [float(r.duration_ms) for r in records]
self._baseline = PerformanceBaseline.from_timings(benchmark_test_id, timings)
logger.debug(
"perf_sentinel: baseline captured | test=%s p50=%.1fms p95=%.1fms n=%d",
benchmark_test_id,
self._baseline.p50_ms,
self._baseline.p95_ms,
self._baseline.sample_count,
)
return self._baseline
def check_regression(
self, runner: Any, benchmark_test_id: Optional[str] = None
) -> SentinelResult:
"""
Run the benchmark again and compare against the stored baseline.
Returns a SentinelResult. If no baseline has been captured or the
benchmark test cannot be found, returns a safe no-regression result.
"""
if self._baseline is None:
return SentinelResult(
is_regression=False,
median_ratio=1.0,
p_value=None,
post_p50_ms=0.0,
baseline_p50_ms=0.0,
message="No baseline captured — skipping performance check.",
)
test_id = benchmark_test_id or self._baseline.benchmark_test_id
try:
post_records = runner.run_test_n_times(test_id, n=self.n_benchmark_runs)
except Exception as exc:
logger.warning("perf_sentinel: benchmark run failed: %s", exc)
return SentinelResult(
is_regression=False,
median_ratio=1.0,
p_value=None,
post_p50_ms=0.0,
baseline_p50_ms=self._baseline.p50_ms,
message=f"Benchmark run error — skipping: {exc}",
)
post_timings = [float(r.duration_ms) for r in post_records]
if not post_timings:
return SentinelResult(
is_regression=False,
median_ratio=1.0,
p_value=None,
post_p50_ms=0.0,
baseline_p50_ms=self._baseline.p50_ms,
message="No timings collected — skipping.",
)
post_p50 = sorted(post_timings)[len(post_timings) // 2]
baseline_p50 = self._baseline.p50_ms
median_ratio = (post_p50 / baseline_p50) if baseline_p50 > 0 else 1.0
# Statistical test (requires scipy)
p_value: Optional[float] = None
stat_significant = True # default: rely only on median ratio when scipy absent
if _SCIPY_AVAILABLE and len(self._baseline.timing_distribution_ms) >= 5:
try:
_, p_value = _scipy_stats.mannwhitneyu(
self._baseline.timing_distribution_ms,
post_timings,
alternative="less", # H1: baseline < post (i.e. post is slower)
)
stat_significant = p_value < self.p_value_cutoff
# ── Adaptive re-sampling ─────────────────────────────────────────
# If p-value is in the inconclusive zone (0.05 – 0.15), gather up
# to max_adaptive_samples total observations and re-test.
# This prevents false-negatives on subtle 1.5x regressions.
_INCONCLUSIVE_UPPER = 0.15
adaptive_timings = list(post_timings)
while (
self.p_value_cutoff <= p_value <= _INCONCLUSIVE_UPPER
and len(adaptive_timings) < self.max_adaptive_samples
):
extra_needed = min(
10,
self.max_adaptive_samples - len(adaptive_timings),
)
try:
extra_records = runner.run_test_n_times(test_id, n=extra_needed)
adaptive_timings.extend([float(r.duration_ms) for r in extra_records])
_, p_value = _scipy_stats.mannwhitneyu(
self._baseline.timing_distribution_ms,
adaptive_timings,
alternative="less",
)
stat_significant = p_value < self.p_value_cutoff
post_p50 = sorted(adaptive_timings)[len(adaptive_timings) // 2]
median_ratio = (post_p50 / baseline_p50) if baseline_p50 > 0 else 1.0
except Exception as _exc:
logger.debug("perf_sentinel: adaptive sampling error: %s", _exc)
break
# ────────────────────────────────────────────────────────────────
except Exception as exc:
logger.debug("perf_sentinel: scipy test error: %s", exc)
is_regression = stat_significant and (median_ratio > self.slowdown_threshold)
if is_regression:
pct = (median_ratio - 1.0) * 100
msg = (
f"PERFORMANCE REGRESSION: {pct:.1f}% slower "
f"(baseline p50={baseline_p50:.1f}ms → post p50={post_p50:.1f}ms, "
f"ratio={median_ratio:.2f}x, p={p_value:.4f if p_value is not None else 'N/A'})"
)
else:
msg = (
f"Performance OK: ratio={median_ratio:.2f}x "
f"(baseline={baseline_p50:.1f}ms → post={post_p50:.1f}ms)"
)
logger.debug("perf_sentinel: %s", msg)
return SentinelResult(
is_regression=is_regression,
median_ratio=median_ratio,
p_value=p_value,
post_p50_ms=post_p50,
baseline_p50_ms=baseline_p50,
message=msg,
)
|