From 8e27ce0e0b42e98310745abee410bf8f5a369b14 Mon Sep 17 00:00:00 2001 From: Brandon Date: Fri, 10 Apr 2026 21:12:17 -0400 Subject: [PATCH] feat: add cross-run pass-rate trend analysis --- README.md | 19 ++++ src/agent_release_gate/cli.py | 32 ++++++ src/agent_release_gate/history.py | 172 ++++++++++++++++++++++++++++++ tests/test_cli.py | 103 +++++++++++++++++- tests/test_history.py | 79 ++++++++++++++ 5 files changed, 404 insertions(+), 1 deletion(-) create mode 100644 src/agent_release_gate/history.py create mode 100644 tests/test_history.py diff --git a/README.md b/README.md index fb0af4e..e54b54a 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ This tool makes those problems visible before release. - Fails if quality drops below your threshold - Optionally compares against a baseline report and blocks regressions - Optionally enforces baseline latency/cost drift caps so slower or pricier runs fail fast +- Optionally records run summaries and detects sustained pass-rate drift across releases - Supports per-case latency/cost limits to catch outliers hidden by averages - Enforces telemetry presence when global average latency/cost limits are configured - Outputs both JSON (for machines) and Markdown (for humans) @@ -48,6 +49,24 @@ The command exits with: Perfect for CI pipelines. +## Track trend drift across runs + +Single-baseline checks catch point-in-time regressions. Trend analysis catches slow degradation over many runs. + +```bash +# Record this run in a history folder +argate evaluate \ + --spec examples/spec.yaml \ + --results examples/results.json \ + --record-history .gate-history + +# Analyze the latest 10 runs and print trend JSON +argate trend --history .gate-history --window 10 + +# CI mode: fail when pass-rate trend is declining +argate trend --history .gate-history --fail-on-regression +``` + ## Spec format ```yaml diff --git a/src/agent_release_gate/cli.py b/src/agent_release_gate/cli.py index dc5493c..6733680 100644 --- a/src/agent_release_gate/cli.py +++ b/src/agent_release_gate/cli.py @@ -7,6 +7,7 @@ from typing import List, Optional from .evaluator import evaluate, to_dict, to_markdown +from .history import analyze_history, record_history, trend_to_dict def build_parser() -> argparse.ArgumentParser: @@ -19,6 +20,20 @@ def build_parser() -> argparse.ArgumentParser: eval_cmd.add_argument("--baseline", help="Optional baseline report JSON") eval_cmd.add_argument("--output", help="Write report JSON to this path") eval_cmd.add_argument("--markdown", help="Write markdown report to this path") + eval_cmd.add_argument( + "--record-history", + help="Optional directory to append run summaries for cross-run trend analysis", + ) + + trend_cmd = sub.add_parser("trend", help="Analyze pass-rate trend across historical run summaries") + trend_cmd.add_argument("--history", required=True, help="Directory containing historical run JSON") + trend_cmd.add_argument("--window", type=int, default=10, help="Number of most recent runs to analyze") + trend_cmd.add_argument("--output", help="Write trend report JSON to this path") + trend_cmd.add_argument( + "--fail-on-regression", + action="store_true", + help="Exit non-zero when the trend direction is declining", + ) return parser @@ -39,8 +54,25 @@ def main(argv: Optional[List[str]] = None) -> int: if args.markdown: Path(args.markdown).write_text(to_markdown(report), encoding="utf-8") + if args.record_history: + record_history(report, history_dir=args.record_history) + return 0 if report.summary.gate_passed else 1 + if args.command == "trend": + trend_report = analyze_history(history_dir=args.history, window=args.window) + payload = trend_to_dict(trend_report) + + print(json.dumps(payload, indent=2)) + + if args.output: + Path(args.output).write_text(json.dumps(payload, indent=2), encoding="utf-8") + + if args.fail_on_regression and trend_report.any_regression: + return 1 + + return 0 + parser.print_help() return 2 diff --git a/src/agent_release_gate/history.py b/src/agent_release_gate/history.py new file mode 100644 index 0000000..90a5a7d --- /dev/null +++ b/src/agent_release_gate/history.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional, Union + +from .models import GateReport + + +@dataclass +class GateRunSummary: + run_id: str + timestamp: str + pass_rate: float + avg_latency_ms: Optional[float] + avg_cost_usd: Optional[float] + gate_passed: bool + + +@dataclass +class GateTrendReport: + window: int + runs: list[GateRunSummary] + pass_rate_slope: float + pass_rate_direction: str + any_regression: bool + + +def _timestamp_id() -> str: + return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + +def _iso_utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _run_payload(run: GateRunSummary) -> dict: + return { + "run_id": run.run_id, + "timestamp": run.timestamp, + "summary": { + "pass_rate": run.pass_rate, + "avg_latency_ms": run.avg_latency_ms, + "avg_cost_usd": run.avg_cost_usd, + "gate_passed": run.gate_passed, + }, + } + + +def _run_from_payload(data: dict, fallback_run_id: str) -> GateRunSummary: + summary = data.get("summary", data) or {} + return GateRunSummary( + run_id=str(data.get("run_id") or fallback_run_id), + timestamp=str(data.get("timestamp") or ""), + pass_rate=float(summary.get("pass_rate", 0.0)), + avg_latency_ms=( + float(summary["avg_latency_ms"]) + if summary.get("avg_latency_ms") is not None + else None + ), + avg_cost_usd=( + float(summary["avg_cost_usd"]) + if summary.get("avg_cost_usd") is not None + else None + ), + gate_passed=bool(summary.get("gate_passed", False)), + ) + + +def _sort_key(run: GateRunSummary) -> tuple[str, str]: + return (run.timestamp, run.run_id) + + +def _ols_slope(values: list[float]) -> float: + n = len(values) + if n < 2: + return 0.0 + + x_mean = (n - 1) / 2 + y_mean = sum(values) / n + denominator = sum((x - x_mean) ** 2 for x in range(n)) + if denominator == 0: + return 0.0 + + numerator = sum((x - x_mean) * (y - y_mean) for x, y in enumerate(values)) + return numerator / denominator + + +def summary_from_report(report: GateReport, run_id: Optional[str] = None) -> GateRunSummary: + return GateRunSummary( + run_id=run_id or _timestamp_id(), + timestamp=_iso_utc_now(), + pass_rate=report.summary.pass_rate, + avg_latency_ms=report.summary.avg_latency_ms, + avg_cost_usd=report.summary.avg_cost_usd, + gate_passed=report.summary.gate_passed, + ) + + +def record_history( + report: GateReport, + history_dir: Union[str, Path], + run_id: Optional[str] = None, +) -> Path: + run = summary_from_report(report=report, run_id=run_id) + history_path = Path(history_dir) + history_path.mkdir(parents=True, exist_ok=True) + + file_path = history_path / f"{run.run_id}.json" + if file_path.exists(): + file_path = history_path / f"{run.run_id}-{int(datetime.now(timezone.utc).timestamp())}.json" + + file_path.write_text(json.dumps(_run_payload(run), indent=2), encoding="utf-8") + return file_path + + +def load_recent_runs(history_dir: Union[str, Path], window: int = 10) -> list[GateRunSummary]: + history_path = Path(history_dir) + if not history_path.exists(): + return [] + + runs: list[GateRunSummary] = [] + for file_path in history_path.glob("*.json"): + data = json.loads(file_path.read_text(encoding="utf-8")) + runs.append(_run_from_payload(data, fallback_run_id=file_path.stem)) + + runs.sort(key=_sort_key) + if window > 0: + return runs[-window:] + return runs + + +def analyze_history(history_dir: Union[str, Path], window: int = 10) -> GateTrendReport: + runs = load_recent_runs(history_dir=history_dir, window=window) + slope = _ols_slope([r.pass_rate for r in runs]) if len(runs) >= 3 else 0.0 + + if slope < -0.001: + direction = "declining" + elif slope > 0.001: + direction = "improving" + else: + direction = "stable" + + return GateTrendReport( + window=window, + runs=runs, + pass_rate_slope=round(slope, 6), + pass_rate_direction=direction, + any_regression=direction == "declining", + ) + + +def trend_to_dict(report: GateTrendReport) -> dict: + return { + "window": report.window, + "runs": [ + { + "run_id": run.run_id, + "timestamp": run.timestamp, + "pass_rate": run.pass_rate, + "avg_latency_ms": run.avg_latency_ms, + "avg_cost_usd": run.avg_cost_usd, + "gate_passed": run.gate_passed, + } + for run in report.runs + ], + "pass_rate_slope": report.pass_rate_slope, + "pass_rate_direction": report.pass_rate_direction, + "any_regression": report.any_regression, + } diff --git a/tests/test_cli.py b/tests/test_cli.py index b305607..bd64e48 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,4 +1,7 @@ -from agent_release_gate.cli import build_parser +import json +from pathlib import Path + +from agent_release_gate.cli import build_parser, main def test_parser_has_evaluate_command(): @@ -7,3 +10,101 @@ def test_parser_has_evaluate_command(): assert args.command == "evaluate" assert args.spec == "a.yaml" assert args.results == "b.json" + + +def test_parser_evaluate_supports_record_history(): + parser = build_parser() + args = parser.parse_args( + [ + "evaluate", + "--spec", + "a.yaml", + "--results", + "b.json", + "--record-history", + ".gate-history", + ] + ) + + assert args.command == "evaluate" + assert args.record_history == ".gate-history" + + +def test_parser_has_trend_command(): + parser = build_parser() + args = parser.parse_args(["trend", "--history", ".gate-history", "--window", "8"]) + + assert args.command == "trend" + assert args.history == ".gate-history" + assert args.window == 8 + + +def test_main_evaluate_records_history(tmp_path: Path): + spec = tmp_path / "spec.yaml" + spec.write_text( + """ +global: + minimum_pass_rate: 0.5 +cases: + - id: case_1 + expected_all: ["refund"] +""".strip(), + encoding="utf-8", + ) + + results = tmp_path / "results.json" + results.write_text( + json.dumps( + { + "cases": [ + { + "id": "case_1", + "response": "refund approved", + } + ] + } + ), + encoding="utf-8", + ) + + history_dir = tmp_path / "history" + code = main( + [ + "evaluate", + "--spec", + str(spec), + "--results", + str(results), + "--record-history", + str(history_dir), + ] + ) + + assert code == 0 + assert len(list(history_dir.glob("*.json"))) == 1 + + +def test_main_trend_fail_on_regression(tmp_path: Path): + history_dir = tmp_path / "history" + history_dir.mkdir(parents=True) + + runs = [ + ("run-1", "2026-04-01T00:00:00+00:00", 0.92), + ("run-2", "2026-04-02T00:00:00+00:00", 0.88), + ("run-3", "2026-04-03T00:00:00+00:00", 0.84), + ] + for run_id, timestamp, pass_rate in runs: + payload = { + "run_id": run_id, + "timestamp": timestamp, + "summary": { + "pass_rate": pass_rate, + "avg_latency_ms": 800.0, + "avg_cost_usd": 0.01, + "gate_passed": True, + }, + } + (history_dir / f"{run_id}.json").write_text(json.dumps(payload), encoding="utf-8") + + code = main(["trend", "--history", str(history_dir), "--fail-on-regression"]) + assert code == 1 diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 0000000..2c75b80 --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,79 @@ +import json +from pathlib import Path + +from agent_release_gate.history import analyze_history, record_history +from agent_release_gate.models import GateReport, GateSummary, ScoredCase + + +def _report(pass_rate: float, gate_passed: bool = True) -> GateReport: + return GateReport( + summary=GateSummary( + total_cases=10, + passed_cases=int(pass_rate * 10), + pass_rate=pass_rate, + avg_latency_ms=850.0, + avg_cost_usd=0.011, + gate_passed=gate_passed, + gate_reasons=["ok"], + ), + cases=[ScoredCase(id="case_1", score=0.9, passed=gate_passed, notes=["ok"])], + ) + + +def _write_run(history_dir: Path, run_id: str, timestamp: str, pass_rate: float) -> None: + payload = { + "run_id": run_id, + "timestamp": timestamp, + "summary": { + "pass_rate": pass_rate, + "avg_latency_ms": 800.0, + "avg_cost_usd": 0.01, + "gate_passed": pass_rate >= 0.8, + }, + } + (history_dir / f"{run_id}.json").write_text(json.dumps(payload), encoding="utf-8") + + +def test_record_history_writes_run_summary(tmp_path: Path): + history_dir = tmp_path / "history" + path = record_history(_report(0.9), history_dir=history_dir, run_id="run-001") + + assert path.exists() + payload = json.loads(path.read_text(encoding="utf-8")) + assert payload["run_id"] == "run-001" + assert payload["summary"]["pass_rate"] == 0.9 + + +def test_analyze_history_detects_declining_trend(tmp_path: Path): + history_dir = tmp_path / "history" + history_dir.mkdir(parents=True) + + _write_run(history_dir, "run-1", "2026-04-01T00:00:00+00:00", 0.92) + _write_run(history_dir, "run-2", "2026-04-02T00:00:00+00:00", 0.89) + _write_run(history_dir, "run-3", "2026-04-03T00:00:00+00:00", 0.86) + _write_run(history_dir, "run-4", "2026-04-04T00:00:00+00:00", 0.83) + _write_run(history_dir, "run-5", "2026-04-05T00:00:00+00:00", 0.80) + + trend = analyze_history(history_dir=history_dir, window=10) + + assert trend.pass_rate_direction == "declining" + assert trend.any_regression is True + assert trend.pass_rate_slope < 0 + + +def test_analyze_history_window_uses_most_recent_runs(tmp_path: Path): + history_dir = tmp_path / "history" + history_dir.mkdir(parents=True) + + _write_run(history_dir, "run-1", "2026-04-01T00:00:00+00:00", 0.90) + _write_run(history_dir, "run-2", "2026-04-02T00:00:00+00:00", 0.85) + _write_run(history_dir, "run-3", "2026-04-03T00:00:00+00:00", 0.80) + _write_run(history_dir, "run-4", "2026-04-04T00:00:00+00:00", 0.93) + _write_run(history_dir, "run-5", "2026-04-05T00:00:00+00:00", 0.95) + _write_run(history_dir, "run-6", "2026-04-06T00:00:00+00:00", 0.97) + + trend = analyze_history(history_dir=history_dir, window=3) + + assert len(trend.runs) == 3 + assert trend.runs[0].run_id == "run-4" + assert trend.pass_rate_direction == "improving"