Building Eval Datasets

# script_id: day_043_eval_datasets/coverage_grid
# Tally the 30 qa_gold questions into a kind x version grid. No install, no API key.
# Env: ORBIT_DATA (optional, default datasets/orbit)
import itertools
import json
import os
import re
from collections import Counter
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
KINDS = ["lookup", "how_to", "troubleshooting"]
VERSIONS = ["2.x", "3.x", "all"]
PERSONAS = ["new_user", "sre_on_call", "upgrader"]

def guess_kind(question: str) -> str:
    """A rough first-pass tag from the wording. A reviewer corrects it later."""
    q = question.lower()
    if re.search(r"orb-\d{4}|error|fail|why|stuck|loop", q):
        return "troubleshooting"
    if re.search(r"^how (do|should|does)|in what order|work around", q):
        return "how_to"
    return "lookup"

lines = (DATA_DIR / "eval/qa_gold.jsonl").read_text(encoding="utf-8").splitlines()
rows = [json.loads(line) for line in lines if line.strip()]
cells = Counter((guess_kind(r["question"]), r["applies_to"]) for r in rows)
print(f"{'kind':<16}" + "".join(f"{v:>6}" for v in VERSIONS))
for kind in KINDS:
    print(f"{kind:<16}" + "".join(f"{cells[(kind, v)]:>6}" for v in VERSIONS))
full = list(itertools.product(KINDS, VERSIONS, PERSONAS))
print(f"\nkind x version cells filled: {len(cells)}/{len(KINDS) * len(VERSIONS)}")
print(f"kind x version x persona cells filled: 0/{len(full)} (qa_gold has no persona)")

kind               2.x   3.x   all
lookup               3     7     8
how_to               0     2     3
troubleshooting      0     5     2

kind x version cells filled: 7/9
kind x version x persona cells filled: 0/27 (qa_gold has no persona)

# script_id: day_043_eval_datasets/eval_datasets
# eval_datasets.py: in your repo, atlas/evals/datasets.py. Schema, filters, dedup, versions, slices.
# Install: uv add pydantic numpy · Env: ORBIT_DATA (optional). No API key: callers pass `embed`.
import hashlib
import itertools
import json
import os
import random
import re
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Iterable, Literal

import numpy as np
from pydantic import BaseModel, Field, field_validator

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
DATASETS_DIR = Path(os.environ.get("ATLAS_DATASETS", "evals/datasets"))
DOC_IDS = {p.stem for p in (DATA_DIR / "docs").glob("*.md")}

class EvalCase(BaseModel):
    id: str                                    # never reuse or renumber: later days key on it
    question: str
    expected: str | None                       # None = the correct behaviour is to abstain
    must_cite: list[str] = Field(default_factory=list)
    applies_to: Literal["2.x", "3.x", "all"] = "all"
    tags: dict[str, str] = Field(default_factory=dict)
    source: Literal["golden", "synthetic", "production"]
    status: Literal["draft", "reviewed"] = "draft"

    @field_validator("must_cite")
    @classmethod
    def known_docs(cls, ids: list[str]) -> list[str]:
        unknown = [i for i in ids if i not in DOC_IDS]
        if unknown:
            raise ValueError(f"unknown doc id(s) {unknown}")
        return ids

# script_id: day_043_eval_datasets/eval_datasets
def guess_kind(question: str) -> str:
    """Same rough rule as coverage_grid; reviewers fix the tag when it's wrong."""
    q = question.lower()
    if re.search(r"orb-\d{4}|error|fail|why|stuck|loop", q):
        return "troubleshooting"
    if re.search(r"^how (do|should|does)|in what order|work around", q):
        return "how_to"
    return "lookup"

def golden_cases() -> list[EvalCase]:
    """The 30 hand-written qa_gold rows (Day 20) as reviewed EvalCases."""
    lines = (DATA_DIR / "eval/qa_gold.jsonl").read_text(encoding="utf-8").splitlines()
    rows = [json.loads(line) for line in lines if line.strip()]
    return [EvalCase(id=r["id"], question=r["question"], expected=r["answer"],
                     must_cite=r["citations"], applies_to=r["applies_to"], source="golden",
                     status="reviewed", tags={"kind": guess_kind(r["question"]), "lang": "en",
                                              "persona": "maintainer", "version": r["applies_to"]})
            for r in rows]

# script_id: day_043_eval_datasets/first_synthetic
# uv add "openai>=3.46,<4" pydantic · OPENAI_API_KEY · one call, < $0.01 (as of 2026-11, verify)
import os
from pathlib import Path
from openai import OpenAI
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
client = OpenAI()

class Draft(BaseModel):
    question: str
    expected: str   # the answer, one or two sentences
    quote: str      # exact sentence from the doc that supports `expected`
class Drafts(BaseModel):
    cases: list[Draft]

doc = (DATA_DIR / "docs/leases-and-heartbeats.md").read_text(encoding="utf-8")
resp = client.responses.parse(model=MODEL, reasoning={"effort": "low"}, text_format=Drafts,
    instructions="You write test questions for a docs assistant. Use only facts in the document.",
    input="Persona: SRE on call at 3 a.m. Kind: troubleshooting, mention one ORB error code.\n"
          f"Write 5 questions this persona would type, in their own words, with answers.\n\n<doc>\n{doc}\n</doc>")
for case in (resp.output_parsed or Drafts(cases=[])).cases:   # None on refusal or truncation
    print("-", case.question)

- Paged for ORB-1102 WorkerHeartbeatLost on the report workers. Is the box dead or just busy?
- Logs full of ORB-1101 LeaseExpired right after a deploy. Are jobs running twice now?
- One job shows ORB-1101 every night at 02:00. Do I raise lease_seconds or heartbeat_seconds?
- Worker froze for 10 minutes, came back. Did its jobs get picked up by someone else?
- orbit doctor complains heartbeat_seconds is too high. What ratio does it want?

# script_id: day_043_eval_datasets/first_synthetic
# Same doc, no persona and no kind. Compare the questions with the run above.
naive = client.responses.parse(
    model=MODEL, reasoning={"effort": "low"}, text_format=Drafts,
    instructions="You write test questions for a docs assistant. Use only facts in the document.",
    input=f"Write 5 questions about this document, with answers.\n\n<doc>\n{doc}\n</doc>")
for case in (naive.output_parsed or Drafts(cases=[])).cases:
    print("-", case.question)

- What is a lease in Orbit?
- How do heartbeats work in Orbit?
- What happens when a lease expires?
- What does heartbeat_seconds control?
- What is the relationship between lease_seconds and heartbeat_seconds?

# script_id: day_043_eval_datasets/eval_datasets
DIMENSIONS = {"kind": ["lookup", "how_to", "troubleshooting"],
              "persona": ["new_user", "sre_on_call", "upgrader"]}

def make_tuples(dims: dict[str, list[str]], k: int, seed: int) -> list[dict[str, str]]:
    """k distinct combinations, sampled with a fixed seed so a rebuild picks the same ones."""
    combos = [dict(zip(dims, values)) for values in itertools.product(*dims.values())]
    return random.Random(seed).sample(combos, min(k, len(combos)))

def norm(text: str) -> str:
    """Lowercase words only: drops markdown, backticks and punctuation before comparing."""
    return " ".join(re.findall(r"[a-z0-9_.-]+", text.lower()))

def grounded(quote: str, doc_text: str) -> bool:
    """The supporting quote must really be in the doc. Short quotes prove nothing."""
    return len(quote.split()) >= 4 and norm(quote) in norm(doc_text)

def copies_doc(question: str, doc_text: str, n: int = 6) -> bool:
    """True if the question shares an n-word run with the doc: it reads like the docs, not a user."""
    words, doc = norm(question).split(), f" {norm(doc_text)} "
    return any(f" {' '.join(words[i:i + n])} " in doc for i in range(len(words) - n + 1))

# script_id: day_043_eval_datasets/eval_datasets
Embed = Callable[[list[str]], np.ndarray]   # texts -> (n, dim) array of unit-length rows

def dedup(cases: list[EvalCase], against: Iterable[EvalCase] = (), threshold: float = 0.9,
          *, embed: Embed, dropped: list | None = None) -> list[EvalCase]:
    """Keep a case only if no earlier kept case (or `against` case) is within `threshold` cosine."""
    pool = list(against)
    if not cases:
        return []
    vecs = embed([c.question for c in pool + cases])
    ref_ids, ref_vecs = [c.id for c in pool], list(vecs[:len(pool)])
    kept: list[EvalCase] = []
    for case, vec in zip(cases, vecs[len(pool):]):
        if ref_vecs:
            sims = np.asarray(ref_vecs) @ vec            # unit vectors: dot product = cosine
            j = int(np.argmax(sims))
            if sims[j] >= threshold:
                if dropped is not None:
                    dropped.append((case.id, ref_ids[j], float(sims[j])))
                continue
        kept.append(case)
        ref_ids.append(case.id)
        ref_vecs.append(vec)
    return kept

# script_id: day_043_eval_datasets/build_dataset
# Generate, filter and dedup synthetic cases, then write a draft for human review.
# Install: uv add "openai>=3.46,<4" pydantic numpy · Env: OPENAI_API_KEY; ATLAS_MODEL, EMBED_MODEL, ORBIT_DATA
# Cost: 13 generation calls + one embeddings call ≈ $0.01-0.03 on gpt-6-luna (as of 2026-11, verify)
import json
import os
import re
from collections import Counter

import numpy as np
from openai import OpenAI
from pydantic import BaseModel, ValidationError

from eval_datasets import (DATA_DIR, DATASETS_DIR, DIMENSIONS, EvalCase, copies_doc, dedup,
                           golden_cases, grounded, make_tuples)

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "text-embedding-3-small")
client = OpenAI()
REJECTS: Counter[str] = Counter()

class Draft(BaseModel):
    question: str
    expected: str
    quote: str

class Drafts(BaseModel):
    cases: list[Draft]

GEN_INSTRUCTIONS = (
    "You write test questions for Atlas, an assistant that answers questions about the Orbit job "
    "library from its docs. For EACH tuple, in order, write one question that persona would really "
    "type: their own words and situation, never sentences copied from the doc. Answer it from the "
    "doc only, and copy the exact supporting sentence from the doc into `quote`.")

def embed_texts(texts: list[str]) -> np.ndarray:
    """Day 23's helper. OpenAI embeddings are unit length, so a dot product is the cosine."""
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    return np.asarray([d.embedding for d in resp.data], dtype=np.float32)

# script_id: day_043_eval_datasets/build_dataset
def generate_cases(doc_id: str, tuples: list[dict[str, str]], n: int | None = None,
                   *, prefix: str = "syn") -> list[EvalCase]:
    doc = (DATA_DIR / "docs" / f"{doc_id}.md").read_text(encoding="utf-8")
    applies_to = re.search(r"^applies_to: (\S+)", doc, re.M).group(1).strip('"')
    tuples = tuples[:n] if n else tuples
    resp = client.responses.parse(
        model=MODEL, reasoning={"effort": "low"}, text_format=Drafts, instructions=GEN_INSTRUCTIONS,
        input=f"Tuples:\n{json.dumps(tuples)}\n\n<doc id=\"{doc_id}\">\n{doc}\n</doc>")
    drafts = resp.output_parsed.cases if resp.output_parsed else []
    if len(drafts) != len(tuples):
        REJECTS["wrong_count"] += abs(len(tuples) - len(drafts))
    cases = []
    for i, (tags, d) in enumerate(zip(tuples, drafts)):
        if not grounded(d.quote, doc):
            REJECTS["quote_not_in_doc"] += 1
        elif copies_doc(d.question, doc):
            REJECTS["copies_doc_wording"] += 1
        else:
            try:
                cases.append(EvalCase(
                    id=f"{prefix}-{doc_id}-{i}", question=d.question, expected=d.expected,
                    must_cite=[doc_id], applies_to=applies_to, source="synthetic",
                    tags={**tags, "version": applies_to, "lang": "en"}))
            except ValidationError:
                REJECTS["schema"] += 1
    return cases

# script_id: day_043_eval_datasets/build_dataset
DOCS = ["leases-and-heartbeats", "retries-and-backoff", "idempotency", "migrating-2-to-3",
        "environment-variables-2x", "broker-redis", "broker-sqs", "scheduling-cron",
        "dead-letter-queue", "flows-error-handling", "dashboard-auth-oidc",
        "troubleshooting-duplicates", "deployment-kubernetes"]

if __name__ == "__main__":
    candidates: list[EvalCase] = []
    for seed, doc_id in enumerate(DOCS):
        candidates += generate_cases(doc_id, make_tuples(DIMENSIONS, k=5, seed=seed))
    dropped: list = []
    kept = dedup(candidates, against=golden_cases(), threshold=0.9, embed=embed_texts, dropped=dropped)
    for case_id, near_id, sim in sorted(dropped, key=lambda d: d[2])[:3]:
        print(f"  dropped {case_id} ~ {near_id} ({sim:.2f})")   # eyeball the pairs nearest 0.9
    total = len(candidates) + sum(REJECTS.values())
    print(f"kept {len(kept)}/{total} synthetic; filter rejects {dict(REJECTS)}; dedup drops {len(dropped)}")
    DATASETS_DIR.mkdir(parents=True, exist_ok=True)
    out = DATASETS_DIR / "ask_v1.synthetic.draft.jsonl"
    out.write_text("".join(c.model_dump_json() + "\n" for c in kept), encoding="utf-8")
    print(f"wrote {out}")

  dropped syn-idempotency-3 ~ qa-021 (0.91)
  dropped syn-broker-sqs-1 ~ syn-broker-sqs-0 (0.92)
  dropped syn-migrating-2-to-3-2 ~ qa-010 (0.93)
kept 50/65 synthetic; filter rejects {'quote_not_in_doc': 5, 'copies_doc_wording': 3}; dedup drops 7
wrote evals/datasets/ask_v1.synthetic.draft.jsonl

# script_id: day_043_eval_datasets/production_cases
# Real-shaped cases from Orbit `type:question` issues + hand-written unanswerables -> a draft file.
# Install: uv add pydantic numpy · Env: ORBIT_DATA (optional). No API key.
import json
import random
import re

from eval_datasets import DATA_DIR, DATASETS_DIR, DOC_IDS, EvalCase, guess_kind

MAINTAINERS = {"mara-k", "devon-li", "sam-okafor", "ines-v"}
NON_LATIN = re.compile(r"[À-ÿऀ-ॿ぀-ヿ一-鿿]")  # accents, Devanagari, kana, CJK
QUOTAS = {"non_en": 4, "vague": 3, "2.x": 2, "3.x": 3}   # rare strata oversampled on purpose
INJECTION_FIXTURE = 20

def stratum(issue: dict) -> str:
    if NON_LATIN.search(issue["title"] + issue["body"][:400]):
        return "non_en"
    if issue["orbit_version"] is None:
        return "vague"                                    # the reporter never said which version
    return "2.x" if issue["orbit_version"].startswith("2.") else "3.x"

def to_case(issue: dict, group: str) -> EvalCase:
    replies = [c["body"] for c in issue["comments"] if c["author"] in MAINTAINERS]
    cited = sorted({m for r in replies for m in re.findall(r"`([a-z0-9-]+)`", r) if m in DOC_IDS})
    v = issue["orbit_version"] or ""
    version = "2.x" if v.startswith("2.") else "3.x" if v else "all"
    return EvalCase(
        id=f"prod-issue-{issue['id']}", question=f"{issue['title']}\n\n{issue['body']}",
        expected=replies[0][:600] if replies else None,  # a starting point; the reviewer rewrites it
        must_cite=cited, applies_to=version, source="production",
        tags={"kind": guess_kind(issue["title"]), "persona": "real_user", "version": version,
              "lang": "non_en" if group == "non_en" else "en", "stratum": group})

# script_id: day_043_eval_datasets/production_cases
UNANSWERABLE = [
    "Is there a Grafana dashboard template for Orbit's metrics?",
    "What's on the roadmap for Orbit 4.0?",
    "Is there an Orbit Slack or Discord where I can ask questions?",
    "Which open-source license is Orbit released under?",
    "Is there an official TypeScript client for enqueuing Orbit jobs from Node?",
    "Can Orbit report failed jobs to Sentry?",
    "How many companies run Orbit in production?",
    "Is there a mobile app for the Orbit dashboard?",
]

def sample_issues(seed: int = 43) -> list[EvalCase]:
    lines = (DATA_DIR / "issues/issues.jsonl").read_text(encoding="utf-8").splitlines()
    issues = [json.loads(line) for line in lines if line.strip()]
    questions = [i for i in issues if "type:question" in i["labels"]]
    rng = random.Random(seed)                             # same sample on every rebuild
    picked = []
    for group, k in QUOTAS.items():
        pool = [i for i in questions if stratum(i) == group]
        picked += [to_case(i, group) for i in rng.sample(pool, min(k, len(pool)))]
    fixture = next(i for i in issues if i["id"] == INJECTION_FIXTURE)
    return picked + [to_case(fixture, "injection")]

if __name__ == "__main__":
    cases = sample_issues() + [
        EvalCase(id=f"unans-{n:02d}", question=q, expected=None, source="golden",
                 tags={"kind": "unanswerable", "persona": "maintainer", "version": "all", "lang": "en"})
        for n, q in enumerate(UNANSWERABLE, 1)]
    DATASETS_DIR.mkdir(parents=True, exist_ok=True)
    out = DATASETS_DIR / "ask_v1.production.draft.jsonl"
    out.write_text("".join(c.model_dump_json() + "\n" for c in cases), encoding="utf-8")
    for c in cases[:5] + cases[-2:]:
        print(f"{c.id:<16}{c.tags.get('stratum', c.tags['kind']):<13}"
              f"{','.join(c.must_cite)[:30]:<32}{c.question.splitlines()[0][:32]}")
    print(f"{len(cases)} draft cases -> {out}")

prod-issue-25   non_en       concepts-jobs,performance-tuni  ORB-1202 PayloadTooLarge al enco
prod-issue-63   non_en       deployment-docker,troubleshoot  Docker Compose में worker शुरू क
prod-issue-188  non_en       faq,scheduling-cron             cron式で ORB-1401 ScheduleParseErr
prod-issue-97   non_en       configuration-reference,enviro  Como migrar variáveis ORBIT_* pa
prod-issue-127  vague        flows-error-handling,flows-ove  flow not working
unans-07        unanswerable                                 How many companies run Orbit in
unans-08        unanswerable                                 Is there a mobile app for the Or
21 draft cases -> evals/datasets/ask_v1.production.draft.jsonl

# script_id: day_043_eval_datasets/review_cases
# Accept / edit / reject draft cases. No API key. Usage: python review_cases.py <draft.jsonl> ...
import sys
from pathlib import Path

from pydantic import ValidationError

from eval_datasets import EvalCase

HELP = "edit: e=expected (blank = abstain) c=must_cite k=kind · then a=accept r=reject q=quit"

def review(draft: Path) -> None:
    accepted = draft.with_name(draft.name.replace(".draft", ".reviewed"))
    rejected = draft.with_name(draft.name.replace(".draft", ".rejected"))
    done = {EvalCase.model_validate_json(line).id for p in (accepted, rejected) if p.exists()
            for line in p.read_text(encoding="utf-8").splitlines() if line}
    todo = [EvalCase.model_validate_json(line) for line in draft.read_text(encoding="utf-8").splitlines() if line]
    for case in (c for c in todo if c.id not in done):
        print(f"\n[{case.id}] {case.tags}\nQ: {case.question[:600]}\nA: {case.expected}\ncite: {case.must_cite}")
        update: dict = {}
        try:
            while (choice := input(f"{HELP}\n> ").strip().lower()) in ("e", "c", "k"):
                if choice == "e":
                    update["expected"] = input("expected: ").strip() or None
                elif choice == "c":
                    update["must_cite"] = [s.strip() for s in input("doc ids: ").split(",") if s.strip()]
                else:
                    update["tags"] = {**case.tags, "kind": input("kind: ").strip()}
        except EOFError:                       # Ctrl-D, or no terminal attached: stop cleanly
            choice = "q"
        if choice not in ("a", "r"):           # q, or an unknown key: save progress and stop
            break
        try:                                   # re-validate: an edited must_cite may be a typo
            case = EvalCase.model_validate({**case.model_dump(), **update, "status": "reviewed"})
        except ValidationError as err:
            print(err)
            continue
        with (rejected if choice == "r" else accepted).open("a", encoding="utf-8") as f:
            f.write(case.model_dump_json() + "\n")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("usage: python review_cases.py <draft.jsonl> [more drafts...]")
    for arg in sys.argv[1:]:
        review(Path(arg))

[prod-issue-28] {'kind': 'lookup', 'persona': 'real_user', 'version': 'all', 'lang': 'en', 'stratum': 'vague'}
Q: orbit is slow

after update everything is slow. jobs take forever. pls fix asap
A: Sorry to hear that. To dig in we need: ...
cite: ['performance-tuning']
edit: e=expected (blank = abstain) c=must_cite k=kind · then a=accept r=reject q=quit
> k
kind: troubleshooting
> e
expected: First tell apart jobs that wait longer to start from jobs that run longer; ask for the Orbit version, Python version and broker.
> a

# script_id: day_043_eval_datasets/eval_datasets
class Manifest(BaseModel):
    name: str
    version: str
    sha256: str                            # of the JSONL bytes; load_dataset() checks it
    n: int
    counts: dict[str, dict[str, int]]      # "source" and every tag -> value -> count
    created: str
    parent: str | None = None              # the version this one was built from

def _counts(cases: list[EvalCase]) -> dict[str, dict[str, int]]:
    counts: dict[str, Counter] = defaultdict(Counter)
    for case in cases:
        counts["source"][case.source] += 1
        for tag, value in case.tags.items():
            counts[tag][value] += 1
    return {tag: dict(sorted(c.items())) for tag, c in sorted(counts.items())}

def save_dataset(cases: list[EvalCase], name: str, version: str, parent: str | None = None) -> Manifest:
    path = DATASETS_DIR / f"{name}_{version}.jsonl"
    if path.exists():
        raise FileExistsError(f"{path} exists; versions are immutable, so bump the version")
    ids = [c.id for c in cases]
    if len(ids) != len(set(ids)):
        raise ValueError(f"duplicate case ids: {sorted({i for i in ids if ids.count(i) > 1})}")
    if any(c.status != "reviewed" for c in cases):
        raise ValueError("only reviewed cases go into a version; finish review first")
    payload = "".join(c.model_dump_json() + "\n" for c in cases).encode("utf-8")
    DATASETS_DIR.mkdir(parents=True, exist_ok=True)
    path.write_bytes(payload)
    manifest = Manifest(name=name, version=version, sha256=hashlib.sha256(payload).hexdigest(),
                        n=len(cases), counts=_counts(cases), parent=parent,
                        created=datetime.now(timezone.utc).isoformat(timespec="seconds"))
    path.with_suffix(".manifest.json").write_text(manifest.model_dump_json(indent=2), encoding="utf-8")
    return manifest

def load_dataset(name: str, version: str) -> list[EvalCase]:
    path = DATASETS_DIR / f"{name}_{version}.jsonl"
    manifest = Manifest.model_validate_json(path.with_suffix(".manifest.json").read_text(encoding="utf-8"))
    payload = path.read_bytes()
    if hashlib.sha256(payload).hexdigest() != manifest.sha256:
        raise ValueError(f"{path.name} changed after it was versioned (sha256 mismatch). "
                         "Scores on it are no longer comparable: save a new version instead.")
    return [EvalCase.model_validate_json(line) for line in payload.decode("utf-8").splitlines() if line]

# script_id: day_043_eval_datasets/freeze_v1
# Assemble ask_v1 = golden + reviewed synthetic + reviewed production, then freeze it.
# Install: uv add pydantic numpy · No API key. Run after review_cases.py on both draft files.
import shutil

from eval_datasets import DATASETS_DIR, EvalCase, golden_cases, load_dataset, save_dataset

def read_reviewed(file_name: str) -> list[EvalCase]:
    path = DATASETS_DIR / file_name
    if not path.exists():
        print(f"(no {path} yet: run review_cases.py on the draft first)")
        return []
    return [EvalCase.model_validate_json(line) for line in path.read_text(encoding="utf-8").splitlines() if line]

cases = (golden_cases() + read_reviewed("ask_v1.synthetic.reviewed.jsonl")
         + read_reviewed("ask_v1.production.reviewed.jsonl"))
manifest = save_dataset(cases, "ask", "v1")
print(f"ask v1: n={manifest.n} sha256={manifest.sha256[:12]}… counts by source={manifest.counts['source']}")
assert len(load_dataset("ask", "v1")) == manifest.n

# script_id: day_043_eval_datasets/freeze_v1
# Break it: "fix" one answer in a copy without bumping the version.
for suffix in (".jsonl", ".manifest.json"):
    shutil.copy(DATASETS_DIR / f"ask_v1{suffix}", DATASETS_DIR / f"ask_vtamper{suffix}")
tampered = DATASETS_DIR / "ask_vtamper.jsonl"
tampered.write_text(tampered.read_text(encoding="utf-8").replace("300", "600", 1), encoding="utf-8")
try:
    load_dataset("ask", "vtamper")
except ValueError as err:
    print("ValueError:", err)

(no evals/datasets/ask_v1.synthetic.reviewed.jsonl yet: run review_cases.py on the draft first)
(no evals/datasets/ask_v1.production.reviewed.jsonl yet: run review_cases.py on the draft first)
ask v1: n=30 sha256=e0e57794edd8… counts by source={'golden': 30}
ValueError: ask_vtamper.jsonl changed after it was versioned (sha256 mismatch). Scores on it are no longer comparable: save a new version instead.

# script_id: day_043_eval_datasets/eval_datasets
class RunRecord(BaseModel):
    """One case's result. Days 44-47 read this shape, so keep the field names."""
    case_id: str
    question: str
    source: str
    tags: dict[str, str]
    retrieved_ids: list[str]
    retrieved_applies_to: list[str]
    answer: str
    citations: list[str]
    checks: dict[str, bool]
    passed: bool
    trace_id: str | None = None            # fill from Day 42's step_span when tracing is on
    latency_s: float = 0.0
    cost_usd: float = 0.0

def write_run(results: list[RunRecord], path: str | Path) -> Path:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("".join(r.model_dump_json() + "\n" for r in results), encoding="utf-8")
    return path

def _slice(r: RunRecord, tag: str) -> str:
    return r.source if tag == "source" else r.tags.get(tag, "(none)")

def pass_rate_by(results: list[RunRecord], tag: str) -> dict[str, float]:
    groups: dict[str, list[bool]] = defaultdict(list)
    for r in results:
        groups[_slice(r, tag)].append(r.passed)
    return {value: sum(v) / len(v) for value, v in sorted(groups.items())}

def slice_table(results: list[RunRecord], tags=("source", "kind", "version", "lang")) -> str:
    rows = [f"{'slice':<26}{'n':>4}{'pass':>7}"]
    for tag in tags:
        n = Counter(_slice(r, tag) for r in results)
        rows += [f"{tag + '=' + v:<26}{n[v]:>4}{rate:>7.0%}" for v, rate in pass_rate_by(results, tag).items()]
    return "\n".join(rows)

# script_id: day_043_eval_datasets/baseline_r2
# Run mini-R2 over ask_v1, grade every case, write runs/ask_v1_baseline.jsonl, print slices.
# Install: uv add "openai>=3.46,<4" pydantic numpy rank-bm25 · Env: OPENAI_API_KEY; ATLAS_MODEL, ORBIT_DATA, EVAL_LIMIT
# Cost: ~100 answers + ~90 judge calls ≈ $0.05-0.10 on gpt-6-luna (as of 2026-11, verify)
import os
import re
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass

from openai import OpenAI
from pydantic import BaseModel
from rank_bm25 import BM25Okapi

from eval_datasets import DATA_DIR, EvalCase, RunRecord, load_dataset, slice_table, write_run

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
PRICES = {"gpt-6-luna": (0.10, 0.50)}     # USD per 1M input/output tokens, as of 2026-11, verify
client = OpenAI()
IDK = "I can't find that in the Orbit docs."

class Citation(BaseModel):
    doc_id: str
    quote: str

class Answer(BaseModel):
    text: str
    citations: list[Citation]

@dataclass
class Chunk:
    doc_id: str
    applies_to: str
    text: str

def tokenize(text: str) -> list[str]:
    return re.findall(r"[a-z0-9_]+", text.lower())

CHUNKS: list[Chunk] = []
for path in sorted((DATA_DIR / "docs").glob("*.md")):
    raw = path.read_text(encoding="utf-8")
    applies_to = re.search(r"^applies_to: (\S+)", raw, re.M).group(1).strip('"')
    body = re.sub(r"\A---\n.*?\n---\n", "", raw, flags=re.S)
    CHUNKS += [Chunk(path.stem, applies_to, s.strip()) for s in re.split(r"\n(?=## )", body) if s.strip()]
BM25 = BM25Okapi([tokenize(c.text) for c in CHUNKS])

# script_id: day_043_eval_datasets/baseline_r2
def cost(usage) -> float:
    p_in, p_out = PRICES.get(MODEL, (0.0, 0.0))
    return (usage.input_tokens * p_in + usage.output_tokens * p_out) / 1e6

def ask(question: str, k: int = 6) -> tuple[Answer, list[Chunk], float]:
    """Mini-R2: a keyword-only stand-in for your Day 34 atlas.rag.ask()."""
    scores = BM25.get_scores(tokenize(question))
    hits = [CHUNKS[i] for i in sorted(range(len(CHUNKS)), key=lambda i: -scores[i])[:k]]
    context = "\n\n".join(f'<source doc_id="{h.doc_id}">\n{h.text}\n</source>' for h in hits)
    resp = client.responses.parse(
        model=MODEL, text_format=Answer, reasoning={"effort": "low"},
        instructions="You are Atlas, a support assistant for the Orbit job library. Answer ONLY from "
                     "<context> and cite each claim with its doc_id and an exact short quote. If the "
                     f"sources don't contain the answer, reply exactly '{IDK}' with no citations.",
        input=f"<context>\n{context}\n</context>\n\nQuestion: {question}")
    return resp.output_parsed or Answer(text=IDK, citations=[]), hits, cost(resp.usage)

class Verdict(BaseModel):
    reason: str        # evidence first, so the verdict follows from it
    passed: bool

def judge_correct(question: str, answer: str, reference: str) -> tuple[bool, float]:
    """Day 30's correctness judge, compact. Day 45 designs judges properly."""
    resp = client.responses.parse(
        model=MODEL, text_format=Verdict, reasoning={"effort": "low"},
        instructions="Grade one answer. In `reason`, quote the evidence; then set `passed`. When unsure, FAIL.",
        input="Does <answer> agree with <reference> on every fact <question> asks about? Extra correct "
              f"detail is fine.\n<question>{question}</question>\n<reference>{reference}</reference>\n"
              f"<answer>{answer}</answer>")
    return bool(resp.output_parsed and resp.output_parsed.passed), cost(resp.usage)

# script_id: day_043_eval_datasets/baseline_r2
def grade(case: EvalCase) -> RunRecord:
    start = time.perf_counter()
    answer, hits, spent = ask(case.question)
    cited = [c.doc_id for c in answer.citations]
    abstained = answer.text.strip() == IDK and not cited
    if case.expected is None:                           # unanswerable: only a clean abstain passes
        checks = {"abstains": abstained}
    else:
        ok, judge_cost = (False, 0.0) if abstained else judge_correct(case.question, answer.text, case.expected)
        spent += judge_cost
        checks = {"correct": ok}
        if case.must_cite:
            checks["cites_expected_doc"] = bool(set(cited) & set(case.must_cite))
    return RunRecord(case_id=case.id, question=case.question, source=case.source, tags=case.tags,
                     retrieved_ids=[h.doc_id for h in hits], retrieved_applies_to=[h.applies_to for h in hits],
                     answer=answer.text, citations=cited, checks=checks, passed=all(checks.values()),
                     latency_s=time.perf_counter() - start, cost_usd=spent)

def run_eval(cases: list[EvalCase], workers: int = 8) -> list[RunRecord]:
    """Day 20's runner with threads: the calls are I/O-bound, so 8 workers cut wall time a lot."""
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(grade, cases))

if __name__ == "__main__":
    cases = load_dataset("ask", "v1")[: int(os.environ.get("EVAL_LIMIT", "1000"))]
    results = run_eval(cases)
    path = write_run(results, "runs/ask_v1_baseline.jsonl")
    golden = [r for r in results if r.case_id.startswith("qa-")]
    print(f"qa_gold only: {sum(r.passed for r in golden) / max(len(golden), 1):.0%} of {len(golden)}")
    print(f"all of v1:    {sum(r.passed for r in results) / len(results):.0%} of {len(results)} "
          f"(${sum(r.cost_usd for r in results):.3f}) -> {path}\n")
    print(slice_table(results))

qa_gold only: 83% of 30
all of v1:    61% of 100 ($0.071) -> runs/ask_v1_baseline.jsonl

slice                        n   pass
source=golden               38    76%
source=production           13    23%
source=synthetic            49    59%
kind=how_to                 22    64%
kind=lookup                 34    74%
kind=troubleshooting        36    50%
kind=unanswerable            8    50%
version=2.x                  9    33%
version=3.x                 51    61%
version=all                 40    68%
lang=en                     96    64%
lang=non_en                  4     0%