Your First Eval Set

# script_id: day_020_first_eval_set/schemas
# schemas.py: compact copy of Day 15's atlas/schemas.py + atlas/data.py, trimmed for Day 20.
# Install: uv add "openai>=3.46,<4" pydantic
# Env: ORBIT_DATA (optional, path to datasets/orbit). No API key needed for this file.
import json
import os
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
Component = Literal["core", "scheduler", "flows", "broker-redis", "broker-postgres", "broker-sqs",
                    "cli", "dashboard", "retries", "results", "observability", "auth", "docs"]


class Comment(BaseModel):
    author: str
    body: str


class Issue(BaseModel):
    """One row of issues/issues.jsonl. Only the fields the extractor reads; the rest are ignored."""
    id: int
    title: str
    body: str
    comments: list[Comment]

# script_id: day_020_first_eval_set/schemas
class IssueRecord(BaseModel):
    """The facts Atlas extracts from one issue. Fields mirror eval/extraction_gold.jsonl."""
    model_config = ConfigDict(extra="forbid")

    type: Literal["bug", "feature", "question", "docs"] = Field(description="bug, feature, question or docs")
    components: list[Component] = Field(description="Every Orbit component the issue is about")
    severity: Literal["S1", "S2", "S3", "S4"] | None = Field(
        description="Bugs only: S1 outage or data loss, S2 major breakage, S3 has a workaround, "
                    "S4 cosmetic. null otherwise")
    orbit_version: str | None = Field(description="Full Orbit version, e.g. 3.1.2. null when not stated")
    python_version: str | None = Field(description="Python version as written, e.g. 3.12.6. null when not stated")
    error_codes: list[str] = Field(description="ORB-NNNN codes quoted in the issue or its comments")
    summary: str = Field(description="One or two English sentences: symptom, trigger, version")


def load_jsonl(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def load_issues(path: Path | None = None) -> list[Issue]:
    return [Issue.model_validate(row) for row in load_jsonl(path or DATA_DIR / "issues" / "issues.jsonl")]


if __name__ == "__main__":
    gold = load_jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl")
    print(f"{len(load_issues())} issues · {len(gold)} gold records")

240 issues · 30 gold records

# script_id: day_020_first_eval_set/extract
# extract.py: compact copy of Day 16-17's atlas/extract.py (extract_issue + reliable_extract).
# Install: uv add "openai>=3.46,<4" pydantic
# Env: OPENAI_API_KEY (required), ATLAS_MODEL, ATLAS_EFFORT (optional). Imports schemas.py.
import os
from dataclasses import dataclass, field

from openai import OpenAI
from pydantic import ValidationError

from schemas import Issue, IssueRecord

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
EFFORT = os.environ.get("ATLAS_EFFORT", "low")      # always explicit: the model default is "medium"
# USD per 1M tokens, as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES_PER_1M = {"gpt-6-luna": {"input": 0.10, "cached_input": 0.01, "output": 0.50},
                 "gpt-6.1-terra": {"input": 1.00, "cached_input": 0.05, "output": 5.00}}
client = OpenAI(timeout=60, max_retries=3)

EXTRACT_INSTRUCTIONS = """You extract facts from GitHub issues filed against Orbit, an open-source
Python background-jobs library. Fill every field of the schema from the issue text only.

Rules:
- Use null when a fact is not stated. Never guess a version from dates or context.
- severity is for bugs only; use null for features, questions and docs.
- error_codes: every ORB-NNNN code that appears anywhere, including logs and comments. [] if none.
- components: usually one; list two only when the issue really spans both.
- summary: one or two sentences in English, even when the issue is in another language.

Component glossary:
core = queues, workers, leases, job execution | scheduler = cron schedules and delays
flows = Flows, steps and step state | broker-redis, broker-postgres, broker-sqs = that broker
cli = the orbit command | dashboard = the web UI | retries = retry policy and backoff
results = the result backend | observability = metrics and tracing
auth = dashboard login and OIDC | docs = the documentation itself"""

# script_id: day_020_first_eval_set/extract
class CostTracker:
    """Day 12's tracker, reduced to a running total."""
    def __init__(self) -> None:
        self.calls, self.total_usd = 0, 0.0

    def record(self, usage, model: str = MODEL) -> float:
        p, cached = PRICES_PER_1M[model], usage.input_tokens_details.cached_tokens
        cost = ((usage.input_tokens - cached) * p["input"] + cached * p["cached_input"]
                + usage.output_tokens * p["output"]) / 1_000_000   # reasoning is inside output_tokens; cache-write surcharge omitted (Day 18)
        self.calls, self.total_usd = self.calls + 1, self.total_usd + cost
        return cost


tracker = CostTracker()


class ExtractionError(Exception):
    def __init__(self, reason: str, detail: str = "") -> None:
        super().__init__(f"{reason}: {detail}")
        self.reason = reason


def issue_to_input(issue: Issue, max_chars: int = 12_000) -> str:
    """Title, body and comments; never the labels (they're the answer key). Day 16 trims by tokens."""
    parts = [f"# {issue.title}", issue.body] + [f"--- comment by {c.author} ---\n{c.body}" for c in issue.comments]
    return "\n\n".join(parts)[:max_chars]


def extract_issue(issue: Issue, *, client=client, model: str = MODEL, feedback: str = "") -> IssueRecord:
    try:
        resp = client.responses.parse(model=model, instructions=EXTRACT_INSTRUCTIONS,
                                      input=issue_to_input(issue) + feedback, text_format=IssueRecord,
                                      reasoning={"effort": EFFORT}, max_output_tokens=25_000)
    except ValidationError as err:            # strict mode fixes the shape, so this is cut-off JSON
        raise ExtractionError("incomplete", "truncated JSON") from err
    tracker.record(resp.usage, model)          # billed even when the result is unusable
    if resp.status == "incomplete":
        raise ExtractionError("incomplete", str(resp.incomplete_details.reason))
    if resp.output_parsed is None:            # a refusal, or no final answer (Day 16)
        raise ExtractionError("refusal")
    return resp.output_parsed

# script_id: day_020_first_eval_set/extract
def semantic_errors(record: IssueRecord, issue: Issue) -> list[str]:
    """Day 17's rules, compact: quoted facts must appear in the text; severity is for bugs."""
    text = issue_to_input(issue)
    errors = [f"error code {c} is not in the issue" for c in record.error_codes if c not in text]
    for name in ("orbit_version", "python_version"):
        value = getattr(record, name)
        if value and value not in text:
            errors.append(f"{name} {value} is not stated in the issue; use null")
    if record.severity and record.type != "bug":
        errors.append("severity is for bugs only; use null")
    if len(record.summary) > 300:
        errors.append("summary is longer than 300 characters")
    return errors


@dataclass
class ExtractResult:
    record: IssueRecord | None
    attempts: int = 0
    errors: list[str] = field(default_factory=list)
    cost_usd: float = 0.0
    needs_review: bool = False


def reliable_extract(issue: Issue, *, client=client, max_attempts: int = 3,
                     max_usd: float = 0.01, model: str = MODEL) -> ExtractResult:
    start, result, feedback = tracker.total_usd, ExtractResult(None), ""
    while result.attempts < max_attempts and tracker.total_usd - start < max_usd:
        result.attempts += 1
        try:
            record = extract_issue(issue, client=client, model=model, feedback=feedback)
        except ExtractionError as err:
            result.errors = [str(err)]
            if err.reason == "refusal":
                break                          # re-asking a refusal rarely helps (Day 17)
            continue
        result.record, result.errors = record, semantic_errors(record, issue)
        if not result.errors:
            break
        feedback = "\n\nYour previous answer broke these rules. Fix them:\n- " + "\n- ".join(result.errors)
    result.cost_usd = tracker.total_usd - start
    result.needs_review = result.record is None or bool(result.errors)
    return result

# script_id: day_020_first_eval_set/predictors
# predictors.py: interchangeable predict functions, case dict -> record.
# Install: uv add "openai>=3.46,<4" pydantic · Env: ORBIT_DATA; OPENAI_API_KEY for llm_predict only
import re

from schemas import Issue, IssueRecord, load_issues

ISSUES: dict[int, Issue] = {issue.id: issue for issue in load_issues()}
CODE_RE = re.compile(r"\bORB-\d{4}\b")
PY_RE = re.compile(r"(?i)\bpython\s*v?(3\.\d+(?:\.\d+)?)")
ORBIT_RE = re.compile(r"(?i)(?:\borbit\s*v?|^)(\d+\.\d+\.\d+)")
TITLE_WORDS = {"broker-redis": "redis", "broker-postgres": "postgres", "broker-sqs": "sqs",
               "scheduler": "cron", "flows": "flow", "dashboard": "dashboard", "auth": "oidc",
               "retries": "retr", "cli": "orbit ", "docs": "docs"}


def keyword_baseline(case: dict) -> IssueRecord:
    """A deliberately dumb extractor: regexes and title keywords. Offline, $0, never reads labels."""
    issue = ISSUES[case["issue_id"]]
    head, title = f"{issue.title}\n{issue.body}", issue.title.lower()
    everything = "\n".join([head, *(c.body for c in issue.comments)])
    kind = ("question" if title.rstrip().endswith("?") or title.startswith(("how", "can", "is", "why"))
            else "feature" if any(w in title for w in ("support", "add ", "allow", "feature"))
            else "docs" if "docs" in title or "documentation" in title else "bug")
    python = PY_RE.search(head)
    orbit = ORBIT_RE.search(head)
    return IssueRecord(
        type=kind, severity="S3" if kind == "bug" else None,       # S3 is the most common bug label
        components=[c for c, word in TITLE_WORDS.items() if word in title][:1] or ["core"],
        orbit_version=orbit.group(1) if orbit else None, python_version=python.group(1) if python else None,
        error_codes=sorted(set(CODE_RE.findall(everything))), summary=issue.title)


def llm_predict(case: dict):
    from extract import reliable_extract    # imported here: creating the client needs OPENAI_API_KEY
    return reliable_extract(ISSUES[case["issue_id"]])

# script_id: day_020_first_eval_set/minimal_eval
# The smallest useful eval: 5 gold cases, one exact check on `type`. ≈ $0.01 (as of 2026-11, verify).
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY, ORBIT_DATA
from extract import reliable_extract
from schemas import DATA_DIR, load_issues, load_jsonl

issues = {issue.id: issue for issue in load_issues()}
gold = sorted(load_jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl"), key=lambda r: r["issue_id"])[:5]
records = {case["issue_id"]: reliable_extract(issues[case["issue_id"]]).record for case in gold}

passed = 0
for case in gold:
    got = records[case["issue_id"]]                # None when reliable_extract gave up
    ok = got is not None and got.type == case["type"]
    passed += ok
    print(f"#{case['issue_id']:<4} want {case['type']:<9} got {got.type if got else '-':<9} {'PASS' if ok else 'FAIL'}")
print(f"type: {passed}/{len(gold)} pass ({passed / len(gold):.0%})")

#10   want question  got question  PASS
#25   want question  got question  PASS
#62   want bug       got bug       PASS
#63   want question  got question  PASS
#74   want feature   got feature   PASS
type: 5/5 pass (100%)

# script_id: day_020_first_eval_set/minimal_eval
# Knob 1: same cases, second run (5 more calls). Knobs 2-3: rescore run 1's records, $0.
again = {cid: reliable_extract(issues[cid]).record for cid in records}
flips = [cid for cid in records if (records[cid] and records[cid].type) != (again[cid] and again[cid].type)]
print("type changed between runs on:", flips or "none")


def rate(check) -> str:
    hits = sum(bool(check(case, records[case["issue_id"]])) for case in gold)
    return f"{hits}/{len(gold)}"


print("too strict, summary == gold summary:", rate(lambda c, r: r and r.summary == c["summary"]))
print("too loose, summary is non-empty:   ", rate(lambda c, r: r and len(r.summary) > 0))

type changed between runs on: [62]
too strict, summary == gold summary: 0/5
too loose, summary is non-empty:    5/5

# script_id: day_020_first_eval_set/checks
# checks.py: atlas/evals/checks.py. Binary, code-based checks; each returns one CheckResult. $0.
# Install: uv add pydantic · Env: none. Imports the lesson-local schemas.py for schema_valid().
import re
from dataclasses import dataclass
from typing import Any, Callable

from pydantic import BaseModel, ValidationError

from schemas import IssueRecord


@dataclass(frozen=True)
class CheckResult:
    name: str
    passed: bool
    detail: str = ""


Check = Callable[[dict, Any], CheckResult]          # (gold case, prediction) -> verdict


def _get(pred: Any, field: str) -> Any:
    """Predictions may be Pydantic records or plain dicts (replayed from a saved report)."""
    return pred.get(field) if isinstance(pred, dict) else getattr(pred, field, None)


def _norm(value: Any) -> Any:
    """One null convention for every check: None, "", "unknown" and "n/a" mean "not stated"."""
    if isinstance(value, str):
        value = value.strip().lower()
        return None if value in {"", "unknown", "n/a", "none", "null"} else value
    return value


def exact(field: str) -> Check:
    def check(case: dict, pred: Any) -> CheckResult:
        want, got = _norm(case[field]), _norm(_get(pred, field))
        return CheckResult(field, want == got, "" if want == got else f"want {want!r}, got {got!r}")
    return check

# script_id: day_020_first_eval_set/checks
def set_match(field: str) -> Check:
    def check(case: dict, pred: Any) -> CheckResult:
        want, got = set(case[field] or []), set(_get(pred, field) or [])   # None and [] both mean "none"
        detail = "" if want == got else f"missing {sorted(want - got)}, extra {sorted(got - want)}"
        return CheckResult(field, want == got, detail)
    return check


def _release(value: Any) -> tuple[int, ...] | None:
    if _norm(value) is None:
        return None
    nums = [int(n) for n in re.findall(r"\d+", str(value))][:3]
    return tuple(nums + [0] * (3 - len(nums)))       # "3.1" -> (3, 1, 0), same release as "3.1.0"


def version_match(field: str) -> Check:
    def check(case: dict, pred: Any) -> CheckResult:
        want, got = _release(case[field]), _release(_get(pred, field))
        return CheckResult(field, want == got, "" if want == got else f"want {case[field]!r}, got {_get(pred, field)!r}")
    return check


def contains(field: str, needle_field: str) -> Check:
    """Pass if the predicted `field` mentions every gold value of `needle_field` (case-insensitive)."""
    def check(case: dict, pred: Any) -> CheckResult:
        text, needles = str(_get(pred, field) or "").lower(), case[needle_field] or []
        missing = [n for n in (needles if isinstance(needles, list) else [needles]) if str(n).lower() not in text]
        return CheckResult(f"{field}_mentions_{needle_field}", not missing, f"missing {missing}" if missing else "")
    return check


def length_between(field: str, lo: int, hi: int) -> Check:
    def check(case: dict, pred: Any) -> CheckResult:
        n = len(str(_get(pred, field) or ""))
        return CheckResult(f"{field}_length", lo <= n <= hi, "" if lo <= n <= hi else f"{n} chars, want {lo}-{hi}")
    return check

# script_id: day_020_first_eval_set/checks
def schema_valid(model: type[BaseModel] = IssueRecord) -> Check:
    def check(case: dict, pred: Any) -> CheckResult:
        if pred is None:
            return CheckResult("schema", False, "no record (refusal, incomplete or over budget)")
        try:
            model.model_validate(pred if isinstance(pred, dict) else pred.model_dump())
            return CheckResult("schema", True)
        except ValidationError as err:
            return CheckResult("schema", False, f"{err.error_count()} validation errors")
    return check


if __name__ == "__main__":
    gold = {"type": "bug", "severity": None, "components": ["flows", "cli"], "orbit_version": "3.0.0",
            "error_codes": ["ORB-1502", "ORB-1301"]}
    cases = [(exact("type"), {"type": "Bug"}, True), (exact("type"), {"type": "question"}, False),
             (exact("severity"), {"severity": "unknown"}, True),            # normalizes to None
             (set_match("components"), {"components": ["cli", "flows"]}, True),   # order ignored
             (set_match("components"), {"components": ["cli"]}, False),
             (version_match("orbit_version"), {"orbit_version": "3.0"}, True),
             (version_match("orbit_version"), {"orbit_version": "3.0.1"}, False),
             (contains("summary", "error_codes"), {"summary": "ORB-1502 then the DLQ (orb-1301)"}, True),
             (length_between("summary", 20, 300), {"summary": "broken"}, False),
             (schema_valid(), None, False)]
    ok = sum(check(gold, pred).passed == want for check, pred, want in cases)
    print(f"checks self-test: {ok}/{len(cases)}")

checks self-test: 10/10

# script_id: day_020_first_eval_set/runner
# runner.py: atlas/evals/runner.py. cases -> predict -> checks -> EvalReport -> gate().
# Install: uv add "openai>=3.46,<4" pydantic · Env: ORBIT_DATA; ATLAS_MODEL/ATLAS_EFFORT recorded in meta
import json
import os
import statistics
import subprocess
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Literal

from checks import (Check, CheckResult, contains, exact, length_between, schema_valid,
                    set_match, version_match)
from schemas import DATA_DIR, load_jsonl

DEV_SIZE = 20
DEFAULT_CHECKS: list[Check] = [
    schema_valid(), exact("type"), exact("severity"), set_match("components"),
    set_match("error_codes"), version_match("orbit_version"),
    contains("summary", "error_codes"), length_between("summary", 20, 300)]


def load_cases(path: Path | None = None, *, split: Literal["dev", "heldout", "all"] = "dev") -> list[dict]:
    """Gold rows sorted by issue_id. First 20 = dev (iterate here); last 10 = held out (release only)."""
    rows = load_jsonl(path or DATA_DIR / "eval" / "extraction_gold.jsonl")
    rows.sort(key=lambda row: row["issue_id"])
    return {"dev": rows[:DEV_SIZE], "heldout": rows[DEV_SIZE:], "all": rows}[split]

# script_id: day_020_first_eval_set/runner
@dataclass
class CaseResult:
    issue_id: int
    checks: list[CheckResult]
    prediction: dict | None
    cost_usd: float = 0.0
    latency_s: float = 0.0
    error: str = ""

    @property
    def passed(self) -> bool:
        return all(c.passed for c in self.checks)


@dataclass
class EvalReport:
    results: list[CaseResult]
    meta: dict = field(default_factory=dict)

    @property
    def pass_rate(self) -> float:
        return sum(r.passed for r in self.results) / max(len(self.results), 1)

    @property
    def field_accuracy(self) -> dict[str, float]:
        names = [c.name for c in self.results[0].checks] if self.results else []
        return {n: sum(c.passed for r in self.results for c in r.checks if c.name == n) / len(self.results)
                for n in names}

    @property
    def cost_usd(self) -> float:
        return sum(r.cost_usd for r in self.results)

    @property
    def p50_s(self) -> float:
        return statistics.median(r.latency_s for r in self.results) if self.results else 0.0

    def failures(self) -> list[tuple[int, str, str]]:
        return [(r.issue_id, c.name, c.detail) for r in self.results for c in r.checks if not c.passed]

# script_id: day_020_first_eval_set/runner
    def summary(self) -> str:
        fields = " · ".join(f"{name} {acc:.2f}" for name, acc in self.field_accuracy.items())
        return (f"pass {sum(r.passed for r in self.results)}/{len(self.results)} ({self.pass_rate:.0%}) · "
                f"{fields} · ${self.cost_usd:.4f} · p50 {self.p50_s:.2f}s")

    def save(self, path: Path) -> Path:
        path.parent.mkdir(parents=True, exist_ok=True)
        data = {"meta": self.meta, "pass_rate": self.pass_rate, "field_accuracy": self.field_accuracy,
                "cost_usd": self.cost_usd, "p50_s": self.p50_s,
                "results": [asdict(r) | {"passed": r.passed} for r in self.results]}
        path.write_text(json.dumps(data, indent=2), encoding="utf-8")
        return path

# script_id: day_020_first_eval_set/runner
def run_meta() -> dict:
    try:
        sha = subprocess.run(["git", "rev-parse", "--short", "HEAD"], capture_output=True,
                             text=True, timeout=5).stdout.strip()
    except (OSError, subprocess.SubprocessError):
        sha = ""
    return {"model": os.environ.get("ATLAS_MODEL", "gpt-6-luna"), "effort": os.environ.get("ATLAS_EFFORT", "low"),
            "prompt": "extract_issue@inline", "git_sha": sha or "unknown",     # Day 21 versions prompts
            "created": datetime.now(timezone.utc).isoformat(timespec="seconds")}


def run_eval(cases: list[dict], predict: Callable[[dict], Any], checks: list[Check], *,
             label: str = "") -> EvalReport:
    results = []
    for case in cases:
        start, error = time.perf_counter(), ""
        try:
            out = predict(case)
        except Exception as err:             # one bad case is a failed case, not a crashed run
            out, error = None, f"{type(err).__name__}: {err}"
        latency = time.perf_counter() - start
        record = getattr(out, "record", out)        # ExtractResult (Day 17) or a bare record
        verdicts = [check(case, record) for check in checks]
        if record is None:                          # no record: every check fails, with the reason
            verdicts = [CheckResult(v.name, False, error or "no record") for v in verdicts]
        prediction = record.model_dump() if hasattr(record, "model_dump") else record
        results.append(CaseResult(case["issue_id"], verdicts, prediction,
                                  getattr(out, "cost_usd", 0.0), latency, error))
    if results and all(r.error for r in results):
        raise RuntimeError(f"every case failed, so this is a setup problem: {results[0].error}")
    return EvalReport(results, meta={"label": label, "n": len(cases), **run_meta()})

# script_id: day_020_first_eval_set/runner
def gate(report: EvalReport, *, threshold: float, min_field: float = 0.6) -> int:
    """0 = good enough to merge, 1 = block. Set threshold = your baseline minus a noise margin."""
    problems = [f"pass_rate {report.pass_rate:.2f} < {threshold:.2f}"] if report.pass_rate < threshold else []
    problems += [f"{name} {acc:.2f} < {min_field:.2f}" for name, acc in report.field_accuracy.items()
                 if acc < min_field]
    print("FAIL: " + "; ".join(problems) if problems else f"PASS: pass_rate {report.pass_rate:.2f} >= {threshold:.2f}")
    return 1 if problems else 0


if __name__ == "__main__":
    from predictors import keyword_baseline    # swap in llm_predict to score the real extractor
    report = run_eval(load_cases(split="dev"), keyword_baseline, DEFAULT_CHECKS, label="keyword_baseline")
    print(report.summary())
    print(f"{'issue':<6} {'check':<28} detail")
    for issue_id, name, detail in report.failures()[:6]:
        print(f"#{issue_id:<5} {name:<28} {detail}")
    print("saved", report.save(Path("reports/baseline.json")), "· gate exit code:", gate(report, threshold=0.65))

pass 4/20 (20%) · schema 1.00 · type 0.70 · severity 0.35 · components 0.65 · error_codes 1.00 · orbit_version 1.00 · summary_mentions_error_codes 0.85 · summary_length 0.95 · $0.0000 · p50 0.00s
issue  check                        detail
#10    type                         want 'question', got 'bug'
#10    severity                     want None, got 's3'
#25    type                         want 'question', got 'bug'
#25    severity                     want None, got 's3'
#63    type                         want 'question', got 'bug'
#63    severity                     want None, got 's3'
FAIL: pass_rate 0.20 < 0.65; severity 0.35 < 0.60
saved reports/baseline.json · gate exit code: 1

pass 14/20 (70%) · schema 1.00 · type 0.95 · severity 0.85 · components 0.75 · error_codes 1.00 · orbit_version 0.95 · summary_mentions_error_codes 0.95 · summary_length 1.00 · $0.0410 · p50 1.40s

# script_id: day_020_first_eval_set/gate
# gate.py: CI entry point. Run the eval, save the report, exit 1 when the gate fails.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY (llm predictor), ATLAS_MODEL, ATLAS_EFFORT
#   python gate.py --sample 5 --threshold 0.65              # PR check, ≈ $0.01 (as of 2026-11, verify)
#   python gate.py --predictor baseline --threshold 0 --min-field 0   # offline dry run, $0
import argparse
import os
import sys
from pathlib import Path

from runner import DEFAULT_CHECKS, gate, load_cases, run_eval


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description="Run the extraction eval and gate on the result.")
    p.add_argument("--split", choices=["dev", "heldout", "all"], default="dev")
    p.add_argument("--sample", type=int, default=0, help="first N cases only; 0 = the whole split")
    p.add_argument("--threshold", type=float, default=0.65, help="minimum pass rate")
    p.add_argument("--min-field", type=float, default=0.6, help="minimum accuracy for every check")
    p.add_argument("--predictor", choices=["llm", "baseline"], default="llm")
    p.add_argument("--out", type=Path, default=Path("reports/latest.json"))
    args = p.parse_args(argv)
    if args.predictor == "llm" and not os.environ.get("OPENAI_API_KEY"):
        print("ERROR: OPENAI_API_KEY is not set. A gate that didn't run must not pass.", file=sys.stderr)
        return 2
    from predictors import keyword_baseline, llm_predict
    cases = load_cases(split=args.split)[: args.sample or None]
    report = run_eval(cases, llm_predict if args.predictor == "llm" else keyword_baseline,
                      DEFAULT_CHECKS, label=f"{args.predictor}:{args.split}:{len(cases)}")
    print(report.summary(), "->", report.save(args.out))
    return gate(report, threshold=args.threshold, min_field=args.min_field)


if __name__ == "__main__":
    sys.exit(main())

$ python gate.py --sample 5 --threshold 0.65
pass 4/5 (80%) · schema 1.00 · type 1.00 · severity 1.00 · components 0.80 · ... · $0.0098 · p50 1.30s -> reports/latest.json
PASS: pass_rate 0.80 >= 0.65
$ echo $?
0

# .github/workflows/evals.yml (next to Day 19's ci.yml). Assumes today's files live in evals/.
name: evals
on:
  pull_request:
  push:
    branches: [main]
  schedule:
    - cron: "0 3 * * *"          # nightly at 03:00 UTC: the full dev set
  workflow_dispatch:
jobs:
  eval-gate:
    runs-on: ubuntu-latest
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}   # empty on fork PRs: GitHub withholds secrets
      ATLAS_MODEL: gpt-6-luna
      ATLAS_EFFORT: low
      ORBIT_DATA: datasets/orbit
    steps:
      - uses: actions/checkout@v7
      - uses: astral-sh/setup-uv@v10.3.3     # no floating major tag after v7: pin the full version
        with: { enable-cache: true }
      - run: uv sync --locked
      - name: Skip notice (no secret, e.g. a fork PR)
        if: env.OPENAI_API_KEY == ''
        run: echo "::notice::OPENAI_API_KEY unavailable; eval gate skipped. A maintainer runs it after merge."
      - name: Eval gate (5 cases on PRs, the whole dev set otherwise)
        if: env.OPENAI_API_KEY != ''
        run: uv run python evals/gate.py --sample ${{ github.event_name == 'pull_request' && 5 || 0 }} --threshold 0.65