feat: add cross-run pass-rate trend gating

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

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
 

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,
+    }

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