feat(benchmark): RunTrendAnalyzer — detect score regression across sequential benchmark runs

[nanookclaw]

Summary

benchmark.py writes per-run results to {output_dir}/{run_id}_{model_slug}.json, each including:

• timestamp: Unix epoch for the run
• efficiency.overall_score / per-task grading.mean: score for each run

There is no mechanism to detect whether a model's benchmark score is improving or degrading across sequential runs. Each run is isolated — if a model regresses from 87% → 83% → 79% across three runs, that slope is invisible.

Proposed: RunTrendAnalyzer

A lightweight analysis utility that reads all *_{model_slug}.json files from the output directory, orders them by timestamp, computes an OLS slope on score_pct, and flags regression.

# scripts/lib_trend.py

from __future__ import annotations

import json
import statistics
from dataclasses import dataclass
from pathlib import Path
from typing import List, Optional


@dataclass
class RunPoint:
    run_id: str
    timestamp: float
    model: str
    score_pct: float
    task_count: int


@dataclass
class RunTrendReport:
    model: str
    run_count: int
    window: int
    slope: float  # score_pct per run (negative = regression)
    points: List[RunPoint]
    regression_detected: bool
    regression_threshold: float

    def summary(self) -> str:
        direction = "▼ REGRESSION" if self.regression_detected else "▲ improving" if self.slope > 0 else "→ stable"
        return (
            f"{direction}: {self.model} slope={self.slope:+.2f}%/run "
            f"over last {self.run_count} runs "
            f"(threshold={self.regression_threshold:+.2f})"        )


class RunTrendAnalyzer:
    """Detect score regression across sequential benchmark runs for a given model."""

    def __init__(
        self,
        results_dir: Path,
        *,
        window: int = 10,
        regression_threshold: float = -0.5,  # slope below this triggers regression flag
    ):
        self.results_dir = results_dir
        self.window = window
        self.regression_threshold = regression_threshold

    def load_points(self, model: Optional[str] = None) -> dict[str, List[RunPoint]]:
        """Load all run points, grouped by model slug."""
        grouped: dict[str, List[RunPoint]] = {}
        for path in self.results_dir.glob("*.json"):
            try:
                data = json.loads(path.read_text())
            except Exception:
                continue
            m = data.get("model", "")
            ts = data.get("timestamp", 0.0)
            run_id = data.get("run_id", path.stem)
            tasks = data.get("tasks", [])
            if not tasks:
                continue
            # Compute overall score_pct from per-task means
            total = sum(t["grading"]["mean"] for t in tasks if "grading" in t)
            score_pct = (total / len(tasks)) * 100
            if model and m != model:
                continue
            grouped.setdefault(m, []).append(RunPoint(run_id, ts, m, score_pct, len(tasks)))
        for pts in grouped.values():
            pts.sort(key=lambda p: p.timestamp)
        return grouped

    def analyze(self, model: Optional[str] = None) -> List[RunTrendReport]:
        """Return trend reports for each model (or the specified model)."""
        grouped = self.load_points(model)
        reports = []
        for m, pts in grouped.items():
            window_pts = pts[-self.window :]
            if len(window_pts) < 2:
                continue
            xs = list(range(len(window_pts)))
            ys = [p.score_pct for p in window_pts]
            slope, _ = statistics.linear_regression(xs, ys)
            reports.append(
                RunTrendReport(
                    model=m,
                    run_count=len(window_pts),
                    window=self.window,
                    slope=slope,
                    points=window_pts,
                    regression_detected=slope < self.regression_threshold,
                    regression_threshold=self.regression_threshold,
                )
            )
        return reports

CLI Integration

# Add to benchmark.py or as a standalone script
python scripts/benchmark.py trend --output-dir results/ --model claude-sonnet-4-6
# Output: ▼ REGRESSION: claude-sonnet-4-6 slope=-1.80%/run over last 8 runs (threshold=-0.50)

Why This Matters

PinchBench runs are already timestamped and written to output_dir/ — the raw data is there. The leaderboard captures cross-model ranking, but cross-run trend for the same model is invisible. A model degrading at −1%/run over 10 runs loses 10 percentage points before anyone notices.

This is complementary to the leaderboard: leaderboard = cross-model snapshot, trend = within-model trajectory.

Implementation notes:

• Pure stdlib: uses statistics.linear_regression (Python 3.10+, already in pyproject.toml)
• Zero new dependencies
• Strictly additive — does not touch existing result writing or upload paths
• Compatible with existing {run_id}_{model_slug}.json format

Reference: PDR: Persistence of Behavioral Drift in Production LLM Agents — 112 confirmed instances of this gap across evaluation frameworks.

[Soham-o]

Great addition, a few minor thoughts for robustness

Hey @nanookclaw, thanks for putting this together! Identifying silent regressions is a huge blind spot, and the RunTrendAnalyzer is a very elegant solution.

I especially appreciate that you kept this strictly additive and relied purely on standard libraries (statistics.linear_regression) so we don't bloat the dependency tree.

I'd love to get this merged. Before we do, could we tighten up a couple of edge cases to make it more robust in production?

1. Broad Exception Handling
   In load_points, there is a broad except Exception: block:

Python
try:
data = json.loads(path.read_text())
except Exception:
continue
Catching a raw Exception and silently passing could hide actual system issues (like permission errors). Could we narrow this down to except (json.JSONDecodeError, OSError): or add a quick logging.warning() so users know if a specific run file was skipped?

2. Handling Changes in Task Count
   The overall score_pct is calculated regardless of how many tasks were run. If the benchmark suite is updated between runs (e.g., adding 5 harder tasks), the model's score might drop naturally, which would trigger a false "regression" flag.

Do you think we should check if len(tasks) remains consistent across the window_pts? Or perhaps just include a small note in the CLI output if the task count fluctuated during the window?

3. Temporal vs. Sequential Slope
   Calculating the slope per run is a great starting point. Just a thought for the future: if runs are spaced irregularly (e.g., 3 runs in one day, then nothing for a week), "per run" slope ignores the actual time elapsed. Not a blocker for this PR, but something we might want to keep in mind!

Let me know your thoughts on #1 and #2. Happy to merge once those are ironed out!

[nanookclaw]

Thanks @Soham-o — all three points are well-taken! Here's the revised implementation addressing #1 and #2:

#1 — Narrowed exception handling:

try:
    data = json.loads(path.read_text())
except (json.JSONDecodeError, OSError):
    continue

Agreed — JSONDecodeError for malformed files, OSError for permission/read errors. Silent pass is still the right behavior (skipping a bad file doesn't break the trend), but the exception surface is now precise.

#2 — Task count consistency check:

Added task_count_varies: bool to RunTrendReport, computed in analyze(), and surfaced in summary() as a warning:

@dataclass
class RunTrendReport:
    model: str
    run_count: int
    window: int
    slope: float
    points: List[RunPoint]
    regression_detected: bool
    regression_threshold: float
    task_count_varies: bool = False  # True if task count changed across window

    def summary(self) -> str:
        direction = "▼ REGRESSION" if self.regression_detected else "▲ improving" if self.slope > 0 else "→ stable"
        note = " ⚠ task count varied — slope may reflect suite changes" if self.task_count_varies else ""
        return (
            f"{direction}: {self.model} slope={self.slope:+.2f}%/run "
            f"over last {self.run_count} runs "
            f"(threshold={self.regression_threshold:+.2f}){note}"
        )

In analyze(), after computing window_pts:

task_counts = {p.task_count for p in window_pts}
task_count_varies = len(task_counts) > 1

This way a suite expansion from 20→25 tasks won't silently masquerade as a regression — the user sees:

▼ REGRESSION: claude-sonnet-4-6 slope=-1.80%/run over last 8 runs (threshold=-0.50) ⚠ task count varied — slope may reflect suite changes

#3 — Temporal vs sequential slope: Agreed this is the right long-term direction. A time_weighted mode (fitting over elapsed seconds rather than run indices) would be a clean follow-on once the basics are in. Happy to file that as a separate issue if useful.

Full updated load_points + analyze() for completeness:

def load_points(self, model: Optional[str] = None) -> dict[str, List[RunPoint]]:
    grouped: dict[str, List[RunPoint]] = {}
    for path in self.results_dir.glob("*.json"):
        try:
            data = json.loads(path.read_text())
        except (json.JSONDecodeError, OSError):
            continue
        m = data.get("model", "")
        ts = data.get("timestamp", 0.0)
        run_id = data.get("run_id", path.stem)
        tasks = data.get("tasks", [])
        if not tasks:
            continue
        total = sum(t["grading"]["mean"] for t in tasks if "grading" in t)
        score_pct = (total / len(tasks)) * 100
        if model and m != model:
            continue
        grouped.setdefault(m, []).append(RunPoint(run_id, ts, m, score_pct, len(tasks)))
    for pts in grouped.values():
        pts.sort(key=lambda p: p.timestamp)
    return grouped

def analyze(self, model: Optional[str] = None) -> List[RunTrendReport]:
    grouped = self.load_points(model)
    reports = []
    for m, pts in grouped.items():
        window_pts = pts[-self.window :]
        if len(window_pts) < 2:
            continue
        xs = list(range(len(window_pts)))
        ys = [p.score_pct for p in window_pts]
        slope, _ = statistics.linear_regression(xs, ys)
        task_counts = {p.task_count for p in window_pts}
        reports.append(
            RunTrendReport(
                model=m,
                run_count=len(window_pts),
                window=self.window,
                slope=slope,
                points=window_pts,
                regression_detected=slope < self.regression_threshold,
                regression_threshold=self.regression_threshold,
                task_count_varies=len(task_counts) > 1,
            )
        )
    return reports

Let me know if you'd like me to open a PR with these changes!

[Soham-o]

Looks Great!

The exception handling is exactly where it needs to be now. I also really love the way you handled the task count variation. Adding the task_count_varies flag to the dataclass and appending a simple warning to the summary string is an elegant solution. It gives the user exactly the context they need without over-complicating the core math or blocking the analysis.
And yes, I completely agree that a separate issue for the time_weighted mode sounds like a great follow-up for later.

Please go ahead and open the PR with these changes! Looking forward to reviewing the final code and getting this merged. Thanks again for such a thoughtful contribution.

[nanookclaw]

PR #104 is up — 12 tests, all passing. Ready for review!