Evals in CI

# script_id: day_047_evals_in_ci/flake_odds
# Stdlib only. Env: none. Services: none.
def p_spurious(n_cases: int, flake_rate: float) -> float:
    """Chance that at least one of n healthy, independent cases fails by bad luck."""
    return 1 - (1 - flake_rate) ** n_cases


for n in (1, 5, 20, 100):
    print(f"{n:>3} cases, each 5% flaky -> {p_spurious(n, 0.05):5.1%} of CI runs show a false failure")

  1 cases, each 5% flaky ->  5.0% of CI runs show a false failure
  5 cases, each 5% flaky -> 22.6% of CI runs show a false failure
 20 cases, each 5% flaky -> 64.2% of CI runs show a false failure
100 cases, each 5% flaky -> 99.4% of CI runs show a false failure

# script_id: day_047_evals_in_ci/minimal_gate
# Install: uv add pytest · Env: none · Services: none. Replays a recorded run: no API calls, $0.
import sys

import pytest

RUN = {"qa-001": True, "qa-002": True, "qa-004": False, "qa-u01": True}  # stand-in for runs/ask_v1_fix.jsonl
BASELINE = {"pass_rate": 0.75, "max_drop": 0.05, "must_pass": ["qa-u01"]}  # stand-in for baseline.json


@pytest.mark.eval
def test_no_regression() -> None:
    broken = [c for c in BASELINE["must_pass"] if not RUN.get(c, False)]
    assert not broken, f"must-pass cases failed: {broken}"
    rate, floor = sum(RUN.values()) / len(RUN), BASELINE["pass_rate"] - BASELINE["max_drop"]
    assert rate >= floor, f"pass rate {rate:.2f} is below baseline {BASELINE['pass_rate']:.2f} - allowed drop"


if __name__ == "__main__":  # `-o markers=...` registers the marker; in a project, use pyproject.toml
    sys.exit(pytest.main([__file__, "-q", "-m", "eval", "-p", "no:cacheprovider", "-o", "markers=eval: model evals"]))

.                                                                        [100%]
1 passed in 0.01s

E       AssertionError: pass rate 0.75 is below baseline 0.90 - allowed drop
E       assert 0.75 >= 0.85
FAILED minimal_gate.py::test_no_regression - AssertionError: pass rate 0.75 i...

# pyproject.toml
[tool.pytest.ini_options]
markers = ["eval: model-backed eval suites (cost money; selected with -m eval)"]

# script_id: day_047_evals_in_ci/test_eval_gate
# tests/test_eval_gate.py: Atlas R2's eval gate, with atlas/evals/ci.py inlined (self-contained lesson).
# Install: uv add "openai>=3.46,<4" pydantic pytest
# Env: OPENAI_API_KEY (unset: offline fake locally, cache-only replay in CI), ATLAS_MODEL, ATLAS_PROMPT,
#      EVAL_SUITE (fast|full), EVAL_OUT (where report/cache files go), ORBIT_DATA. Services: none.
# Run: pytest -m eval tests/test_eval_gate.py   or   python tests/test_eval_gate.py --help
from __future__ import annotations

import argparse, hashlib, json, os, random, re, sys, time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Callable, Literal, Sequence

import pytest
from openai import OpenAI
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
JUDGE_MODEL = os.environ.get("ATLAS_JUDGE_MODEL", "gpt-6-luna")  # Day 45's judge model
PROMPT_VERSION = os.environ.get("ATLAS_PROMPT", "v1")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
OUT = Path(os.environ.get("EVAL_OUT", "."))      # baseline.json, report.json, pr_comment.md, .eval_cache/
RUBRIC_VERSION = "grounded-v2"                     # bump whenever Day 46 re-validates the judge
PRICE_IN, PRICE_OUT = 0.10 / 1e6, 0.50 / 1e6       # gpt-6-luna $/token (as of 2026-11, verify)
LIVE = bool(os.environ.get("OPENAI_API_KEY"))
CACHE_ONLY = not LIVE and bool(os.environ.get("CI"))  # fork PRs get no secrets: replay the cache only
client = OpenAI() if LIVE else None
ABSTAIN = "I don't know based on the Orbit docs."
PROMPTS = {  # compact stand-in for Day 21's registry; "v1-seeded" is today's deliberate regression
    "v1": "You are Atlas. Answer only from the provided docs. If they don't contain the answer, "
          f"reply exactly: {ABSTAIN} Cite doc ids in [brackets].",
    "v1-seeded": "You are Atlas. Answer the question helpfully. Cite doc ids in [brackets].",
}
COST = {"usd": 0.0}
LATENCIES: list[float] = []


def load_prompt(name: str, version: str) -> str:
    return PROMPTS[version]  # Day 21's load_prompt() reads prompts/<name>/<version>.txt instead

# script_id: day_047_evals_in_ci/test_eval_gate
@dataclass(frozen=True)
class EvalCase:  # compact Day 43 EvalCase
    id: str
    question: str
    citations: tuple[str, ...]
    slice: str  # "2.x" | "3.x" | "all" | "unanswerable"


UNANSWERABLE = [  # stand-ins for Day 43's unanswerable slice: Orbit has no Kafka, GraphQL or MongoDB
    ("qa-u01", "How do I configure Orbit's Kafka broker?"),
    ("qa-u02", "What is the endpoint of Orbit's GraphQL API?"),
    ("qa-u03", "How do I store job results in MongoDB?"),
    ("qa-u04", "Which Kafka topic does Orbit use for the dead-letter queue?"),
]


def load_cases() -> list[EvalCase]:
    """Day 43's ask_v1.jsonl if you have it; here the shipped qa_gold.jsonl + 4 unanswerable cases."""
    rows = [json.loads(line) for line in (DATA_DIR / "eval/qa_gold.jsonl").open(encoding="utf-8")]
    cases = [EvalCase(r["id"], r["question"], tuple(r["citations"]), r["applies_to"]) for r in rows]
    return cases + [EvalCase(i, q, (), "unanswerable") for i, q in UNANSWERABLE]


WORD = re.compile(r"[a-z0-9_.-]{3,}")
STOP = set("the and what how does which when for with orbit use can should are from this that "
           "into why you after its there have has was not but will than then they".split())


def terms(text: str) -> set[str]:
    return {w.strip(".-") for w in WORD.findall(text.lower())} - STOP - {""}


class Index:  # keyword stand-in for Days 26-28's hybrid index: one chunk per "## " section
    def __init__(self, docs_dir: Path) -> None:
        self.chunks = [(p.stem, sec, terms(sec)) for p in sorted(docs_dir.glob("*.md"))
                       for sec in p.read_text(encoding="utf-8").split("\n## ") if sec.strip()]

    def search(self, query: str, k: int = 3) -> list[tuple[str, str]]:
        q = terms(query)
        ranked = sorted(self.chunks, key=lambda c: -len(q & c[2]))  # stable sort: ties keep file order
        return [(doc, text) for doc, text, _ in ranked[:k]]

# script_id: day_047_evals_in_ci/test_eval_gate
def fake_llm(prompt: str, question: str, context: list[tuple[str, str]], attempt: int) -> str:
    """Offline stand-in for the model: deterministic, with the two behaviours this lesson needs."""
    q = terms(question)
    covered = len(q & set().union(*(terms(t) for _, t in context))) / max(len(q), 1)
    doc, text = context[0]
    best = max(re.split(r"(?<=[.!?])\s+", text), key=lambda s: len(q & terms(s)))
    best = re.sub(r"[\[\]`*#]", "", best).strip()  # drop markdown so a stray [x] isn't read as a citation
    if covered < 0.7:  # the docs don't really cover the question
        if "only from the provided docs" in prompt:
            return ABSTAIN
        return f"{best} Orbit also supports this through its plugin system. [{doc}]"  # invented claim
    if random.Random(f"{question}|{attempt}").random() < 0.08:
        return best  # ~8% of calls forget the citation, like a real model's run-to-run variance
    return f"{best} [{doc}]"


def generate(prompt: str, question: str, context: list[tuple[str, str]], attempt: int) -> str:
    if not LIVE:
        text = fake_llm(prompt, question, context, attempt)
        n_in, n_out = (len(prompt) + len(question) + sum(len(t) for _, t in context)) // 4, len(text) // 4
    else:
        docs = "\n\n".join(f"<doc id='{d}'>\n{t}\n</doc>" for d, t in context)
        resp = client.responses.create(model=MODEL, instructions=prompt, reasoning={"effort": "low"},
                                       input=f"{docs}\n\nQuestion: {question}")
        text, n_in, n_out = resp.output_text, resp.usage.input_tokens, resp.usage.output_tokens
    COST["usd"] += n_in * PRICE_IN + n_out * PRICE_OUT  # the fake's tokens are len/4 estimates
    return text

# script_id: day_047_evals_in_ci/test_eval_gate
class Verdict(BaseModel):  # Day 45's binary verdict
    passed: bool
    reason: str


def grounded(case: EvalCase, gen: Gen) -> bool:
    """Day 45 judge (validated on Day 46) when live; offline, a word-overlap stand-in."""
    if gen.text.strip() == ABSTAIN:
        return True
    key = CACHE.key(kind="judge", model=JUDGE_MODEL, rubric=RUBRIC_VERSION, answer=gen.text, ctx=sha(gen.context))
    hit = CACHE.get(key) if (LIVE or CACHE_ONLY) else None
    if hit is None and not LIVE:  # offline (or unrecorded in CI): word-overlap stand-in, not a judge
        claims = terms(re.sub(r"\[[^\]]*\]", "", gen.text))
        return len(claims & terms(gen.context)) / max(len(claims), 1) >= 0.9
    if hit is None:
        r = client.responses.parse(
            model=JUDGE_MODEL, text_format=Verdict, reasoning={"effort": "none"}, temperature=0,
            instructions="Grade ONE criterion: is every claim in ANSWER supported by CONTEXT? "
                         "Return passed=false if any claim is missing from CONTEXT.",
            input=f"CONTEXT:\n{gen.context}\n\nANSWER:\n{gen.text}")
        COST["usd"] += r.usage.input_tokens * PRICE_IN + r.usage.output_tokens * PRICE_OUT
        hit = {"passed": bool(r.output_parsed and r.output_parsed.passed)}
        CACHE.put(key, hit)
    return hit["passed"]


def abstains_correctly(case: EvalCase, gen: Gen) -> bool:  # Day 44 code check
    return (gen.text.strip() == ABSTAIN) == (case.slice == "unanswerable")


def cites_real_doc(case: EvalCase, gen: Gen) -> bool:  # Day 44 code check
    cited = set(re.findall(r"\[([a-z0-9-]+)\]", gen.text))
    return gen.text.strip() == ABSTAIN or (bool(cited) and cited <= DOC_IDS)


CHECKS: dict[str, Callable[[EvalCase, Gen], bool]] = {
    "grounded": grounded, "abstains_correctly": abstains_correctly, "cites_real_doc": cites_real_doc}

# script_id: day_047_evals_in_ci/test_eval_gate
@dataclass
class CaseOutcome:
    case_id: str
    runs: list[bool]
    passed: bool
    flaky: bool
    criteria: dict[str, bool] = field(default_factory=dict)  # from the deciding (last) run
    evaluated: bool = True


def run_with_repeats(case: EvalCase, predict: Callable[[EvalCase, int], Gen | None],
                     checks: dict[str, Callable[[EvalCase, Gen], bool]], *, k: int = 3) -> CaseOutcome:
    runs: list[bool] = []
    last: dict[str, bool] = {}
    for attempt in range(k):
        gen = predict(case, attempt)
        if gen is None:  # cache-only mode and nothing recorded for this case
            return CaseOutcome(case.id, [], False, False, evaluated=False)
        last = {name: check(case, gen) for name, check in checks.items()}
        runs.append(all(last.values()))
        if runs[0] or max(runs.count(True), runs.count(False)) > k // 2:
            break  # passes are never rerun; failures rerun until a majority is reached
    return CaseOutcome(case.id, runs, runs.count(True) > runs.count(False), len(set(runs)) > 1, last)

# script_id: day_047_evals_in_ci/flake_knob
# Stdlib only. Simulates 1,000 CI runs of a 20-case suite: 19 healthy cases that flake 5% of
# the time, plus 1 case that is truly broken. Compares k=1 (no reruns) with k=3 (rerun failures).
import random


def decide(healthy: bool, rng: random.Random, k: int, flake: float = 0.05) -> tuple[bool, bool]:
    """Rerun only failures, up to k runs in total; the majority decides. Returns (passed, flaky)."""
    runs = [healthy and rng.random() >= flake]
    while not runs[0] and len(runs) < k and max(runs.count(True), runs.count(False)) <= k // 2:
        runs.append(healthy and rng.random() >= flake)
    return runs.count(True) > runs.count(False), len(set(runs)) > 1


for k in (1, 3):
    rng = random.Random(7)
    false_blocks = caught = flaky = 0
    for _ in range(1000):
        healthy = [decide(True, rng, k) for _ in range(19)]
        false_blocks += any(not passed for passed, _ in healthy)
        caught += not decide(False, rng, k)[0]
        flaky += sum(f for _, f in healthy)
    print(f"k={k}: blocked by a healthy case in {false_blocks / 10:.1f}% of runs · "
          f"broken case caught {caught / 10:.0f}% · {flaky / 1000:.2f} flaky reports per run")

k=1: blocked by a healthy case in 63.7% of runs · broken case caught 100% · 0.00 flaky reports per run
k=3: blocked by a healthy case in 8.1% of runs · broken case caught 100% · 0.92 flaky reports per run

# script_id: day_047_evals_in_ci/cache_key_knob
# Stdlib only. Env: none. Services: none. Why every input that changes the output belongs in the key.
import hashlib
import json


def key(**parts: object) -> str:
    return hashlib.sha256(json.dumps(parts, sort_keys=True).encode("utf-8")).hexdigest()[:12]


v1 = "Answer only from the provided docs. Cite the doc ID."
seeded = "Answer helpfully. Cite the doc ID."  # the grounding rule is gone
base = {"model": "gpt-6-luna", "question": "Does Orbit support Kafka?", "attempt": 0}

print("full key,   v1    :", key(**base, prompt=v1))
print("full key,   seeded:", key(**base, prompt=seeded), "<- miss: the edited prompt really runs")
print("no prompt,  v1    :", key(**base))
print("no prompt,  seeded:", key(**base), "<- hit: v1's old answer is replayed")

full key,   v1    : d182422c6459
full key,   seeded: ce6142796f80 <- miss: the edited prompt really runs
no prompt,  v1    : fda5743e1216
no prompt,  seeded: fda5743e1216 <- hit: v1's old answer is replayed

# script_id: day_047_evals_in_ci/test_eval_gate
class EvalCache:
    """Content-addressed: each result is stored under a hash of every input that can change it."""

    def __init__(self, path: Path) -> None:
        self.path = path
        path.mkdir(parents=True, exist_ok=True)
        self.hits = self.misses = 0

    def key(self, **parts: object) -> str:
        blob = json.dumps(parts, sort_keys=True, ensure_ascii=False)  # sorted: same inputs, same key
        return hashlib.sha256(blob.encode("utf-8")).hexdigest()

    def get(self, key: str) -> dict | None:
        f = self.path / f"{key}.json"
        if f.exists():
            self.hits += 1
            return json.loads(f.read_text(encoding="utf-8"))
        self.misses += 1
        return None

    def put(self, key: str, value: dict) -> None:
        tmp = self.path / f"{key}.tmp"
        tmp.write_text(json.dumps(value), encoding="utf-8")
        tmp.replace(self.path / f"{key}.json")  # atomic rename: a killed job never leaves half a file


def sha(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]

# script_id: day_047_evals_in_ci/test_eval_gate
INDEX = Index(DATA_DIR / "docs")
CACHE = EvalCache(OUT / ".eval_cache")
DOC_IDS = {p.stem for p in (DATA_DIR / "docs").glob("*.md")}
INDEX_VERSION = sha("".join(text for _, text, _ in INDEX.chunks))  # re-chunk or edit docs: new keys


@dataclass
class Gen:  # one generation, exactly what the cache stores
    text: str
    context: str
    latency_ms: float


def predict(case: EvalCase, attempt: int) -> Gen | None:
    """Mini-R2 ask() behind the cache. Each attempt has its own key, so reruns are real reruns."""
    prompt = load_prompt("ask", PROMPT_VERSION)
    key = CACHE.key(kind="ask", model=MODEL, prompt=sha(prompt), question=case.question,
                    index=INDEX_VERSION, top_k=3, attempt=attempt)
    if (hit := CACHE.get(key)) is None:
        if CACHE_ONLY:
            return None  # no key and nothing recorded: report "not evaluated", never guess
        context = INDEX.search(case.question)
        t0 = time.perf_counter()
        text = generate(prompt, case.question, context, attempt)
        ms = (time.perf_counter() - t0) * 1000 if LIVE else 400 + len(text) * 2  # fake latency model
        hit = asdict(Gen(text, "\n\n".join(t for _, t in context), round(ms, 1)))
        CACHE.put(key, hit)
    LATENCIES.append(hit["latency_ms"])
    return Gen(**hit)

# script_id: day_047_evals_in_ci/test_eval_gate
def select_suite(cases: Sequence[EvalCase], suite: Literal["fast", "full"], *, n: int = 20,
                 seed: int = 42, must_pass: Sequence[str] = ()) -> list[EvalCase]:
    """Full = everything. Fast = must-pass cases + a stratified, seeded sample of the rest."""
    if suite == "full":
        return list(cases)
    pinned = [c for c in cases if c.id in must_pass]
    rest = [c for c in cases if c.id not in must_pass]
    slices: dict[str, list[EvalCase]] = {}
    for c in rest:
        slices.setdefault(c.slice, []).append(c)
    rng, budget, picked = random.Random(seed), max(n - len(pinned), 0), []
    for name in sorted(slices):  # proportional share per slice, at least one each
        group = slices[name]
        share = max(1, round(budget * len(group) / len(rest)))
        picked += rng.sample(group, min(share, len(group)))
    return pinned + picked

# script_id: day_047_evals_in_ci/sample_knob
# Stdlib only. Env: none. Services: none. Slice sizes match today's 34 cases.
import random

SLICES = {"2.x": 3, "3.x": 14, "all": 13, "unanswerable": 4}
CASES = [(name, i) for name, size in SLICES.items() for i in range(size)]


def stratified(rng: random.Random, n: int) -> list[tuple[str, int]]:
    picked = []
    for name, size in SLICES.items():  # proportional share, at least one per slice
        group = [c for c in CASES if c[0] == name]
        picked += rng.sample(group, max(1, round(n * size / len(CASES))))
    return picked


for label, pick in (("random", lambda r: r.sample(CASES, 12)), ("stratified", lambda r: stratified(r, 12))):
    rng = random.Random(0)
    missed = sum(all(s != "unanswerable" for s, _ in pick(rng)) for _ in range(10_000))
    print(f"{label:>10}: {missed / 100:4.1f}% of 12-case samples contain no unanswerable case")

    random: 15.3% of 12-case samples contain no unanswerable case
stratified:  0.0% of 12-case samples contain no unanswerable case

# script_id: day_047_evals_in_ci/test_eval_gate
@dataclass
class EvalReport:
    suite: str
    model: str
    prompt_version: str
    per_criterion: dict[str, float]
    outcomes: list[CaseOutcome]
    cost_usd: float
    p95_ms: float

    def save(self, path: Path) -> None:
        path.write_text(json.dumps(asdict(self), indent=1), encoding="utf-8")

    @classmethod
    def load(cls, path: Path) -> EvalReport:
        d = json.loads(path.read_text(encoding="utf-8"))
        return cls(**{**d, "outcomes": [CaseOutcome(**o) for o in d["outcomes"]]})


def scores(outcomes: Sequence[CaseOutcome], names: Sequence[str]) -> dict[str, float]:
    done = [o for o in outcomes if o.evaluated]
    return {n: round(sum(o.criteria[n] for o in done) / max(len(done), 1), 3) for n in names}


def run_suite(cases: Sequence[EvalCase], *, suite: str, k: int = 3) -> EvalReport:
    cost0 = COST["usd"]
    LATENCIES.clear()
    CACHE.hits = CACHE.misses = 0  # the PR comment reports this run's hit rate
    outcomes = [run_with_repeats(c, predict, CHECKS, k=k) for c in cases]
    lat = sorted(LATENCIES) or [0.0]
    return EvalReport(suite, MODEL, PROMPT_VERSION, scores(outcomes, list(CHECKS)), outcomes,
                      round(COST["usd"] - cost0, 5), lat[int(0.95 * (len(lat) - 1))])

# script_id: day_047_evals_in_ci/test_eval_gate
@dataclass
class ReportDiff:
    base: EvalReport
    cur: EvalReport
    rows: list[tuple[str, float, float]]  # criterion, main, PR (main re-scored on the PR's cases)
    newly_failing: list[str]
    flaky: list[str]
    not_evaluated: list[str]


def diff_reports(baseline: EvalReport, current: EvalReport) -> ReportDiff:
    ids = {o.case_id for o in current.outcomes if o.evaluated}
    base_same = [o for o in baseline.outcomes if o.case_id in ids]  # compare like with like
    names = [n for n in current.per_criterion if n in baseline.per_criterion]  # new criteria wait for main
    base_scores = scores(base_same, names)
    was_ok = {o.case_id for o in base_same if o.passed}
    return ReportDiff(
        baseline, current,
        [(n, base_scores[n], current.per_criterion[n]) for n in names],
        [o.case_id for o in current.outcomes if o.evaluated and not o.passed and o.case_id in was_ok],
        [o.case_id for o in current.outcomes if o.flaky],
        [o.case_id for o in current.outcomes if not o.evaluated])


def gate(diff: ReportDiff, *, max_drop: float, must_pass: Sequence[str]) -> list[str]:
    """Reasons to block; an empty list means the gate passes."""
    failed = {o.case_id for o in diff.cur.outcomes if o.evaluated and not o.passed}
    reasons = []
    if bad := sorted(failed & set(must_pass)):
        reasons.append(f"{len(bad)} must-pass case(s) failed ({', '.join(bad)})")
    reasons += [f"{n} {b:.2f} -> {v:.2f} (allowed drop {max_drop:.2f})"
                for n, b, v in diff.rows if b - v > max_drop]
    return reasons

# script_id: day_047_evals_in_ci/test_eval_gate
def render_pr_comment(diff: ReportDiff, reasons: Sequence[str] = ()) -> str:
    c, b = diff.cur, diff.base
    out = [f"### Atlas eval gate: {'BLOCKED' if reasons else 'passed'}",
           f"`{c.suite}` suite, {len(c.outcomes)} cases ({len(MUST_PASS)} must-pass) · model `{c.model}` "
           f"(main `{b.model}`) · prompt `{c.prompt_version}` (main `{b.prompt_version}`)", "",
           "| criterion | main | PR | change |", "|---|---|---|---|"]
    out += [f"| {n} | {bv:.2f} | {v:.2f} | {v - bv:+.2f} |" for n, bv, v in diff.rows]
    out += ["", *[f"- **Blocking:** {r}" for r in reasons],
            f"- Newly failing: {', '.join(diff.newly_failing) or 'none'}",
            f"- Flaky (decided by rerun; reported, not blocking): {', '.join(diff.flaky) or 'none'}",
            f"- Cost ${c.cost_usd:.4f} (cache {CACHE.hits} hits / {CACHE.misses} misses) · "
            f"p95 {c.p95_ms:.0f} ms (main {b.p95_ms:.0f} ms)"]
    if diff.not_evaluated:
        out.append(f"- Not evaluated (no API key, nothing cached): {len(diff.not_evaluated)} cases")
    return "\n".join(out) + "\n"


MUST_PASS = ("qa-u01", "qa-u02")  # abstention cases the grounding rule exists for
FAST_N = int(os.environ.get("EVAL_FAST_N", "12"))  # use 20 with Day 43's 100-case set


def evaluate(suite: Literal["fast", "full"]) -> tuple[EvalReport, list[str]]:
    cases = select_suite(load_cases(), suite, n=FAST_N, must_pass=MUST_PASS)
    report = run_suite(cases, suite=suite)
    report.save(OUT / "report.json")
    diff = diff_reports(EvalReport.load(OUT / "baseline.json"), report)
    reasons = gate(diff, max_drop=0.15 if suite == "fast" else 0.05, must_pass=MUST_PASS)
    (OUT / "pr_comment.md").write_text(render_pr_comment(diff, reasons), encoding="utf-8")
    return report, reasons

# script_id: day_047_evals_in_ci/test_eval_gate
@pytest.mark.eval
def test_r2_gate() -> None:
    if not (OUT / "baseline.json").exists():
        pytest.skip("no baseline.json: run `python tests/test_eval_gate.py --write-baseline` on main")
    _, reasons = evaluate(os.environ.get("EVAL_SUITE", "fast"))
    assert not reasons, "gate failed: " + "; ".join(reasons)


def write_baseline() -> None:
    if os.environ.get("GITHUB_REF", "refs/heads/main") != "refs/heads/main":
        sys.exit("refusing to write baseline.json outside main")  # a PR must never move its own goalposts
    report = run_suite(load_cases(), suite="full")
    report.save(OUT / "baseline.json")
    print(f"baseline.json: {report.per_criterion} · ${report.cost_usd:.4f} · {len(report.outcomes)} cases")


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description="Atlas R2 eval gate")
    ap.add_argument("--suite", choices=["fast", "full"], default=os.environ.get("EVAL_SUITE", "fast"))
    ap.add_argument("--write-baseline", action="store_true", help="main only: full run -> baseline.json")
    args = ap.parse_args()
    if args.write_baseline or not (OUT / "baseline.json").exists():
        write_baseline()
    _, reasons = evaluate(args.suite)
    print((OUT / "pr_comment.md").read_text(encoding="utf-8"))
    sys.exit(1 if reasons else 0)

python tests/test_eval_gate.py

baseline.json: {'grounded': 1.0, 'abstains_correctly': 0.882, 'cites_real_doc': 0.971} · $0.0059 · 34 cases
### Atlas eval gate: passed
`fast` suite, 12 cases (2 must-pass) · model `gpt-6-luna` (main `gpt-6-luna`) · prompt `v1` (main `v1`)

| criterion | main | PR | change |
|---|---|---|---|
| grounded | 1.00 | 1.00 | +0.00 |
| abstains_correctly | 0.83 | 0.83 | +0.00 |
| cites_real_doc | 1.00 | 1.00 | +0.00 |

- Newly failing: none
- Flaky (decided by rerun; reported, not blocking): qa-011, qa-010
- Cost $0.0000 (cache 18 hits / 0 misses) · p95 1432 ms (main 1432 ms)

# .github/workflows/evals.yml
name: evals
on:
  pull_request:
    types: [opened, synchronize, reopened, labeled]
  schedule:
    - cron: "17 3 * * *"          # nightly full suite on main
  workflow_dispatch:
    inputs:
      model:
        description: "Candidate model id (runs the full suite against main's baseline)"
        type: string
        default: gpt-6-luna

permissions:
  contents: read
  pull-requests: write            # gh pr comment needs this

concurrency:
  group: evals-${{ github.ref }}
  cancel-in-progress: true        # a new push cancels the old, half-paid run

jobs:
  evals:
    runs-on: ubuntu-latest
    timeout-minutes: 20
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}   # empty on fork PRs -> cache-only mode
      ATLAS_MODEL: ${{ inputs.model || 'gpt-6-luna' }}
      EVAL_SUITE: ${{ (github.event_name == 'pull_request' && !contains(github.event.pull_request.labels.*.name, 'full-evals')) && 'fast' || 'full' }}
      EVAL_OUT: evals
      ORBIT_DATA: datasets/orbit
    steps:
      - uses: actions/checkout@v7
      - uses: astral-sh/setup-uv@v10.3.3
        with:
          enable-cache: true
      - uses: actions/cache@v6
        with:
          path: evals/.eval_cache
          key: eval-cache-${{ github.run_id }}   # unique, so every run saves its new entries
          restore-keys: eval-cache-               # restore the most recent cache
      - run: uv sync --locked --dev
      - name: Eval gate
        run: uv run pytest -m eval tests/test_eval_gate.py
      - name: Job summary
        if: always()
        run: if [ -f evals/pr_comment.md ]; then cat evals/pr_comment.md >> "$GITHUB_STEP_SUMMARY"; fi
      - name: PR comment
        if: always() && github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository
        env:
          GH_TOKEN: ${{ github.token }}
        run: |
          if [ -f evals/pr_comment.md ]; then
            gh pr comment "${{ github.event.pull_request.number }}" --body-file evals/pr_comment.md --edit-last --create-if-none
          fi

ATLAS_PROMPT=v1-seeded python tests/test_eval_gate.py

### Atlas eval gate: BLOCKED
`fast` suite, 12 cases (2 must-pass) · model `gpt-6-luna` (main `gpt-6-luna`) · prompt `v1-seeded` (main `v1`)

| criterion | main | PR | change |
|---|---|---|---|
| grounded | 1.00 | 0.92 | -0.08 |
| abstains_correctly | 0.83 | 0.75 | -0.08 |
| cites_real_doc | 1.00 | 1.00 | +0.00 |

- **Blocking:** 2 must-pass case(s) failed (qa-u01, qa-u02)
- Newly failing: qa-u01, qa-u02
- Flaky (decided by rerun; reported, not blocking): qa-011, qa-010
- Cost $0.0032 (cache 0 hits / 19 misses) · p95 4742 ms (main 1432 ms)

ATLAS_PROMPT=v1-seeded python tests/test_eval_gate.py --suite full

### Atlas eval gate: BLOCKED
`full` suite, 34 cases (2 must-pass) · model `gpt-6-luna` (main `gpt-6-luna`) · prompt `v1-seeded` (main `v1`)

| criterion | main | PR | change |
|---|---|---|---|
| grounded | 1.00 | 0.88 | -0.12 |
| abstains_correctly | 0.88 | 0.88 | +0.00 |
| cites_real_doc | 0.97 | 0.97 | +0.00 |

- **Blocking:** 2 must-pass case(s) failed (qa-u01, qa-u02)
- **Blocking:** grounded 1.00 -> 0.88 (allowed drop 0.05)
- Newly failing: qa-u01, qa-u02, qa-u04
- Flaky (decided by rerun; reported, not blocking): qa-010, qa-011, qa-013
- Cost $0.0035 (cache 19 hits / 28 misses) · p95 1722 ms (main 1432 ms)