# Project export: Apprentice

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: OpenAI Build Week
- Tagline: Fine-tuning made small models cheap. Knowing when to trust one is now the expensive part. Apprentice makes the small model earn the job.
- Devpost: https://devpost.com/software/apprentice-e6twiq
- GitHub: https://github.com/singhabhishekkk/apprentice-build-week
- Demo: https://runapprentice.com/
- Video: https://www.youtube.com/embed/ZjQGyK_V4FU?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — Abhishek Singh (9 commits)

## Devpost submission (written by the team)

### Inspiration

Fine-tuning made small models cheap. Knowing when to trust one is now the expensive part. Swapping a frontier model for a small fine-tuned one is a two-hour job. Convincing an engineering lead that quality won't quietly fall over on the 3% of inputs nobody looked at is the part that never happens. So teams keep paying frontier prices for extraction and classification work a 4B model could do in its sleep and they aren't being irrational. They're being asked to bet on a number nobody produced. So we built the number.

### What it does

Apprentice watches a task your app already runs on a frontier model, then earns the job: Capture : two lines of SDK, and real production traffic starts landing. Verify : rows become gold (a human checked it) or silver (deterministic checks passed). Everything else stays raw and never counts. Optimize : DSPy GEPA rewrites the prompt against the verified set. Train : a small model is fine-tuned on gold rows only. Eval gate : every candidate is scored on held-out gold. It is promoted, or it is refused. There is no third option and no human override. Watch for drift : after takeover, the console charts captured traffic and the feedback your app reports, and offers a retrain only when enough new gold has arrived. The refuse path is the product. A model that fails the gate doesn't ship, and you keep paying the model that works. Everything else exists so we can say no with a straight face.

### How we built it

Codex is the implementing engineer on the backend, not an autocomplete. The workflow is written into the repo's AGENTS.md and followed every time: a bounded contract (the exact unit, the files to mirror, the guardrails, the gate commands), work isolated in a git worktree, Codex implements and runs mypy/ruff/pytest itself, then a mandatory human review reads the diff before anything reaches main. Codex does not commit. 152 Codex sessions. Codex runs on gpt-5.6-sol. GPT-5.6 also runs inside the product: the GEPA optimizer's student and reflection models are configurable, and this submission's run uses gpt-5.6-luna for both. The feature built during the window is the drift panel the surface that catches a model going quietly wrong in production.

### Challenges we ran into

The review kept catching what green tests could not. The retrain endpoint counted gold and silver rows as training data. Training reads gold only. Every test passed, because the tests encoded the same wrong contract I had handed to Codex. Two smaller bugs rode along: the cutoff used queue time instead of the moment the worker snapshots rows, and the eligibility flag ignored the minimum-rows gate so the panel would have offered a retrain the API rejects with a 400. A dead end, one click from the user. Then a worse one, and it was ours. The run report showed a cost of $0.796 next to token counts worth about $0.05. A DSPy cache hit is replayed from disk it never reaches the provider and costs nothing but the replayed entry still carries the original call's cost, and we were summing all of them. We had been billing for calls that never happened, on a panel whose entire job is to be trusted about money. Fixed; reported cost now matches the token math to the cent. And the thesis, demonstrated on us. Our public benchmark script first scored 72.92 instead of 100. GEPA rewrites a prompt from the feedback the metric returns, and ours only said "field names or values differ" so GEPA invented a separate key for each document type (po_number, receipt_number) while every gold answer uses invoice_id. It wrote a long, confident, carefully-reasoned, wrong prompt and was sure of itself. Naming the missing keys in the feedback took it to 100.00. A model optimizing against a vague signal converges on something plausible and wrong. That is exactly why the eval gate scores against human-verified data instead of a model's opinion of its own work.

### Accomplishments we're proud of

The eval gate refusing a bad model is a first-class demo beat, not a footnote. And every number is real or labeled projected. When we found our own cost figure inflating — making us look more expensive than we are we still stopped and fixed it before shipping.

### What we learned

Depth of agent collaboration is a review discipline, not a prompting trick. Codex ships production code when the brief carries the contract and a human reads the diff against the codebase instead of against the test output. Green gates prove the code ran. They do not prove it is right.

### What's next

Design partners: three teams through a real migration, structured extraction first, eval-gated end to end. Then the fine-tuned models get published with their benchmarks so anyone can check our arithmetic. Reproduce everything: https://github.com/singhabhishekkk/apprentice-build-week drift-demo/ runs in one command. benchmark/ reproduces the GPT-5.6 numbers with nothing of ours in the loop: baseline 57.87 → 100.00 on a 12-row held-out split, $0.047, 67s. Codex Session ID: 019f5eb6-27b4-7e00-af1b-04285e89a907

## README (from the GitHub repository)

<div align="center">

<img src="assets/logo.svg" width="96" alt="Apprentice logo">

# Apprentice

**The apprentice watches the expensive model work. Then earns the job.**

[![website](https://img.shields.io/badge/runapprentice.com-visit-EDE6D6)](https://runapprentice.com)
[![license](https://img.shields.io/badge/license-MIT-green)](LICENSE)
[![sdk](https://img.shields.io/badge/pip-runapprentice-3775A9)](https://pypi.org/project/runapprentice/)
[![built with](https://img.shields.io/badge/built%20with-Codex-412991)](https://openai.com/codex)
[![models](https://img.shields.io/badge/GPT--5.6-sol%20%C2%B7%20luna-10A37F)](https://platform.openai.com/docs/models)
[![every number](https://img.shields.io/badge/every%20number-measured-orange)](#the-numbers)

**OpenAI Build Week · Developer Tools**

[![Watch the demo film](https://img.youtube.com/vi/ZjQGyK_V4FU/maxresdefault.jpg)](https://youtu.be/ZjQGyK_V4FU)

**[▶ Watch the demo film](https://youtu.be/ZjQGyK_V4FU)** — 1:47, the whole loop, ending on
the eval gate refusing a model.

</div>

---

## About this repository

**The production codebase is a private monorepo. This repository is the Build Week
artifact, and it is public.**

Saying that plainly up front, because the alternative would be to imply this is the whole
system, and it is not. What is here is real and runnable:

- the **feature we shipped during the submission window**, running standalone,
- the **benchmark that reproduces every GPT-5.6 number we claim**, with nothing of ours
  in the loop,
- an honest account of **how Codex was used**, and which work is new this week.

The product itself is live. You do not need our source to test it: sign in at
[runapprentice.com](https://runapprentice.com), `pip install runapprentice`, and the
console will show your own traffic. Instructions below.

## What is in here

| Piece | What it does | Needs |
|---|---|---|
| [`drift-demo/`](drift-demo) | **The feature built during the window**, running standalone: the drift endpoints and the console page, over a seeded SQLite database. `seed.py` writes 30 days of traffic whose feedback score decays from 0.95 to 0.55, so you can watch a model quietly get worse. | nothing |
| [`benchmark/`](benchmark) | **Reproduces the GPT-5.6 numbers.** Talks to OpenAI and DSPy directly. No Apprentice API, no account, no database. Bring a key and check our arithmetic. | `OPENAI_API_KEY` |
| [`drift-panel/`](drift-panel) | The **production source** of the two endpoints and their tests, lifted verbatim from the private monorepo, so the code being judged is the code that shipped. | reading only |

---

## What Apprentice is

Everyone can make a model cheaper. Nobody can tell you when the cheap one is *safe*.

That is the whole problem. Swapping a frontier model for a small fine-tuned one is a
two-hour job; convincing an engineering lead that quality will not quietly collapse on the
3% of inputs nobody looked at is the part that never happens. So teams keep paying frontier
prices for extraction and classification work a 4B model could do in its sleep.

Apprentice watches a task your app already runs on a frontier model, then earns the job:

1. **Capture** — two lines of SDK, and real production traffic starts landing.
2. **Verify** — rows become **gold** (a human checked it) or **silver** (deterministic
   checks passed). Everything else stays raw and never counts.
3. **Optimize** — DSPy GEPA rewrites the prompt against the verified set.
4. **Train** — a small model is fine-tuned on gold rows only.
5. **Eval gate** — every candidate is scored on held-out gold. It is promoted, or it is
   **refused**. There is no third option and no human override.
6. **Watch for drift** — after takeover, the panel shows captured traffic, the feedback
   your app reports, and offers a retrain only when enough new gold has arrived to be
   worth it.

The refuse path is the product. A model that fails the gate does not ship, and you keep
paying the model that works.

---

## Built with Codex

Codex is the implementing engineer on the backend, not an autocomplete. The process is
written into the monorepo's `AGENTS.md` and followed every time:

1. **A bounded contract, not a vibe.** The brief names the exact unit of work, the files to
   mirror, the hard guardrails, and the gate commands (mypy, ruff, pytest).
2. **Isolation.** Feature work runs in a dedicated git worktree on its own branch, so a bad
   run cannot touch `main`.
3. **Codex writes and verifies.** It implements, runs the gates itself, and reports what
   passed. It does not commit; a human does.
4. **A senior review is mandatory.** Green gates are necessary and never sufficient.

**Codex Session ID:** `019f5eb6-27b4-7e00-af1b-04285e89a907`
*(This is the Codex thread. The Devpost field calls it a "/feedback Session ID"; it has
nothing to do with this project's own `POST /v1/feedback` endpoint.)*

**Scale:** 152 Codex sessions on the monorepo between 2026-06-10 and 2026-07-14.

### The rule that earned its keep

On this very feature, Codex's tests went green and the code was still wrong three ways —
all caught by the human review, none caught by the gates:

1. The endpoint counted **gold + silver** rows toward retraining. **Training reads gold
   only**, so silver rows would never reach the model. The panel would have overstated the
   retrain payload — on a surface whose entire job is to be trusted.
2. The cutoff used the job's **queue** time instead of the moment the worker **snapshots**
   rows, so rows created during a long training run were miscounted.
3. `eligible` ignored the minimum-rows gate, so the panel would have offered a one-click
   retrain that the API rejects with a 400. A dead end, one click away from the user.

The tests passed because **they encoded the same wrong contract the brief did.** The brief
was mine, not Codex's. The lesson is that a spec must be written against the function that
enforces the rule, not against a doc that describes it.

Codex is also a good engineer inside the thread. In the session above it root-caused its
own failing test run (the reused venv's editable install pointed at the main checkout, so
pytest was exercising the old API) and caught a UTC bucketing bug in its own diff during
self-review, before handing anything back.

### GPT-5.6 inside the product

The GEPA optimizer's student and reflection models are configurable. The submission run
uses **`gpt-5.6-luna`** for both. Reproduce it yourself: [`benchmark/`](benchmark).

Codex itself runs on **`gpt-5.6-sol`** at medium effort.

---

## New this week, and what predates it

The rules require this distinction, so here it is without spin.

| | |
|---|---|
| **Built during the window (Jul 13–21)** | **The drift panel.** Two read-only endpoints and the console page that reads them. It wires the previously orphaned `POST /v1/feedback` signal into a surface a user can act on, and closes the loop after a model takes over. Written by driving Codex; see the session above. |
| **Also in-window** | The first scored GEPA run on `gpt-5.6-luna`, and a real bug it exposed: our run-cost accounting was billing for DSPy **cache hits** — LLM calls that never reached the provider — overstating cost ~15x. Fixed, and the run below is the verification. |
| **Predates the window** | Everything else: capture, the tier system, the GEPA optimizer, fine-tuning, the eval gate, the console, the SDK, the docs site. Apprentice has been in development since June 2026. |

---

## The numbers

Measured on 2026-07-14. Nothing here is projected.

| | |
|---|---|
| Baseline prompt (held out) | **50.17** |
| GEPA-optimized (held out) | **100.00** |
| Rows improved / regressed | **12 of 12 / 0** |
| Wall time | **67 seconds** |
| Cost | **$0.047** |
| Models | `gpt-5.6-luna`, student and reflection |
| Metric | deterministic JSON field F1 — no LLM judge marks its own homework |
| Split | 24 rows, seed 42, 12-row holdout |

Those numbers come from **Apprentice's own optimizer**, which i

[README truncated for size]

## Detected evidence (automated analysis)

Indexed codebase: 8 recognized source files, 53 KB.
- FastAPI (technology) — detected in the code
- HTML (language) — detected in the code
- OpenAI (technology) — detected in the code
- Python (language) — detected in the code
- Next.js (technology) — claimed on Devpost, not found in the code
- PostgreSQL (technology) — claimed on Devpost, not found in the code
- React (technology) — claimed on Devpost, not found in the code
- Supabase (technology) — claimed on Devpost, not found in the code
- Tailwind CSS (technology) — claimed on Devpost, not found in the code
- TypeScript (language) — claimed on Devpost, not found in the code

## Codebase structure (from repository index)

### Files (13 of 13)

```
.gitignore
assets/cover.html
benchmark/gepa_bench.py
benchmark/golden.csv
benchmark/pyproject.toml
drift-demo/app.py
drift-demo/pyproject.toml
drift-demo/seed.py
drift-demo/static/index.html
drift-panel/endpoints.py
drift-panel/test_endpoints.py
LICENSE
README.md
```

### Dependencies

- benchmark/pyproject.toml: dspy@>=3.0.0, openai@>=1.0.0
- drift-demo/pyproject.toml: fastapi@>=0.110, sqlalchemy@>=2.0, uvicorn@>=0.27

### Recent commits (newest first)

- docs: make the feedback example use sync capture, async returns no trace_id
- assets: gallery at 4K, rendered rather than screen-grabbed
- assets: re-grab the gallery frames at settled scene midpoints
- assets: logo-forward cover
- assets: Devpost gallery images (3:2)
- docs: link the demo film
- docs: the benchmark reproduces 100.00, and the 72.92 story stays
- fix: the benchmark does not reproduce the production number, and the README now says so
- Apprentice for OpenAI Build Week

## Key source files (fetched from GitHub, selected and truncated for size)

### benchmark/pyproject.toml

```
[project]
name = "apprentice-gepa-bench"
version = "1.0.0"
description = "Reproduce Apprentice's GPT-5.6 prompt-optimization numbers"
requires-python = ">=3.11"
dependencies = ["dspy>=3.0.0", "openai>=1.0.0"]

```

### drift-demo/pyproject.toml

```
[project]
name = "apprentice-drift-demo"
version = "1.0.0"
description = "The Apprentice drift panel, standalone"
requires-python = ">=3.11"
dependencies = ["fastapi>=0.110", "uvicorn>=0.27", "sqlalchemy>=2.0"]

```

### drift-demo/app.py

```python
"""The drift panel, standalone and runnable.

This is the feature Apprentice shipped during OpenAI Build Week (2026-07-13 to 07-21),
written by driving Codex. The production version lives in a private monorepo behind auth,
multi-tenancy and a job queue. This file is the same two endpoints and the same rules,
over a seeded SQLite database, so a judge can run it in one command and see it work.

The rules it enforces are the ones that matter, and they are not cosmetic:

  * `drift` counts only rows whose source is a captured trace. Uploads and synthetic rows
    are not production traffic and must never appear on a traffic chart.

  * `retrain-candidate` counts GOLD rows only. Training reads gold only, so counting
    silver would overstate what a retrain would actually learn from. (This is the bug the
    human review caught after Codex's tests went green: the tests encoded the same wrong
    contract the brief did. See the README.)

  * The cutoff for "new" rows is when the last training job STARTED, not when it was
    queued: the worker snapshots rows at start, so anything created after that was never
    seen by the model.

  * `eligible` is false unless a retrain would actually clear the same gate the training
    endpoint enforces (MIN_TRAIN_ROWS). Offering a button that the API would reject with a
    400 is a dead end, and a dead end is a bug.

Run:  uv run uvicorn app:app --reload    (then open http://localhost:8000)
"""

from __future__ import annotations

from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from sqlalchemy import (
    Boolean,
    DateTime,
    Float,
    ForeignKey,
    String,
    Text,
    create_engine,
    func,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

DB_PATH = Path(__file__).parent / "drift-demo.db"
ENGINE = create_engine(f"sqlite:///{DB_PATH}")

# The production gate. A task cannot train below this many gold rows, so the panel must
# not offer a retrain below it either.
MIN_TRAIN_ROWS = 500


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[str] = mapped_column(String(32), primary_key=True)
    name: Mapped[str] = mapped_column(String(120), unique=True)


class Row(Base):
    """One captured input/output pair.

    `tier` is the trust level: gold means a human verified it, silver means it passed
    deterministic checks, raw means neither. `source` distinguishes production traffic
    (trace) from an upload.
    """

    __tablename__ = "rows"
    id: Mapped[str] = mapped_column(String(32), primary_key=True)
    task_id: Mapped[str] = mapped_column(ForeignKey("tasks.id"), index=True)
    input: Mapped[str] = mapped_column(Text)
    output: Mapped[str] = mapped_column(Text)
    tier: Mapped[str] = mapped_column(String(10), default="raw", index=True)
    source: Mapped[str] = mapped_column(String(20), default="upload")
    feedback_score: Mapped[float | None] = mapped_column(Float, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))


class Job(Base):
    __tablename__ = "jobs"
    id: Mapped[str] = mapped_column(String(32), primary_key=True)
    task_id: Mapped[str] = mapped_column(ForeignKey("tasks.id"), index=True)
    kind: Mapped[str] = mapped_column(String(20))
    status: Mapped[str] = mapped_column(String(20))
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))


app = FastAPI(title="Apprentice drift panel (Build Week demo)")


def _utc(value: datetime) -> datetime:
    """SQLite hands back naive datetimes. Treat them as UTC rather than local time.

    Getting this wrong shifts every row into the wrong day bucket, which is exactly the
    kind of quiet error a drift chart would show as real behaviour.
    """
    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)


def _iso(value: datetime | None) -> str | None:
    return _utc(value).isoformat() if value else None


def _task(session: Session, name: str) -> Task:
    task = session.query(Task).filter(Task.name == name).one_or_none()
    if task is None:
        raise HTTPException(status_code=404, detail=f"No task named {name!r}. Run `python seed.py`.")
    return task


@app.get("/v1/tasks/{name}/drift")
def task_drift(name: str, days: int = 30) -> dict[str, Any]:
    """Daily captured traffic and the feedback your app reported, over 1 to 90 days.

    Every day in the window is present, zero-filled. A quiet day is a real fact and must
    read as a quiet day, not as a gap in the line.
    """
    days = max(1, min(days, 90))
    today = datetime.now(timezone.utc).date()
    start_date = today - timedelta(days=days - 1)
    start_at = datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc)
    end_at = datetime.combine(today + timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc)

    series: dict[date, dict[str, Any]] = {
        start_date
        + timedelta(days=offset): {
            "date": (start_date + timedelta(days=offset)).isoformat(),
            "traces": 0,
            "feedback_count": 0,
            "mean_feedback_score": None,
        }
        for offset in range(days)
    }

    with Session(ENGINE) as session:
        task = _task(session, name)
        traces = (
            session.query(Row.created_at, Row.feedback_score)
            .filter(
                Row.task_id == task.id,
                Row.source == "trace",  # production traffic only
                Row.created_at >= start_at,
                Row.created_at < end_at,
            )
            .all()
        )

    scores: dict[date, list[float]] = {}
    for created_
[truncated — 2351 more characters]
```

### assets/cover.html

```html
<!doctype html>
<html>
  <head>
    <meta charset="utf-8" />
    <style>
      /* Variant B: the real logo tile at full size, then the thesis, then the proof.
         Brand is unmissable; the verdict chips still carry the argument. 1200x800 (3:2). */
      @import url("https://fonts.googleapis.com/css2?family=Geist:wght@400;500;700&display=swap");
      * { margin: 0; padding: 0; box-sizing: border-box; }
      body {
        width: 1200px;
        height: 800px;
        background: #0a0a0b;
        font-family: Geist, ui-sans-serif, -apple-system, sans-serif;
        color: #ededef;
        display: flex;
        flex-direction: column;
        justify-content: center;
        padding: 0 76px;
        position: relative;
        overflow: hidden;
      }
      .glow {
        position: absolute;
        width: 1000px;
        height: 760px;
        right: -240px;
        top: 20px;
        background: radial-gradient(closest-side, rgba(127, 176, 105, 0.15), transparent);
      }
      .top { display: flex; align-items: center; gap: 22px; margin-bottom: 34px; }
      .logo { width: 92px; height: 92px; border-radius: 20px; }
      .name { font-size: 44px; font-weight: 700; letter-spacing: -0.02em; }
      .name small {
        display: block;
        font-size: 17px;
        font-weight: 500;
        letter-spacing: 0.14em;
        text-transform: uppercase;
        color: #66666e;
        margin-top: 6px;
      }
      h1 {
        font-size: 47px;
        font-weight: 700;
        line-height: 1.16;
        letter-spacing: -0.02em;
        max-width: 1010px;
      }
      h1 em { font-style: normal; color: #7fb069; }
      .verdicts { display: flex; gap: 16px; margin-top: 40px; }
      .v {
        display: flex;
        align-items: baseline;
        gap: 12px;
        padding: 15px 23px;
        border-radius: 13px;
        border: 2px solid;
        font-variant-numeric: tabular-nums;
      }
      .v b { font-size: 36px; font-weight: 700; }
      .v span { font-size: 17px; font-weight: 500; letter-spacing: 0.06em; text-transform: uppercase; }
      .refused { border-color: rgba(210, 105, 78, 0.55); background: rgba(210, 105, 78, 0.09); }
      .refused b, .refused span { color: #d2694e; }
      .promoted { border-color: rgba(127, 176, 105, 0.6); background: rgba(127, 176, 105, 0.12); }
      .promoted b, .promoted span { color: #7fb069; }
      .foot {
        position: absolute;
        bottom: 40px;
        left: 76px;
        font-size: 21px;
        font-weight: 500;
        color: #66666e;
      }
    </style>
  </head>
  <body>
    <div class="glow"></div>

    <div class="top">
      <svg class="logo" viewBox="0 0 100 100" xmlns="http://www.w3.org/2000/svg">
        <rect width="100" height="100" rx="22" fill="#1B1A16" />
        <g transform="translate(50 50) scale(0.82) translate(-50 -50)">
          <circle cx="46" cy="46" r="34" fill="#EDE6D6" />
          <circle cx="74" cy="74" r="14" fill="#7FB069" />
        </g>
      </svg>
      <div class="name">Apprentice<small>Developer Tools</small></div>
    </div>

    <h1>Fine-tuning made small models cheap.<br /><em>Knowing when to trust one</em> is now the expensive part.</h1>

    <div class="verdicts">
      <div class="v refused"><b>42.50</b><span>refused</span></div>
      <div class="v promoted"><b>89.17</b><span>promoted</span></div>
    </div>

    <div class="foot">runapprentice.com</div>
  </body>
</html>

```

### drift-panel/endpoints.py

```python
"""The drift panel endpoints, verbatim from the production monorepo.

Copied from `apprentice-api/src/apprentice_api/main.py` so that the code being judged is
the code that actually ships, not a reimplementation. It will not run on its own: it needs
the surrounding app (auth, the DB session, the collaborator ACL). For something runnable,
see ../drift-demo.

Written by driving Codex (session 019f5eb6-27b4-7e00-af1b-04285e89a907), then corrected in
human review. The three defects that review caught, which the green tests did not, are
described in the README.
"""

@app.get("/v1/tasks/{task_id}/drift")
def task_drift(
    task_id: str,
    days: int = 30,
    user: AuthUser = Depends(get_current_user),
) -> dict[str, Any]:
    from apprentice_api import collaborators

    days = max(1, min(days, 90))
    today = datetime.now(timezone.utc).date()
    start_date = today - timedelta(days=days - 1)
    start_at = datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc)
    end_at = datetime.combine(today + timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc)
    series: dict[date, dict[str, Any]] = {
        start_date + timedelta(days=offset): {
            "date": (start_date + timedelta(days=offset)).isoformat(),
            "traces": 0,
            "feedback_count": 0,
            "mean_feedback_score": None,
        }
        for offset in range(days)
    }

    with db.get_session() as session:
        collaborators.authorize_task_by_id(session, user, task_id, "read")
        traces = (
            session.query(db.Row.created_at, db.Row.feedback_score)
            .filter(
                db.Row.task_id == task_id,
                db.Row.source == "trace",
                db.Row.created_at >= start_at,
                db.Row.created_at < end_at,
            )
            .all()
        )

    feedback_scores: dict[date, list[float]] = {}
    for created_at, feedback_score in traces:
        created_at_utc = (
            created_at.replace(tzinfo=timezone.utc)
            if created_at.tzinfo is None
            else created_at.astimezone(timezone.utc)
        )
        day = created_at_utc.date()
        series[day]["traces"] += 1
        if feedback_score is not None:
            series[day]["feedback_count"] += 1
            feedback_scores.setdefault(day, []).append(feedback_score)
    for day, scores in feedback_scores.items():
        series[day]["mean_feedback_score"] = sum(scores) / len(scores)

    return {"task_id": task_id, "days": days, "series": list(series.values())}


@app.get("/v1/tasks/{task_id}/retrain-candidate")
def retrain_candidate(
    task_id: str, user: AuthUser = Depends(get_current_user)
) -> dict[str, Any]:
    """Gold rows added since the last successful train, and whether a retrain would pass its gate.

    Gold only: both the train gate and the train worker select `tier == "gold"` rows,
    so silver rows would never reach the model and must not be counted here.
    """
    from apprentice_api import collaborators

    with db.get_session() as session:
        collaborators.authorize_task_by_id(session, user, task_id, "read")
        last_train = (
            session.query(db.Job)
            .filter(
                db.Job.task_id == task_id,
                db.Job.kind == "train",
                db.Job.status == "succeeded",
            )
            .order_by(func.coalesce(db.Job.finished_at, db.Job.created_at).desc())
            .first()
        )
        gold = session.query(func.count(db.Row.id)).filter(
            db.Row.task_id == task_id,
            db.Row.tier == "gold",
        )
        total_gold = gold.scalar() or 0
        new_gold = total_gold
        if last_train is not None:
            # The worker snapshots rows when the job starts, so rows created after that
            # start were never trained on. ponytail: `created_at` still misses rows
            # re-tiered to gold after the run; add a `verified_at` column to close that.
            snapshot_at = last_train.started_at or last_train.created_at
            new_gold = gold.filter(db.Row.created_at > snapshot_at).scalar() or 0

    return {
        "task_id": task_id,
        "last_train_job_id": last_train.id if last_train else None,
        "last_trained_at": _iso(last_train.finished_at or last_train.created_at)
        if last_train
        else None,
        "new_gold_rows": new_gold,
        "total_gold_rows": total_gold,
        "min_train_rows": MIN_TRAIN_ROWS,
        # A retrain only helps if there is new gold data, and it only starts if the task
        # clears the same gate POST /v1/tasks/{task}/train enforces.
        "eligible": new_gold > 0 and total_gold >= MIN_TRAIN_ROWS,
    }

```

### drift-demo/seed.py

```python
"""Seeds a task whose quality is visibly decaying, which is the thing a drift panel is for.

30 days of captured traffic on a support-triage task. The model was trained on day 18 and
was good; since then the inputs have drifted and the feedback your app reports has fallen
from ~0.95 to ~0.55. Enough gold rows have accumulated since that training run that a
retrain is now worth doing, and the panel says so.

The numbers are generated, not measured. They are here so the endpoints have something
real to chart; nothing in this file is presented as a benchmark result. The measured
numbers in this repo are the ones in benchmark/, which you can reproduce yourself.

Run:  uv run python seed.py
"""

from __future__ import annotations

import random
from datetime import datetime, timedelta, timezone
from uuid import uuid4

from app import DB_PATH, ENGINE, Base, Job, Row, Task
from sqlalchemy.orm import Session

TASK = "support-triage"
DAYS = 30
TRAINED_DAYS_AGO = 12

rng = random.Random(42)  # deterministic: two people running this see the same chart


def _id() -> str:
    return uuid4().hex[:32]


def main() -> None:
    if DB_PATH.exists():
        DB_PATH.unlink()
    Base.metadata.create_all(ENGINE)

    now = datetime.now(timezone.utc)
    task = Task(id=_id(), name=TASK)

    rows: list[Row] = []
    for day in range(DAYS):
        day_start = (now - timedelta(days=DAYS - 1 - day)).replace(hour=0, minute=0, second=0, microsecond=0)

        # Traffic grows slowly, with weekends quieter. A flat line would look synthetic.
        weekday = day_start.weekday()
        base = 28 + day
        traffic = int(base * (0.45 if weekday >= 5 else 1.0) * rng.uniform(0.85, 1.15))

        # The story: quality holds, then drifts down after the model was trained.
        progress = day / (DAYS - 1)
        quality = 0.95 - 0.40 * max(0.0, (progress - 0.35)) / 0.65

        for _ in range(traffic):
            created = day_start + timedelta(
                hours=rng.randint(7, 21), minutes=rng.randint(0, 59), seconds=rng.randint(0, 59)
            )
            # Only about half of production traffic gets rated by the app. Pretending every
            # call comes back with a score would be a lie about how integrations behave.
            rated = rng.random() < 0.45
            score = None
            if rated:
                score = max(0.0, min(1.0, rng.gauss(quality, 0.12)))

            # A slice of traffic gets verified by a human and becomes gold.
            tier = "gold" if rng.random() < 0.22 else "raw"

            rows.append(
                Row(
                    id=_id(),
                    task_id=task.id,
                    input="Customer email about a delayed refund on order #%d" % rng.randint(1000, 9999),
                    output='{"intent": "refund_status", "priority": "normal"}',
                    tier=tier,
                    source="trace",
                    feedback_score=score,
                    created_at=created,
                )
            )

    # Enough historical gold to clear MIN_TRAIN_ROWS, so the panel has a real decision to
    # make rather than being blocked on data volume.
    for _ in range(520):
        rows.append(
            Row(
                id=_id(),
                task_id=task.id,
                input="Historical verified example",
                output='{"intent": "refund_status", "priority": "normal"}',
                tier="gold",
                source="upload",
                feedback_score=None,
                created_at=now - timedelta(days=DAYS + rng.randint(1, 40)),
            )
        )

    started = now - timedelta(days=TRAINED_DAYS_AGO)
    train = Job(
        id=_id(),
        task_id=task.id,
        kind="train",
        status="succeeded",
        started_at=started,
        finished_at=started + timedelta(minutes=41),
        created_at=started - timedelta(minutes=3),
    )

    # Count BEFORE the commit: SQLAlchemy expires instances on commit, so reading these
    # attributes afterwards detaches and raises.
    traces = sum(1 for r in rows if r.source == "trace")
    rated = sum(1 for r in rows if r.feedback_score is not None)
    gold = sum(1 for r in rows if r.tier == "gold")
    new_gold = sum(1 for r in rows if r.tier == "gold" and r.created_at > started)

    with Session(ENGINE) as session:
        session.add(task)
        session.add(train)
        session.add_all(rows)
        session.commit()

    print(f"task           {TASK}")
    print(f"traces (30d)   {traces}")
    print(f"rated by app   {rated}")
    print(f"gold rows      {gold}  ({new_gold} since the last training run started)")
    print(f"last trained   {TRAINED_DAYS_AGO} days ago")
    print("\nuv run uvicorn app:app  ->  http://localhost:8000")


if __name__ == "__main__":
    main()

```

### drift-panel/test_endpoints.py

```python
"""The tests for those endpoints, verbatim from the production monorepo.

Copied from `apprentice-api/tests/test_api.py`. They rely on the app's fixtures, so they do
not run standalone; they are here to be read.

Worth reading the retrain-candidate tests specifically. An earlier version of them passed
against WRONG code, because they encoded the same wrong contract the brief handed to Codex
(gold + silver, when training reads gold only). Green tests prove the code runs. They do
not prove it is right.
"""


def test_drift_panel_zero_data_task(client: TestClient) -> None:
    task_id = client.post("/v1/tasks", json={"name": "empty"}).json()["id"]

    drift = client.get(f"/v1/tasks/{task_id}/drift").json()
    candidate = client.get(f"/v1/tasks/{task_id}/retrain-candidate").json()

    assert drift["days"] == 30
    assert len(drift["series"]) == 30
    assert all(
        day["traces"] == 0
        and day["feedback_count"] == 0
        and day["mean_feedback_score"] is None
        for day in drift["series"]
    )
    assert candidate == {
        "task_id": task_id,
        "last_train_job_id": None,
        "last_trained_at": None,
        "new_gold_rows": 0,
        "total_gold_rows": 0,
        "min_train_rows": 2,
        "eligible": False,
    }


def test_drift_panel_routes_hide_non_owner_task(client: TestClient) -> None:
    task_id = client.post("/v1/tasks", json={"name": "private-drift"}).json()["id"]
    _as_user("user-test-2", "b@example.com")

    assert client.get(f"/v1/tasks/{task_id}/drift").status_code == 404
    assert client.get(f"/v1/tasks/{task_id}/retrain-candidate").status_code == 404


def test_retrain_candidate_without_succeeded_train_counts_all_gold(
    client: TestClient,
) -> None:
    from apprentice_api import db

    task_id = client.post("/v1/tasks", json={"name": "never-trained"}).json()["id"]
    with db.get_session() as session:
        session.add_all(
            [
                db.Row(task_id=task_id, input="g1", output="o", tier="gold"),
                db.Row(task_id=task_id, input="g2", output="o", tier="gold"),
                db.Row(task_id=task_id, input="s", output="o", tier="silver"),
                db.Row(task_id=task_id, input="r", output="o", tier="raw"),
                db.Job(task_id=task_id, kind="train", status="failed"),
            ]
        )
        session.commit()

    body = client.get(f"/v1/tasks/{task_id}/retrain-candidate").json()

    # Training reads gold only, so silver and raw never count towards a retrain.
    assert body["last_train_job_id"] is None
    assert body["new_gold_rows"] == 2
    assert body["total_gold_rows"] == 2
    assert body["eligible"] is True


def test_retrain_candidate_not_eligible_below_min_train_rows(client: TestClient) -> None:
    from apprentice_api import db

    task_id = client.post("/v1/tasks", json={"name": "too-thin"}).json()["id"]
    with db.get_session() as session:
        session.add(db.Row(task_id=task_id, input="g", output="o", tier="gold"))
        session.commit()

    body = client.get(f"/v1/tasks/{task_id}/retrain-candidate").json()

    # New gold data exists, but POST /train would reject it, so the panel must not offer a retrain.
    assert body["new_gold_rows"] == 1
    assert body["total_gold_rows"] == 1
    assert body["min_train_rows"] == 2
    assert body["eligible"] is False


def test_retrain_candidate_counts_gold_rows_created_after_latest_train_started(
    client: TestClient,
) -> None:
    from apprentice_api import db

    task_id = client.post("/v1/tasks", json={"name": "trained"}).json()["id"]
    now = datetime.now(timezone.utc)
    latest_started = now - timedelta(days=2)
    latest_finished = now - timedelta(days=2) + timedelta(hours=1)
    with db.get_session() as session:
        older_train = db.Job(
            task_id=task_id,
            kind="train",
            status="succeeded",
            created_at=now - timedelta(days=5),
            started_at=now - timedelta(days=5),
            finished_at=now - timedelta(days=4),
        )
        latest_train = db.Job(
            task_id=task_id,
            kind="train",
            status="succeeded",
            created_at=now - timedelta(days=3),
            started_at=latest_started,
            finished_at=latest_finished,
        )
        session.add_all(
            [
                older_train,
                latest_train,
                # Queued before the run started, so it was in the training snapshot.
                db.Row(
                    task_id=task_id,
                    input="trained-gold",
                    output="o",
                    tier="gold",
                    created_at=now - timedelta(days=4),
                ),
                db.Row(
                    task_id=task_id,
                    input="new-gold-1",
                    output="o",
                    tier="gold",
                    created_at=now - timedelta(days=1),
                ),
                db.Row(
                    task_id=task_id,
                    input="new-gold-2",
                    output="o",
                    tier="gold",
                    created_at=now,
                ),
                db.Row(
                    task_id=task_id,
                    input="new-silver",
                    output="o",
                    tier="silver",
                    created_at=now,
                ),
            ]
        )
        session.commit()
        latest_train_id = latest_train.id

    body = client.get(f"/v1/tasks/{task_id}/retrain-candidate").json()

    assert body["last_train_job_id"] == latest_train_id
    assert body["last_trained_at"].startswith(latest_finished.replace(tzinfo=None).isoformat())
    # Cutoff is when the run started (the worker snapshots rows then), not when it was queued.
    assert body["new_gold_rows"] == 2
    assert body["total_gold_rows"] == 3
    assert body["eligible"] is True


# -- auth -------------------------------------------------
[truncated — 22 more characters]
```

### benchmark/gepa_bench.py

```python
"""Reproduce the GPT-5.6 prompt-optimization numbers, with nothing of ours in the loop.

This talks to OpenAI and to DSPy. It does NOT need Apprentice's API, our database, or an
account. Bring your own key and check our arithmetic.

    export OPENAI_API_KEY=sk-...
    uv run python gepa_bench.py                 # gpt-5.6-luna, the cheapest 5.6
    uv run python gepa_bench.py --model gpt-5.6-terra

What it does, which is what Apprentice's optimizer does:

  1. Splits the 24-row invoice extraction set with seed 42 into train/val/holdout.
  2. Scores a deliberately weak baseline prompt on the HELD-OUT rows.
  3. Runs DSPy GEPA, which rewrites the prompt using the train and val rows only.
  4. Scores the rewritten prompt on the same held-out rows.

The score is deterministic JSON field F1: parse the output as JSON, compare leaf fields
against the gold answer. No LLM judge marks its own homework.

Expect the shape, not the digits. GEPA is stochastic and the baseline is a weak prompt, so
the starting score moves between runs (we have seen 50.2 and 56.0). What is stable is that
the optimized prompt scores 100 on the holdout. Our own run, on 2026-07-14:

    baseline 50.17  ->  optimized 100.00     12/12 held-out rows improved, 0 regressed
    67 seconds, $0.047, gpt-5.6-luna as both student and reflection model
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import random
import time
from pathlib import Path
from typing import Any

import dspy

GOLDEN = Path(__file__).parent / "golden.csv"
SEED = 42

# The prompt a developer actually starts with, before anyone has thought about it. The
# point of the exercise is that you should not have to think about it.
BASELINE_PROMPT = "Extract the fields from the text. Return JSON."


def leaf_pairs(obj: Any, prefix: str = "") -> set[tuple[str, str]]:
    """Flatten a JSON object to (path, value) pairs so nested answers score fairly."""
    pairs: set[tuple[str, str]] = set()
    if isinstance(obj, dict):
        for key, value in obj.items():
            pairs |= leaf_pairs(value, f"{prefix}.{key}" if prefix else str(key))
    elif isinstance(obj, list):
        for i, value in enumerate(obj):
            pairs |= leaf_pairs(value, f"{prefix}[{i}]")
    else:
        # Numbers compare by value, not by formatting: 42.10 and 42.1 are the same answer.
        if isinstance(obj, float) and obj.is_integer():
            obj = int(obj)
        pairs.add((prefix, str(obj).strip().lower()))
    return pairs


def field_f1(expected: str, actual: str) -> float:
    try:
        want = json.loads(expected)
    except json.JSONDecodeError:
        return 0.0
    try:
        got = json.loads(extract_json(actual))
    except (json.JSONDecodeError, TypeError):
        # Not JSON at all is a zero. A model that ignores the format has not done the task.
        return 0.0
    if not isinstance(want, dict) or not isinstance(got, dict):
        return 0.0

    w, g = leaf_pairs(want), leaf_pairs(got)
    if not w and not g:
        return 1.0
    overlap = len(w & g)
    if overlap == 0:
        return 0.0
    precision = overlap / len(g)
    recall = overlap / len(w)
    return 2 * precision * recall / (precision + recall)


def extract_json(text: str) -> str:
    """Models like to wrap JSON in prose or fences. Take the outermost object."""
    text = (text or "").strip()
    if text.startswith("```"):
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
    start, end = text.find("{"), text.rfind("}")
    return text[start : end + 1] if start != -1 and end > start else text


class Extract(dspy.Signature):
    """Extract the requested fields from the text and return them as JSON."""

    text: str = dspy.InputField()
    extracted: str = dspy.OutputField()


def metric(example: dspy.Example, prediction: Any, *_: Any) -> Any:
    """Score a prediction, and tell GEPA precisely what was wrong with it.

    The feedback is not decoration. GEPA rewrites the prompt from it, so vague feedback
    produces a vague prompt. An earlier version of this file said only "field names or
    values differ", and GEPA duly invented a different key per document type
    (`po_number`, `receipt_number`, ...) while the gold answers always use `invoice_id`.
    It scored 72.92 and the prompt it wrote was confidently wrong.

    Naming the missing and unexpected keys is what makes it converge.
    """
    raw = getattr(prediction, "extracted", "")
    score = field_f1(example.expected, raw)

    if score == 1.0:
        return dspy.Prediction(score=score, feedback="Every field matched.")

    try:
        want = json.loads(example.expected)
    except json.JSONDecodeError:
        want = {}
    try:
        got = json.loads(extract_json(raw))
        if not isinstance(got, dict):
            got = {}
    except (json.JSONDecodeError, TypeError):
        return dspy.Prediction(
            score=score,
            feedback=(
                "The output was not a JSON object. Return exactly one JSON object and nothing else: "
                f"no prose, no code fences. Expected keys: {sorted(want)}."
            ),
        )

    missing = sorted(set(want) - set(got))
    unexpected = sorted(set(got) - set(want))
    wrong_value = sorted(k for k in set(want) & set(got) if str(want[k]).strip().lower() != str(got[k]).strip().lower())

    parts = [f"Field F1 {score:.2f}."]
    if missing:
        parts.append(f"Missing required keys: {missing}. Use these exact key names for every document type.")
    if unexpected:
        parts.append(f"Do not emit these keys: {unexpected}.")
    if wrong_value:
        parts.append(
            "Wrong values for: "
            + ", ".join(f"{k} (expected {want[k]!r}, got {got[k]!r})" for k in wrong_value)
        )
    return dspy.Prediction(score=score, feedback=" ".join(parts))


def load_rows() -> list[dspy.Example]:
    with GOLDEN.open(newline="", encoding="utf-8") as han
[truncated — 1870 more characters]
```

[1 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]