felab package

The shared helper package every lesson imports.

felab/__init__.py Python · 12 lines

"""felab — the shared toolkit every lesson imports.

    from felab import add_target_args, resolve, client, banner   # where to send requests
    from felab import stream_once, astream_once, percentile, user # how to time them
    from felab import record, table                               # where the numbers go
"""
from .measure import Sample, astream_once, percentile, stream_once, user
from .results import record, table
from .targets import TARGETS, Target, add_target_args, banner, client, resolve

__all__ = ["Sample", "astream_once", "percentile", "stream_once", "user", "record", "table",
           "TARGETS", "Target", "add_target_args", "banner", "client", "resolve"]

felab/hardware.py Python · 54 lines

"""
felab.hardware — detect the machine and know its memory bandwidth.

Why bandwidth? During decode, every new token requires reading (roughly) all
active weights from memory once. So:

    decode ceiling (tok/s) ≈ memory bandwidth (GB/s) ÷ bytes read per token (GB)

That single line explains why quantization, MoE and batching matter. Lessons 04
and 12 compare this ceiling with what you measure (expect 60–85% of it).
"""
from __future__ import annotations

import platform
import re
import subprocess

# Approximate peak unified-memory bandwidth, GB/s. Where a chip ships in two
# bandwidth bins (e.g. M3 Max, M4 Max) we list the higher one — check yours.
APPLE_BW = {
    "M1": 68, "M1 Pro": 200, "M1 Max": 400, "M1 Ultra": 800,
    "M2": 100, "M2 Pro": 200, "M2 Max": 400, "M2 Ultra": 800,
    "M3": 100, "M3 Pro": 150, "M3 Max": 400, "M3 Ultra": 819,
    "M4": 120, "M4 Pro": 273, "M4 Max": 546,
    "M5": 153,
}
OTHER_BW = {"DGX Spark (GB10)": 273, "RTX 4090": 1008, "H100 SXM": 3350, "H200": 4800, "B200": 8000}


def _sysctl(key: str) -> str:
    try:
        return subprocess.run(["sysctl", "-n", key], capture_output=True, text=True, timeout=3).stdout.strip()
    except Exception:
        return ""


def detect() -> dict:
    """Return {'chip': str, 'ram_gb': float, 'bw_gbs': int|None}."""
    if platform.system() == "Darwin":
        chip = _sysctl("machdep.cpu.brand_string") or "Apple ?"
        ram = int(_sysctl("hw.memsize") or 0) / 1e9
        m = re.search(r"(M\d)(\s+(Pro|Max|Ultra))?", chip)
        key = (m.group(1) + (" " + m.group(3) if m.group(3) else "")) if m else ""
        return {"chip": chip, "ram_gb": round(ram, 1), "bw_gbs": APPLE_BW.get(key)}
    try:
        ram = int(open("/proc/meminfo").read().split()[1]) / 1e6
    except Exception:
        ram = 0.0
    return {"chip": platform.processor() or platform.machine(), "ram_gb": round(ram, 1), "bw_gbs": None}


def ceiling(bw_gbs: float, gb_per_token: float) -> float:
    """Upper bound on single-stream decode speed."""
    return bw_gbs / gb_per_token if gb_per_token else float("inf")

felab/measure.py Python · 99 lines

"""
felab.measure — the two numbers every inference conversation is about.

    TTFT  time to first token   = queue + PREFILL (reading the whole prompt)
    ITL   inter-token latency   = one DECODE step (writing one more token)

Users feel TTFT as "is it thinking?" and ITL as "how fast does it type?".
They are driven by different things (compute vs memory bandwidth), so we
ALWAYS report them separately — an average "latency" hides both.

Note: we count streamed chunks as tokens. Most servers send one token per
chunk; a few batch several. For exact counts use `usage` (non-streaming).
"""
from __future__ import annotations

import math
import time
from dataclasses import dataclass, field


@dataclass
class Sample:
    ttft_ms: float                 # prompt sent → first visible token
    itl_ms: list[float] = field(default_factory=list)  # gap between consecutive tokens
    tokens: int = 0                # streamed chunks with content
    total_s: float = 0.0           # whole request wall time
    text: str = ""

    @property
    def itl_median(self) -> float:
        return percentile(self.itl_ms, 50) if self.itl_ms else 0.0

    @property
    def decode_tps(self) -> float:
        """Tokens per second for ONE user once text starts flowing."""
        return 1000.0 / self.itl_median if self.itl_median else 0.0


def percentile(xs: list[float], p: float) -> float:
    """Nearest-rank percentile. p95 of 10 samples is the worst one — that is
    deliberate: tails are what customers complain about."""
    if not xs:
        return float("nan")
    s = sorted(xs)
    k = max(0, min(len(s) - 1, math.ceil(p / 100 * len(s)) - 1))
    return s[k]


def _content(chunk) -> str:
    """Pull visible text out of a streamed chunk. Reasoning models may stream
    `reasoning_content` first; we count that too because the user waits for it."""
    if not chunk.choices:
        return ""
    d = chunk.choices[0].delta
    return (getattr(d, "content", None) or getattr(d, "reasoning_content", None) or "")


def stream_once(cli, model: str, messages: list[dict], max_tokens: int = 192, **extra) -> Sample:
    """Send one streaming request and time every token (sync client)."""
    t0 = time.perf_counter()
    stamps: list[float] = []
    parts: list[str] = []
    for ch in cli.chat.completions.create(model=model, messages=messages, max_tokens=max_tokens,
                                          stream=True, **extra):
        piece = _content(ch)
        if piece:
            stamps.append(time.perf_counter())
            parts.append(piece)
    return _to_sample(t0, stamps, parts)


async def astream_once(cli, model: str, messages: list[dict], max_tokens: int = 128, **extra) -> Sample:
    """Same as stream_once for an AsyncOpenAI client (used by the concurrency sweep)."""
    t0 = time.perf_counter()
    stamps: list[float] = []
    parts: list[str] = []
    stream = await cli.chat.completions.create(model=model, messages=messages, max_tokens=max_tokens,
                                               stream=True, **extra)
    async for ch in stream:
        piece = _content(ch)
        if piece:
            stamps.append(time.perf_counter())
            parts.append(piece)
    return _to_sample(t0, stamps, parts)


def _to_sample(t0: float, stamps: list[float], parts: list[str]) -> Sample:
    end = time.perf_counter()
    if not stamps:  # server returned nothing visible
        return Sample(ttft_ms=(end - t0) * 1000, total_s=end - t0)
    itl = [(b - a) * 1000 for a, b in zip(stamps, stamps[1:])]
    return Sample(ttft_ms=(stamps[0] - t0) * 1000, itl_ms=itl, tokens=len(stamps),
                  total_s=end - t0, text="".join(parts))


def user(prompt: str, system: str | None = None) -> list[dict]:
    """Build a chat message list."""
    msgs = [{"role": "system", "content": system}] if system else []
    return msgs + [{"role": "user", "content": prompt}]

felab/mock_server.py Python · 227 lines

"""
felab.mock_server — a fake OpenAI-compatible LLM server with realistic *shape*.

    python -m felab.mock_server            # listens on :9000
    python -m felab.mock_server --slots 4 --itl-ms 25

It generates no real language. What it does simulate, on purpose, is the
behaviour each lesson teaches, so you can run every script offline first
and know what "right" looks like before spending time or money:

  * PREFILL cost grows with prompt length            → lesson 01 (TTFT vs prompt size)
  * DECODE slows a little as more users share a step → lesson 03 (concurrency curve)
  * only N "slots" (batch size); extra users queue   → lesson 03 (the TTFT knee)
  * a PREFIX CACHE in 64-token blocks                → lesson 06 (shared prefixes)
  * JSON: constrained decoding always valid,
    prompt-only JSON sometimes breaks                → lesson 07 (structured output)
  * a "-lora" model id classifies tickets better     → lessons 09–11 (eval delta)

Standard library only. Not a benchmark of anything real — the numbers are made up.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import random
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

CFG = {"slots": 8, "itl_ms": 18.0, "prefill_tok_per_s": 2500.0, "block": 64}
SLOTS: threading.Semaphore
ACTIVE = 0
LOCK = threading.Lock()
PREFIX_CACHE: set[str] = set()   # hashes of cached prompt blocks

WORDS = ("the cache keeps keys and values so each new token only reads what it needs while "
         "bandwidth sets the ceiling and batching shares one pass across many users").split()

# The "ground truth" the mock uses to answer ticket-triage prompts (lesson 07/09).
KEYWORDS = {  # checked in this order — abuse first so "fake invoices" is abuse, not billing
    "abuse": ["spam", "phishing", "abuse", "fraud", "stolen", "scraping"],
    "billing": ["invoice", "card", "charge", "refund", "billing", "payment", "declined"],
    "outage": ["down", "500", "timeout", "time out", "outage", "unreachable", "latency spike", "error rate"],
    "how-to": ["how do i", "how to", "where can i", "docs", "configure", "set up"],
}


def approx_tokens(text: str) -> int:
    return max(1, len(text) // 4)          # ~4 characters per token for English


def prompt_text(messages: list[dict]) -> str:
    out = []
    for m in messages:
        c = m.get("content", "")
        out.append(c if isinstance(c, str) else json.dumps(c))
    return "\n".join(out)


def uncached_tokens(text: str) -> int:
    """Walk the prompt in fixed blocks; blocks whose whole prefix was seen before are free.
    Mirrors how vLLM/SGLang/llama.cpp hash KV blocks. One changed character early on
    invalidates every block after it — that is why prompt ORDER matters."""
    block_chars = CFG["block"] * 4
    h = hashlib.sha1()
    cached_chars = 0
    new_hashes = []
    for i in range(0, len(text) - block_chars + 1, block_chars):
        h.update(text[i:i + block_chars].encode())
        key = h.hexdigest()
        if key in PREFIX_CACHE and cached_chars == i:
            cached_chars = i + block_chars
        new_hashes.append(key)
    PREFIX_CACHE.update(new_hashes)
    return approx_tokens(text[cached_chars:]) if cached_chars < len(text) else 0


def classify(text: str, model: str) -> tuple[str, int]:
    """Pretend-model: finds the true label by keyword, then is sometimes wrong.
    Base model is right ~75% of the time, a '-lora' model ~93%. Deterministic per input."""
    low = text.lower()
    truth = next((c for c, kws in KEYWORDS.items() if any(k in low for k in kws)), "how-to")
    rng = random.Random(hashlib.md5((low + model).encode()).hexdigest())
    acc = 0.93 if "lora" in model else 0.75
    label = truth if rng.random() < acc else rng.choice([c for c in KEYWORDS if c != truth])
    sev = {"outage": 4, "abuse": 4, "billing": 3, "how-to": 2}[label]
    if any(w in low for w in ("urgent", "tomorrow", "all users", "production")):
        sev = min(5, sev + 1)
    return label, sev


def fake_json(messages: list[dict], model: str, schema: dict | None) -> str:
    user_text = next((m["content"] for m in reversed(messages) if m.get("role") == "user"), "")
    label, sev = classify(user_text, model)
    obj = {"category": label, "severity": sev, "next_action": f"route to {label} queue"}
    if schema and "properties" in schema:  # only keep fields the schema asks for
        obj = {k: v for k, v in obj.items() if k in schema["properties"]} or obj
    return json.dumps(obj)


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, *a):  # keep the terminal quiet
        pass

    def _json(self, code: int, obj: dict) -> None:
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path.rstrip("/") in ("/health", "/v1/health"):
            return self._json(200, {"status": "ok"})
        if self.path.startswith("/v1/models"):
            return self._json(200, {"object": "list", "data": [{"id": "mock-8b", "object": "model"},
                                                              {"id": "mock-8b-lora", "object": "model"}]})
        self._json(404, {"error": "not found"})

    def do_POST(self):
        if not self.path.endswith("/chat/completions"):
            return self._json(404, {"error": "only /v1/chat/completions is mocked"})
        req = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}")
        global ACTIVE
        SLOTS.acquire()                      # queue here if all batch slots are busy
        with LOCK:
            ACTIVE += 1
        try:
            self._complete(req)
        finally:
            with LOCK:
                ACTIVE -= 1
            SLOTS.release()

    def _complete(self, req: dict) -> None:
        messages = req.get("messages", [])
        model = req.get("model", "mock-8b")
        max_tokens = int(req.get("max_tokens") or 256)
        text = prompt_text(messages)
        prompt_toks = approx_tokens(text)

        # ---- PREFILL: pay only for the part of the prompt that is not cached ----
        todo = uncached_tokens(text)
        time.sleep(0.015 + todo / CFG["prefill_tok_per_s"])

        # ---- choose what to "generate" ----
        rf = req.get("response_format") or {}
        wants_json = "json" in text.lower() or rf.get("type") in ("json_object", "json_schema")
        if "rubric" in text.lower():          # acting as an LLM judge (lesson 09)
            # score high if the proposed action routes to the ticket's true category
            ticket = text.split("Ticket:")[-1].split("Proposed next_action:")[0].lower()
            action = text.split("Proposed next_action:")[-1].split("Rubric:")[0].lower()
            truth = next((c for c, kws in KEYWORDS.items() if any(k in ticket for k in kws)), "how-to")
            rng = random.Random(text)
            score = rng.choice([4, 5, 5]) if truth in action else rng.choice([1, 2, 3])
            pieces = [json.dumps({"score": score})]
        elif wants_json:
            schema = (rf.get("json_schema") or {}).get("schema")
            out = fake_json(messages, model, schema)
            constrained = rf.get("type") in ("json_object", "json_schema")
            # Without constrained decoding a small model sometimes wraps or truncates JSON.
            if not constrained and random.Random(text).random() < 0.12:
                out = "Sure! Here is the JSON:\n" + out[:-3]
            pieces = [out[i:i + 4] for i in range(0, len(out), 4)]
        else:
            rng = random.Random(text)
            n = min(max_tokens, rng.randint(60, 200))
            pieces = [(" " if i else "") + rng.choice(WORDS) for i in range(n)]
        pieces = pieces[:max_tokens]

        if not req.get("stream"):
            time.sleep(len(pieces) * self._itl())
            return self._json(200, {
                "id": "mock-1", "object": "chat.completion", "model": model,
                "choices": [{"index": 0, "finish_reason": "stop",
                             "message": {"role": "assistant", "content": "".join(pieces)}}],
                "usage": {"prompt_tokens": prompt_toks, "completion_tokens": len(pieces),
                          "total_tokens": prompt_toks + len(pieces),
                          "prompt_tokens_details": {"cached_tokens": prompt_toks - todo}}})

        # ---- DECODE: stream one piece per step, like a real server ----
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("Connection", "close")
        self.end_headers()
        try:
            for p in pieces:
                chunk = {"id": "mock-1", "object": "chat.completion.chunk", "model": model,
                         "choices": [{"index": 0, "delta": {"content": p}, "finish_reason": None}]}
                self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
                self.wfile.flush()
                time.sleep(self._itl())
            done = {"id": "mock-1", "object": "chat.completion.chunk", "model": model,
                    "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
            self.wfile.write(f"data: {json.dumps(done)}\n\ndata: [DONE]\n\n".encode())
            self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError):
            pass
        self.close_connection = True

    @staticmethod
    def _itl() -> float:
        """One decode step. Each extra user in the batch adds ~6% — the step is
        memory-bound, so sharing it is cheap. That is why batching raises throughput."""
        return CFG["itl_ms"] / 1000 * (1 + 0.06 * max(0, ACTIVE - 1))


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--slots", type=int, default=CFG["slots"], help="max concurrent sequences (batch size)")
    ap.add_argument("--itl-ms", type=float, default=CFG["itl_ms"], help="single-user decode step")
    a = ap.parse_args()
    CFG.update(slots=a.slots, itl_ms=a.itl_ms)
    global SLOTS
    SLOTS = threading.Semaphore(a.slots)
    print(f"mock LLM on http://localhost:{a.port}/v1  slots={a.slots}  itl={a.itl_ms}ms  (Ctrl-C to stop)")
    ThreadingHTTPServer(("0.0.0.0", a.port), Handler).serve_forever()


if __name__ == "__main__":
    main()

felab/results.py Python · 51 lines

"""
felab.results — every lesson appends its numbers to results/<lesson>.csv.

Why bother: the capstone (lesson 15) is a sizing memo built from YOUR measurements.
If every run lands in a CSV with the target and model next to it, the memo
writes itself and you can quote real numbers when explaining the result.
"""
from __future__ import annotations

import csv
import datetime as dt
from pathlib import Path

from .targets import REPO_ROOT

RESULTS = REPO_ROOT / "results"


def record(lesson: str, row: dict) -> Path:
    """Append one row (dict) to results/<lesson>.csv, adding a timestamp."""
    RESULTS.mkdir(exist_ok=True)
    path = RESULTS / f"{lesson}.csv"
    row = {"when": dt.datetime.now().isoformat(timespec="seconds"), **row}
    new = not path.exists()
    # If columns changed since the file was created, start a fresh file rather than corrupting it.
    if not new:
        with path.open() as f:
            header = next(csv.reader(f), [])
        if header != list(row):
            path = path.with_name(f"{path.stem}-{dt.datetime.now():%H%M%S}.csv")
            new = True
    with path.open("a", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(row))
        if new:
            w.writeheader()
        w.writerow(row)
    return path


def table(rows: list[dict], cols: list[str] | None = None) -> str:
    """Render rows as a fixed-width text table for the terminal."""
    if not rows:
        return "(no rows)"
    cols = cols or list(rows[0])
    fmt = lambda v: (f"{v:.3f}" if 0 < abs(v) < 1 else f"{v:,.1f}") if isinstance(v, float) else str(v)
    widths = {c: max(len(c), *(len(fmt(r.get(c, ""))) for r in rows)) for c in cols}
    line = "  ".join(c.ljust(widths[c]) for c in cols)
    out = [line, "  ".join("-" * widths[c] for c in cols)]
    for r in rows:
        out.append("  ".join(fmt(r.get(c, "")).rjust(widths[c]) for c in cols))
    return "\n".join(out)

felab/targets.py Python · 114 lines

"""
felab.targets — one place that knows where every model server lives.

Every lesson script takes the same three flags:

    --target  mock | ollama | llamacpp | mlx | fireworks | spark
    --model   override the target's default model id
    --base-url override the target's URL (rarely needed)

so the SAME script measures your laptop, Fireworks and the DGX Spark. That is
the habit the course is built on: one harness, many backends, comparable numbers.

Defaults can be changed in a `.env` file at the repo root (see .env.example).
"""
from __future__ import annotations

import argparse
import os
from dataclasses import dataclass
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent


def load_dotenv(path: Path = REPO_ROOT / ".env") -> None:
    """Tiny .env reader (KEY=value per line) so we need no extra dependency.
    Real environment variables always win over the file."""
    if not path.exists():
        return
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


load_dotenv()


@dataclass
class Target:
    name: str
    base_url: str
    model: str
    api_key: str
    note: str

    @property
    def is_paid(self) -> bool:
        return self.name == "fireworks"


def _targets() -> dict[str, Target]:
    env = os.environ.get
    spark_ip = env("SPARK_IP", "spark.local")
    return {
        # A fake server that ships with this repo (python -m felab.mock_server).
        # Lets you run every lesson on a train with no model downloaded.
        "mock": Target("mock", env("MOCK_URL", "http://localhost:9000/v1"), "mock-8b", "x",
                       "felab mock server — simulated latency, free, offline"),
        # Ollama: zero-config wrapper around llama.cpp.
        "ollama": Target("ollama", "http://localhost:11434/v1", env("OLLAMA_MODEL", "llama3.1:8b"), "x",
                         "Ollama on :11434"),
        # llama-server (llama.cpp): every knob exposed. It ignores the model field.
        "llamacpp": Target("llamacpp", "http://localhost:8080/v1", "local", "x",
                           "llama.cpp llama-server on :8080"),
        # mlx_lm.server: Apple-native. The model field must match what you served.
        "mlx": Target("mlx", env("MLX_URL", "http://localhost:8081/v1"),
                      env("MLX_MODEL", "mlx-community/Meta-Llama-3.1-8B-Instruct-4bit"), "x",
                      "mlx_lm.server on :8081"),
        # Fireworks serverless: pay per token. Check the model library for current ids.
        "fireworks": Target("fireworks", "https://api.fireworks.ai/inference/v1",
                            env("FIREWORKS_MODEL", "accounts/fireworks/models/gpt-oss-120b"),
                            env("FIREWORKS_API_KEY", ""), "Fireworks serverless (PAID)"),
        # vLLM / SGLang on the DGX Spark, reached over your LAN.
        "spark": Target("spark", env("SPARK_URL", f"http://{spark_ip}:8000/v1"),
                        env("SPARK_MODEL", "nvidia/Llama-3.1-8B-Instruct-FP8"), "x",
                        "vLLM/SGLang on the DGX Spark"),
    }


TARGETS = _targets()


def add_target_args(p: argparse.ArgumentParser, default: str | None = None) -> argparse.ArgumentParser:
    """Attach the standard --target/--model/--base-url flags to a script's parser."""
    p.add_argument("--target", default=default or os.environ.get("TARGET", "mock"), choices=TARGETS,
                   help="which server to talk to (default: $TARGET or mock)")
    p.add_argument("--model", help="override the target's default model id")
    p.add_argument("--base-url", help="override the target's URL")
    return p


def resolve(args: argparse.Namespace) -> Target:
    """Turn parsed flags into a Target, failing early with a helpful message."""
    t = TARGETS[args.target]
    t = Target(t.name, args.base_url or t.base_url, args.model or t.model, t.api_key, t.note)
    if t.is_paid and not t.api_key:
        raise SystemExit("FIREWORKS_API_KEY is not set. export it, or put it in .env (see lesson 00).")
    return t


def client(t: Target, asynchronous: bool = False):
    """An OpenAI SDK client pointed at the target. Every server here speaks the
    OpenAI chat-completions dialect, which is exactly why one harness works everywhere."""
    from openai import AsyncOpenAI, OpenAI  # imported late so --help works without the SDK
    cls = AsyncOpenAI if asynchronous else OpenAI
    return cls(base_url=t.base_url, api_key=t.api_key or "x", timeout=600)


def banner(t: Target) -> None:
    paid = "  $$ PAID ENDPOINT $$" if t.is_paid else ""
    print(f"→ target={t.name}  model={t.model}\n  {t.base_url}  ({t.note}){paid}\n")

felab/tickets.py Python · 68 lines

"""
felab.tickets — load the triage dataset and grade answers (lessons 07–11).

Three graders, from cheapest to most expensive — the eval pattern you'll pitch:
  1. parses   : is it JSON at all?
  2. valid    : does it match the schema (right keys, allowed enum, int in range)?
  3. correct  : does category (and severity) match the label?
(Lesson 09 adds a 4th: an LLM judge with a rubric, for fields with no single right answer.)
"""
from __future__ import annotations

import json
import re
from pathlib import Path

from .targets import REPO_ROOT

DATA = REPO_ROOT / "data"
SCHEMA = json.loads((DATA / "triage_schema.json").read_text())
SYSTEM = ("You triage customer support tickets. Reply with JSON only: "
          '{"category": "billing|outage|how-to|abuse", "severity": 1-5, "next_action": "<short>"}')


def load_test(n: int | None = None) -> list[dict]:
    """Held-out tickets with labels. Run `python data/make_tickets.py` if missing."""
    path = DATA / "tickets" / "test.jsonl"
    if not path.exists():
        raise SystemExit("data/tickets/test.jsonl missing — run: python data/make_tickets.py")
    rows = [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
    return rows[:n] if n else rows


def parse(text: str | None) -> dict | None:
    """Strict: the reply must BE a JSON object (whitespace and ```json fences tolerated).
    Being lenient here would hide exactly the failures an integration breaks on."""
    if not text:
        return None
    t = re.sub(r"^```(?:json)?\s*|\s*```$", "", text.strip())
    try:
        obj = json.loads(t)
        return obj if isinstance(obj, dict) else None
    except json.JSONDecodeError:
        return None


def schema_valid(obj: dict | None) -> bool:
    if obj is None:
        return False
    try:
        import jsonschema
        jsonschema.validate(obj, SCHEMA)
        return True
    except ImportError:  # minimal fallback check
        return (obj.get("category") in SCHEMA["properties"]["category"]["enum"]
                and isinstance(obj.get("severity"), int) and 1 <= obj["severity"] <= 5
                and isinstance(obj.get("next_action"), str))
    except Exception:
        return False


def grade(reply: str | None, row: dict) -> dict:
    obj = parse(reply)
    return {
        "parses": obj is not None,
        "valid": schema_valid(obj),
        "category_ok": bool(obj) and obj.get("category") == row["category"],
        "severity_ok": bool(obj) and obj.get("severity") == row["severity"],
    }