Prompts as Code & Model Migrations

{"pass_rate": 0.70, "fields": {"severity": 0.80, "type": 0.95, ...}}

"Summarize the issue in 3 bullets.\n"        sha256[:12] = fc22e7a99665
"Summarize the issue in 3 short bullets.\n"  sha256[:12] = 3e5ebc93b1c6

prompts/
├── prompts.lock              # {"extract_issue@v1": "<sha>", ...}: written once per version
└── extract_issue/
    ├── v1.md                 # TOML front matter between +++ lines, then the prompt text
    ├── v2.md
    └── active.toml           # version = "v1": the only file that changes on promotion

+++
model = "gpt-6-luna"
effort = "low"
created = "2026-11-03"
author = "maya"
notes = "Adds a severity rubric (S3 vs S2 confusions)."
+++
You extract facts from GitHub issues filed against Orbit, ...

# script_id: day_021_prompts_as_code/seed_prompts
# seed_prompts.py: writes the demo prompt files. In a real repo you commit these files directly;
# this script exists only because the course pipeline extracts Python, not .md or .toml files.
# Install: nothing (stdlib). Env: ATLAS_PROMPTS_DIR (optional, default ./prompts). No API key.
import os
from pathlib import Path

PROMPTS_DIR = Path(os.environ.get("ATLAS_PROMPTS_DIR", "prompts"))

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

Rules:
- Use null when a fact is not stated. Never guess a version from dates or context.
- severity is for bugs only; use null for features, questions and docs.
- error_codes: every ORB-NNNN code that appears anywhere, including logs and comments. [] if none.
- components: usually one; list two only when the issue really spans both.
- summary: one or two sentences in English, even when the issue is in another language.
"""
SEVERITY = """
Severity (bugs only):
- S1: outage or data loss with no workaround.
- S2: a major feature is broken for many users.
- S3: broken, but a workaround exists (the reporter or a comment names one).
- S4: cosmetic: typos, misleading log lines, wrong colours.
"""
GLOSSARY = "\nComponent glossary:\n$glossary\n"


def _version(created: str, notes: str, body: str) -> str:
    return (f'+++\nmodel = "gpt-6-luna"\neffort = "low"\ncreated = "{created}"\n'
            f'author = "maya"\nnotes = "{notes}"\n+++\n{body}')


FILES = {
    "extract_issue/v1.md": _version("2026-10-20", "Day 16 instructions, moved to a file.", RULES + GLOSSARY),
    "extract_issue/v2.md": _version("2026-11-03", "Adds a severity rubric (S3 vs S2 confusions).",
                                    RULES + SEVERITY + GLOSSARY),
    "extract_issue/active.toml": 'version = "v1"\npromoted = "2026-10-20"\nreason = "first file version"\n',
}


def seed(root: Path = PROMPTS_DIR) -> list[Path]:
    """Create missing files only. Never overwrite: a published version is immutable."""
    written = []
    for rel, text in FILES.items():
        path = root / rel
        if not path.exists():
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(text, encoding="utf-8")
            written.append(path)
    return written


if __name__ == "__main__":
    print(f"wrote {len(seed())} files under {PROMPTS_DIR}/")

wrote 3 files under prompts/

# script_id: day_021_prompts_as_code/minimal_registry
# minimal_registry.py: the smallest file-based loader. Stdlib only, Python 3.11+. No API key.
import tomllib
from pathlib import Path

from seed_prompts import PROMPTS_DIR, seed


def load_prompt_file(path: Path) -> tuple[dict, str]:
    """Split '+++ <TOML> +++ <text>' into (meta, text)."""
    _, front, body = path.read_text(encoding="utf-8").split("+++\n", 2)   # '', TOML, prompt text
    return tomllib.loads(front), body


if __name__ == "__main__":
    seed()                                       # idempotent, so this demo runs in any order
    meta, text = load_prompt_file(PROMPTS_DIR / "extract_issue" / "v2.md")
    print(meta["model"], meta["effort"], "|", text.splitlines()[0][:60])

gpt-6-luna low | You extract facts from GitHub issues filed against Orbit, an

# script_id: day_021_prompts_as_code/registry
# registry.py (atlas/prompts/registry.py): versioned prompt files, content hashes, a lock file.
# Install: nothing (stdlib, Python 3.11+). Env: ATLAS_PROMPTS_DIR (optional, default ./prompts).
import hashlib
import json
import os
import tomllib
from dataclasses import dataclass, field
from datetime import date
from pathlib import Path
from string import Template

PROMPTS_DIR = Path(os.environ.get("ATLAS_PROMPTS_DIR", "prompts"))
LOCK = "prompts.lock"


@dataclass(frozen=True)
class Prompt:
    name: str
    version: str
    text: str
    meta: dict = field(repr=False)     # front matter: model, effort, created, author, notes
    sha: str = ""

    @property
    def ref(self) -> str:
        return f"{self.name}@{self.version}"

    def render(self, **vars: str) -> str:
        """Day 9's rules: a missing variable raises KeyError; '</' inside data is escaped."""
        safe = {key: str(value).replace("</", "&lt;/") for key, value in vars.items()}
        return Template(self.text).substitute(**safe)


def file_sha(path: Path) -> str:
    """Hash the whole file, front matter included: model and effort change behaviour too.
    read_text() turns \\r\\n into \\n, so a Windows checkout hashes the same."""
    return hashlib.sha256(path.read_text(encoding="utf-8").encode("utf-8")).hexdigest()[:12]

# script_id: day_021_prompts_as_code/registry
def list_versions(name: str, root: Path = PROMPTS_DIR) -> list[str]:
    return sorted((p.stem for p in (root / name).glob("v*.md")), key=lambda v: int(v[1:]))


def load_prompt(name: str, version: str = "active", root: Path = PROMPTS_DIR) -> Prompt:
    """Day 9's load_prompt, upgraded: returns a Prompt (text + meta + sha), not a bare Template."""
    if version == "active":
        pointer = (root / name / "active.toml").read_text(encoding="utf-8")
        version = tomllib.loads(pointer)["version"]
    path = root / name / f"{version}.md"
    if not path.exists():
        known = ", ".join(list_versions(name, root)) or "none"
        raise KeyError(f"unknown prompt {name}@{version}; known versions: {known}")
    _, front, body = path.read_text(encoding="utf-8").split("+++\n", 2)
    return Prompt(name, version, body, tomllib.loads(front), file_sha(path))


def render_prompt(name: str, version: str = "active", **vars: str) -> str:
    """Call-compatible with Day 9: render_prompt("extract_issue", glossary=...) still works."""
    return load_prompt(name, version).render(**vars)


def set_active(name: str, version: str, reason: str, root: Path = PROMPTS_DIR) -> None:
    """Promotion moves the pointer. The version files themselves never change."""
    load_prompt(name, version, root)               # a typo fails here, before anything is written
    (root / name / "active.toml").write_text(
        f'version = "{version}"\npromoted = "{date.today()}"\nreason = {json.dumps(reason)}\n',
        encoding="utf-8")

# script_id: day_021_prompts_as_code/registry
def _all_versions(root: Path) -> dict[str, Path]:
    return {f"{p.parent.name}@{p.stem}": p for p in sorted(root.glob("*/v*.md"))}


def write_lock(root: Path = PROMPTS_DIR) -> list[str]:
    """Record the sha of every version not yet locked. Returns the refs it added."""
    path = root / LOCK
    lock = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
    added = {ref: file_sha(p) for ref, p in _all_versions(root).items() if ref not in lock}
    path.write_text(json.dumps(dict(sorted((lock | added).items())), indent=2) + "\n", encoding="utf-8")
    return list(added)


def verify_lock(root: Path = PROMPTS_DIR) -> list[str]:
    """Problems as strings. [] means every version is byte-identical to when it was locked."""
    lock = json.loads((root / LOCK).read_text(encoding="utf-8"))
    files = _all_versions(root)
    problems = [f"{ref} modified" for ref, p in files.items() if ref in lock and file_sha(p) != lock[ref]]
    problems += [f"{ref} not locked" for ref in files if ref not in lock]
    problems += [f"{ref} deleted" for ref in lock if ref not in files]
    return problems

# script_id: day_021_prompts_as_code/registry
if __name__ == "__main__":
    import shutil
    import tempfile

    from seed_prompts import seed

    seed(PROMPTS_DIR)
    write_lock()                       # in a real repo you run this once, when you publish a version
    p = load_prompt("extract_issue")
    print(f"{p.ref} sha={p.sha} · versions: {', '.join(list_versions('extract_issue'))}")
    print("lock problems:", verify_lock())

    scratch = Path(tempfile.mkdtemp()) / "prompts"
    shutil.copytree(PROMPTS_DIR, scratch)
    v1 = scratch / "extract_issue" / "v1.md"
    v1.write_text(v1.read_text(encoding="utf-8").replace("one or two sentences", "one sentence"),
                  encoding="utf-8")
    print("after an in-place edit:", verify_lock(scratch))
    try:
        p.render()                     # forgot glossary=
    except KeyError as err:
        print("KeyError:", err)

extract_issue@v1 sha=0e9a2cf41539 · versions: v1, v2
lock problems: []
after an in-place edit: ['extract_issue@v1 modified']
KeyError: 'glossary'

# script_id: day_021_prompts_as_code/config
# config.py: the ONE place model IDs, effort and prices live. Everything else imports from here.
# Install: nothing (stdlib). Env: ATLAS_MODEL, ATLAS_EFFORT, ATLAS_TIER_MID (all optional).
import os
import re
from dataclasses import dataclass
from pathlib import Path

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")      # cheapest current model (Day 7)
EFFORT = os.environ.get("ATLAS_EFFORT", "low")           # GPT-6 uses "medium" if you send nothing
TIERS = {"cheap": MODEL, "mid": os.environ.get("ATLAS_TIER_MID", "gpt-6.1-terra")}
# USD per 1M tokens, standard tier, as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES_PER_1M = {
    "gpt-6-luna": {"input": 0.10, "cached_input": 0.01, "output": 0.50},
    "gpt-6.1-terra": {"input": 1.00, "cached_input": 0.05, "output": 5.00},
}


@dataclass(frozen=True)
class Settings:
    model: str
    effort: str


def settings_for(meta: dict, *, model: str | None = None, effort: str | None = None) -> Settings:
    """Explicit argument > prompt front matter > env/default."""
    return Settings(model or meta.get("model") or MODEL, effort or meta.get("effort") or EFFORT)


MODEL_LITERAL = re.compile(r"""["'](gpt-\d[\w.\-]*)["']""")
SKIP_DIRS = {".venv", "venv", "site-packages", "node_modules", ".git"}


def find_hardcoded_models(root: Path = Path("."), allow: tuple[str, ...] = ("config.py", "seed_prompts.py")) -> list[str]:
    """'file:line: id' for every model-ID string literal outside the allowed files."""
    hits = []
    for path in sorted(root.rglob("*.py")):
        if path.name in allow or SKIP_DIRS & set(path.parts):
            continue
        for n, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
            hits += [f"{path}:{n}: {m}" for m in MODEL_LITERAL.findall(line)]
    return hits


if __name__ == "__main__":
    print(settings_for({"model": MODEL, "effort": "low"}))
    print(settings_for({"model": MODEL, "effort": "low"}, model=TIERS["mid"]))
    print("hard-coded model IDs:", find_hardcoded_models() or "none")

Settings(model='gpt-6-luna', effort='low')
Settings(model='gpt-6.1-terra', effort='low')
hard-coded model IDs: none

# script_id: day_021_prompts_as_code/evals
# evals.py: compact copy of Day 20's runner + provenance (prompt ref, sha, model, effort per report).
# Built on Day 20 (../../Phase_2_Building_with_LLM_APIs/Day_020_first_eval_set/first_eval_set.md).
# Install: uv add pydantic. Env: ORBIT_DATA (optional). Needs OPENAI_API_KEY only for live runs.
import hashlib
import json
import os
import random
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Callable, Literal, get_args

from pydantic import BaseModel, ConfigDict, Field

from config import PRICES_PER_1M, Settings, settings_for
from registry import Prompt

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
Component = Literal["core", "scheduler", "flows", "broker-redis", "broker-postgres", "broker-sqs",
                    "cli", "dashboard", "retries", "results", "observability", "auth", "docs"]
COMPONENT_HELP = {
    "core": "queues, workers, leases, job execution", "scheduler": "cron schedules and delays",
    "flows": "Flows, steps and step state", "broker-redis": "the Redis broker",
    "broker-postgres": "the Postgres broker", "broker-sqs": "the SQS broker", "cli": "the orbit command",
    "dashboard": "the web UI", "retries": "retry policy and backoff", "results": "the result backend",
    "observability": "metrics and tracing", "auth": "dashboard login and OIDC",
    "docs": "the documentation itself"}
assert set(get_args(Component)) == set(COMPONENT_HELP), "glossary and schema drifted apart"


class IssueRecord(BaseModel):                     # Day 16's schema, descriptions shortened
    model_config = ConfigDict(extra="forbid")
    type: Literal["bug", "feature", "question", "docs"]
    components: list[Component] = Field(description="Every Orbit component the issue is about")
    severity: Literal["S1", "S2", "S3", "S4"] | None = Field(description="Bugs only; null otherwise")
    orbit_version: str | None = Field(description="Full version, e.g. 3.1.2; null when not stated")
    python_version: str | None = Field(description="As written, e.g. 3.12.6; null when not stated")
    error_codes: list[str] = Field(description="Every ORB-NNNN code in the issue or its comments")
    summary: str = Field(description="One or two English sentences")


def glossary() -> str:
    return "\n".join(f"{name} = {desc}" for name, desc in COMPONENT_HELP.items())

# script_id: day_021_prompts_as_code/evals
FIELDS = ("type", "components", "severity", "orbit_version", "python_version", "error_codes")
SETS = {"components", "error_codes"}               # order doesn't matter for these two


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


def issue_to_input(issue: dict) -> str:
    """Title, body and comments. Never the labels: they are the answer key."""
    parts = [f"# {issue['title']}", issue["body"]]
    parts += [f"--- comment by {c['author']} ---\n{c['body']}" for c in issue["comments"]]
    return "\n\n".join(parts)


def load_cases(split: str = "dev") -> list[tuple[dict, IssueRecord]]:
    issues = {i["id"]: i for i in _jsonl(DATA_DIR / "issues" / "issues.jsonl")}
    gold = sorted(_jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl"), key=lambda g: g["issue_id"])
    rows = {"dev": gold[:20], "heldout": gold[20:], "all": gold}[split]
    return [(issues[g["issue_id"]], IssueRecord.model_validate({k: v for k, v in g.items() if k != "issue_id"}))
            for g in rows]

# script_id: day_021_prompts_as_code/evals
@dataclass
class CaseResult:
    issue_id: int
    fields: dict[str, bool]
    latency_s: float
    cost_usd: float
    error: str | None = None

    @property
    def passed(self) -> bool:
        return self.error is None and all(self.fields.values())


@dataclass
class EvalReport:
    meta: dict                                     # prompt, sha, model, effort, split, simulated, n
    cases: list[CaseResult] = field(default_factory=list)

    def field_accuracy(self) -> dict[str, float]:
        return {f: sum(c.fields[f] for c in self.cases) / len(self.cases) for f in FIELDS}

    @property
    def pass_rate(self) -> float:
        return sum(c.passed for c in self.cases) / len(self.cases)

    def cost_per_1k(self) -> float:
        return 1000 * sum(c.cost_usd for c in self.cases) / len(self.cases)

    def latency(self, q: float) -> float:
        """Nearest-rank percentile. With 20 cases, p95 is the slowest call: treat it as rough."""
        xs = sorted(c.latency_s for c in self.cases)
        return xs[min(len(xs) - 1, int(q * len(xs)))]

    def save(self, path: Path) -> Path:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps({"meta": self.meta, "cases": [asdict(c) for c in self.cases]}, indent=1))
        return path

    @classmethod
    def load(cls, path: Path) -> "EvalReport":
        raw = json.loads(Path(path).read_text())
        return cls(raw["meta"], [CaseResult(**c) for c in raw["cases"]])

# script_id: day_021_prompts_as_code/evals
@dataclass
class Outcome:
    record: IssueRecord | None                     # None = refusal or nothing parseable
    cost_usd: float
    latency_s: float


Extractor = Callable[[dict], Outcome]


def score(pred: IssueRecord, gold: IssueRecord) -> dict[str, bool]:
    p, g = pred.model_dump(), gold.model_dump()
    return {f: set(p[f]) == set(g[f]) if f in SETS else p[f] == g[f] for f in FIELDS}


def run_eval(cases: list[tuple[dict, IssueRecord]], extract: Extractor, meta: dict) -> EvalReport:
    report = EvalReport({**meta, "n": len(cases), "created": time.strftime("%Y-%m-%dT%H:%M:%S")})
    for issue, gold in cases:
        try:
            out = extract(issue)
            ok = score(out.record, gold) if out.record else dict.fromkeys(FIELDS, False)
            error = None if out.record else "no parsed output"
            report.cases.append(CaseResult(issue["id"], ok, out.latency_s, out.cost_usd, error))
        except Exception as err:                   # one bad case counts as a failure, not a crash
            report.cases.append(CaseResult(issue["id"], dict.fromkeys(FIELDS, False), 0.0, 0.0, repr(err)))
    return report

# script_id: day_021_prompts_as_code/evals
WRONG = {"type": ["bug", "question"], "severity": ["S2", "S3"], "components": [["core"], ["docs"]],
         "error_codes": [[], ["ORB-9999"]], "orbit_version": [None, "0.0.0"], "python_version": [None, "3.12"]}
SIM_LATENCY_S = {"none": 0.6, "low": 1.1, "medium": 2.2, "high": 4.0, "xhigh": 7.0, "max": 10.0}
SIM_REASONING_TOKENS = {"none": 0, "low": 150, "medium": 600, "high": 1500, "xhigh": 3000, "max": 5000}


def simulated_extractor(prompt: Prompt, s: Settings, gold: dict[int, IssueRecord]) -> Extractor:
    key = f"{prompt.sha}|{s.model}|{s.effort}"
    error_rate = 0.03 + int(hashlib.sha256(key.encode()).hexdigest(), 16) % 9 / 100   # 3-11% per field
    price = PRICES_PER_1M[s.model]                 # an unpriced model fails here, before any run

    def extract(issue: dict) -> Outcome:
        rng = random.Random(f"{key}|{issue['id']}")   # string seeds give the same draws every run
        rec = gold[issue["id"]].model_dump()
        for f in FIELDS:
            if rng.random() < error_rate:
                rec[f] = next(v for v in WRONG[f] if v != rec[f])
        tokens_in = len(issue_to_input(issue)) // 4 + 700          # + instructions and schema
        tokens_out = 120 + SIM_REASONING_TOKENS[s.effort]
        cost = (tokens_in * price["input"] + tokens_out * price["output"]) / 1_000_000
        return Outcome(IssueRecord.model_validate(rec), cost, SIM_LATENCY_S[s.effort] * rng.uniform(0.7, 1.8))
    return extract


def evaluate(prompt: Prompt, split: str = "dev", *, model: str | None = None, effort: str | None = None,
             simulate: bool = False) -> EvalReport:
    """Run one prompt version + settings over a split. The report records exactly what ran."""
    s = settings_for(prompt.meta, model=model, effort=effort)
    cases = load_cases(split)
    if simulate:
        extract = simulated_extractor(prompt, s, {issue["id"]: gold for issue, gold in cases})
    else:
        from extract import make_extractor         # lazy import: the offline path never needs a key
        extract = make_extractor(prompt, model=s.model, effort=s.effort)
    meta = {"prompt": prompt.ref, "sha": prompt.sha, "model": s.model, "effort": s.effort,
            "split": split, "simulated": simulate}
    return run_eval(cases, extract, meta)


if __name__ == "__main__":
    from registry import load_prompt
    from seed_prompts import seed

    seed()
    report = evaluate(load_prompt("extract_issue"), "dev", simulate=not os.environ.get("OPENAI_API_KEY"))
    print(json.dumps({k: report.meta[k] for k in ("prompt", "sha", "model", "effort", "simulated")}))
    print(f"pass rate {report.pass_rate:.0%} · $/1k {report.cost_per_1k():.2f} · p95 {report.latency(0.95):.1f}s")

# your numbers will differ; simulated runs are seeded noise
{"prompt": "extract_issue@v1", "sha": "0e9a2cf41539", "model": "gpt-6-luna", "effort": "low", "simulated": true}
pass rate 70% · $/1k 0.25 · p95 1.9s

# script_id: day_021_prompts_as_code/extract
# extract.py: the live extractor. One prompt version + settings -> a function run_eval can call.
# Install: uv add "openai>=3.46,<4" pydantic. Env: OPENAI_API_KEY (required), ATLAS_LOG_DIR (optional)
import json
import os
import time
from pathlib import Path

from openai import OpenAI

from config import PRICES_PER_1M, settings_for
from evals import Extractor, IssueRecord, Outcome, glossary, issue_to_input
from registry import Prompt

LOG_DIR = Path(os.environ.get("ATLAS_LOG_DIR", "logs"))
client = OpenAI(timeout=60, max_retries=3)


def cost_from_usage(usage, model: str) -> float:   # Day 12; reasoning tokens are inside output_tokens;
    # simplified: ignores the cache-write surcharge (Day 18)
    p, cached = PRICES_PER_1M[model], usage.input_tokens_details.cached_tokens
    return ((usage.input_tokens - cached) * p["input"] + cached * p["cached_input"]
            + usage.output_tokens * p["output"]) / 1_000_000


def log_call(response, *, label: str, latency_s: float, cost_usd: float, extra: dict | None = None) -> None:
    """Day 12's call log: one JSON line per call, no prompt text, no keys."""
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    row = {"ts": time.time(), "label": label, "model": response.model, "id": response.id,
           "status": response.status, "input_tokens": response.usage.input_tokens,
           "output_tokens": response.usage.output_tokens, "latency_s": round(latency_s, 3),
           "cost_usd": round(cost_usd, 6), **(extra or {})}
    with open(LOG_DIR / "calls.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(row) + "\n")


def shutdown_date(model: str) -> str | None:       # Day 7: 'YYYY-MM-DD' once a shutdown is scheduled
    return client.models.retrieve(model).shutdown_date

# script_id: day_021_prompts_as_code/extract
def make_extractor(prompt: Prompt, *, model: str | None = None, effort: str | None = None) -> Extractor:
    s = settings_for(prompt.meta, model=model, effort=effort)
    instructions = prompt.render(glossary=glossary())
    provenance = {"prompt": prompt.ref, "sha": prompt.sha, "effort": s.effort}

    def extract(issue: dict) -> Outcome:
        t0 = time.perf_counter()
        resp = client.responses.parse(model=s.model, instructions=instructions, input=issue_to_input(issue),
                                      text_format=IssueRecord, reasoning={"effort": s.effort})
        latency = time.perf_counter() - t0
        cost = cost_from_usage(resp.usage, s.model)
        log_call(resp, label="extract_issue", latency_s=latency, cost_usd=cost,
                 extra={"issue_id": issue["id"], **provenance})
        return Outcome(resp.output_parsed, cost, latency)     # None on refusal: scored as a failure
    return extract


if __name__ == "__main__":
    from evals import load_cases
    from registry import load_prompt

    issue, gold = load_cases("dev")[0]
    out = make_extractor(load_prompt("extract_issue"))(issue)
    print(out.record, f"${out.cost_usd:.5f}", f"{out.latency_s:.1f}s")

# script_id: day_021_prompts_as_code/migrate_model
# migrate_model.py (scripts/migrate_model.py): re-run the eval set with a candidate and diff the reports.
# Install: uv add "openai>=3.46,<4" pydantic. Env: OPENAI_API_KEY for live runs; without it, simulated.
import argparse
import json
import os
from dataclasses import asdict, dataclass
from pathlib import Path

from config import TIERS, settings_for
from evals import FIELDS, EvalReport, evaluate
from registry import Prompt, load_prompt


@dataclass
class MigrationDiff:
    a: dict                                        # meta of the baseline report
    b: dict                                        # meta of the candidate report
    fields: dict[str, tuple[float, float]]         # field -> (accuracy on A, accuracy on B)
    pass_rate: tuple[float, float]
    flips: list[int]                               # passed on A, fail on B
    fixes: list[int]                               # failed on A, pass on B
    cost_per_1k: tuple[float, float]
    p50_s: tuple[float, float]
    p95_s: tuple[float, float]


def compare_reports(a: EvalReport, b: EvalReport) -> MigrationDiff:
    pa, pb = ({c.issue_id: c.passed for c in r.cases} for r in (a, b))
    if pa.keys() != pb.keys():
        raise ValueError("reports cover different cases; compare like with like")
    fa, fb = a.field_accuracy(), b.field_accuracy()
    return MigrationDiff(
        a.meta, b.meta, {f: (fa[f], fb[f]) for f in FIELDS}, (a.pass_rate, b.pass_rate),
        sorted(i for i in pa if pa[i] and not pb[i]), sorted(i for i in pa if pb[i] and not pa[i]),
        (a.cost_per_1k(), b.cost_per_1k()), (a.latency(0.5), b.latency(0.5)), (a.latency(0.95), b.latency(0.95)))

# script_id: day_021_prompts_as_code/migrate_model
def decide(diff: MigrationDiff, *, max_field_drop: float = 0.05, max_cost_ratio: float | None = None,
           max_p95_s: float | None = None) -> str:
    """'ship', or 'stay: <reasons>'."""
    reasons = [f"{f} -{round((x - y) * 100)} pts" for f, (x, y) in diff.fields.items()
               if round(x - y, 6) > max_field_drop]     # round(): 1.0 - 0.95 is 0.050000000000000044
    if diff.pass_rate[1] < diff.pass_rate[0]:
        reasons.append(f"pass rate {diff.pass_rate[0]:.0%} -> {diff.pass_rate[1]:.0%}")
    if max_cost_ratio and diff.cost_per_1k[1] > max_cost_ratio * diff.cost_per_1k[0]:
        reasons.append(f"cost x{diff.cost_per_1k[1] / diff.cost_per_1k[0]:.1f}")
    if max_p95_s and diff.p95_s[1] > max_p95_s:
        reasons.append(f"p95 {diff.p95_s[1]:.1f}s")
    return "ship" if not reasons else "stay: " + "; ".join(reasons)


def label(meta: dict) -> str:
    return f"{meta['prompt']} on {meta['model']}/{meta['effort']}"


def format_diff(d: MigrationDiff) -> str:
    rows = [f"{'field':<16}{'A':>5}{'B':>6}{'Δ':>6}"]
    rows += [f"{f:<16}{x:>5.0%}{y:>6.0%}{(y - x) * 100:>+6.0f}" for f, (x, y) in d.fields.items()]
    (pa, pb), (ca, cb) = d.pass_rate, d.cost_per_1k
    rows += [f"{'pass rate':<16}{pa:>5.0%}{pb:>6.0%}{(pb - pa) * 100:>+6.0f}",
             f"flips: {len(d.flips)} {d.flips} · fixes: {len(d.fixes)} {d.fixes}",
             f"$/1k {ca:.2f} -> {cb:.2f} · p50 {d.p50_s[0]:.1f}s -> {d.p50_s[1]:.1f}s"
             f" · p95 {d.p95_s[0]:.1f}s -> {d.p95_s[1]:.1f}s"]
    return f"A = {label(d.a)}\nB = {label(d.b)}\n" + "\n".join(rows)


def drill(base: Prompt, cand: Prompt, *, model: str | None = None, effort: str | None = None,
          splits: tuple[str, ...] = ("dev", "heldout"), simulate: bool = False,
          out_dir: Path = Path("reports"), **rule) -> tuple[str, list[MigrationDiff]]:
    diffs: list[MigrationDiff] = []
    for split in splits:
        a = evaluate(base, split, simulate=simulate)
        b = evaluate(cand, split, model=model, effort=effort, simulate=simulate)
        for r in (a, b):
            r.save(out_dir / f"{split}_{r.meta['prompt']}_{r.meta['model']}_{r.meta['effort']}.json")
        diffs.append(compare_reports(a, b))
        verdict = decide(diffs[-1], **rule)
        print(f"--- {split} ({len(a.cases)} cases)\n{format_diff(diffs[-1])}\n{split}: {verdict}")
        if verdict != "ship":
            return verdict, diffs
    return "ship", diffs

# script_id: day_021_prompts_as_code/prompt_gate
# prompt_gate.py: gate a new prompt version against the active one; promote only with --promote.
# Install: uv add "openai>=3.46,<4" pydantic. Env: OPENAI_API_KEY for live runs (else simulated).
import argparse
import os

from config import settings_for
from migrate_model import drill
from registry import load_prompt, set_active


def gate_prompt(name: str, candidate: str, *, simulate: bool, promote: bool = False) -> str:
    base, cand = load_prompt(name), load_prompt(name, candidate)
    if base.version == cand.version:
        return f"{cand.ref} is already active"
    if settings_for(base.meta) != settings_for(cand.meta):
        print(f"note: {cand.ref} also changes model/effort, so you can't tell which change caused what")
    verdict, _ = drill(base, cand, simulate=simulate)    # each version runs with its own model/effort
    if verdict == "ship" and promote:
        set_active(name, candidate, reason=f"gate passed vs {base.ref}")
    return verdict


if __name__ == "__main__":
    from seed_prompts import seed

    seed()
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("candidate", nargs="?", default="v2")
    ap.add_argument("--name", default="extract_issue")
    ap.add_argument("--promote", action="store_true", help="move active.toml if the gate passes")
    args = ap.parse_args()
    verdict = gate_prompt(args.name, args.candidate, simulate=not os.environ.get("OPENAI_API_KEY"),
                          promote=args.promote)
    print(f"promote {args.candidate}: {verdict}")

# your numbers will differ; simulated runs are seeded noise
--- dev (20 cases)
A = extract_issue@v1 on gpt-6-luna/low
B = extract_issue@v2 on gpt-6-luna/low
field               A     B     Δ
type              95%   95%    +0
components        90%   95%    +5
severity         100%   95%    -5
orbit_version     95%   95%    +0
python_version    85%   95%   +10
error_codes       85%   90%    +5
pass rate         70%   70%    +0
flips: 4 [10, 83, 97, 104] · fixes: 4 [109, 173, 177, 180]
$/1k 0.25 -> 0.25 · p50 1.3s -> 1.5s · p95 1.9s -> 2.0s
dev: ship
--- heldout (10 cases)
A = extract_issue@v1 on gpt-6-luna/low
B = extract_issue@v2 on gpt-6-luna/low
field               A     B     Δ
type             100%   80%   -20
components        90%  100%   +10
severity         100%  100%    +0
orbit_version    100%  100%    +0
python_version    90%  100%   +10
error_codes       80%   80%    +0
pass rate         60%   70%   +10
flips: 1 [209] · fixes: 2 [188, 219]
$/1k 0.48 -> 0.48 · p50 1.5s -> 1.4s · p95 2.0s -> 1.9s
heldout: stay: type -20 pts
promote v2: stay: type -20 pts

# script_id: day_021_prompts_as_code/migrate_model
def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Model-migration drill: same prompt, candidate model/effort.")
    ap.add_argument("--prompt", default="extract_issue")
    ap.add_argument("--to", dest="model", help=f"candidate model (default {TIERS['mid']} unless --effort)")
    ap.add_argument("--effort", help="candidate effort (default: the prompt's own)")
    ap.add_argument("--max-field-drop", type=float, default=0.05)
    ap.add_argument("--max-cost-ratio", type=float, help="e.g. 3 = stay if B costs more than 3x A")
    ap.add_argument("--simulate", action="store_true", help="offline; automatic without OPENAI_API_KEY")
    args = ap.parse_args(argv)
    if args.model is None and args.effort is None:
        args.model = TIERS["mid"]
    simulate = args.simulate or not os.environ.get("OPENAI_API_KEY")
    prompt = load_prompt(args.prompt)
    if simulate:
        print("SIMULATED: no API calls; the numbers exercise the mechanics, not the models")
    else:
        from extract import shutdown_date
        for m in {settings_for(prompt.meta).model, args.model} - {None}:
            if when := shutdown_date(m):
                print(f"WARNING: {m} shuts down on {when}")
    verdict, diffs = drill(prompt, prompt, model=args.model, effort=args.effort, simulate=simulate,
                           max_field_drop=args.max_field_drop, max_cost_ratio=args.max_cost_ratio)
    Path("reports").mkdir(exist_ok=True)
    Path("reports/last_diff.json").write_text(json.dumps([asdict(d) for d in diffs], indent=2))
    print("decision:", verdict)
    return 0


if __name__ == "__main__":
    from seed_prompts import seed

    seed()
    raise SystemExit(main())

python migrate_model.py                        # active prompt: its own model vs the mid tier
python migrate_model.py --to gpt-6.1-terra --max-cost-ratio 3
python migrate_model.py --effort none          # same model, effort only (Exercise 3)

# your numbers will differ; simulated runs are seeded noise
SIMULATED: no API calls; the numbers exercise the mechanics, not the models
--- dev (20 cases)
A = extract_issue@v1 on gpt-6-luna/low
B = extract_issue@v1 on gpt-6.1-terra/low
field               A     B     Δ
type              95%   85%   -10
components        90%   95%    +5
severity         100%   95%    -5
orbit_version     95%   95%    +0
python_version    85%   85%    +0
error_codes       85%   90%    +5
pass rate         70%   60%   -10
flips: 5 [25, 63, 74, 97, 104] · fixes: 3 [173, 177, 180]
$/1k 0.25 -> 2.49 · p50 1.3s -> 1.6s · p95 1.9s -> 2.0s
dev: stay: type -10 pts; pass rate 70% -> 60%
decision: stay: type -10 pts; pass rate 70% -> 60%

## 2026-11-07 · extract_issue@v1 on gpt-6-luna/low -> gpt-6.1-terra/low
- Decision: stay (type -10 pts; pass rate 70% -> 60%; $/1k 0.25 -> 2.49) · simulated run
- Numbers: reports/dev_*.json · flips: [25, 63, 74, 97, 104]
- Rollback: nothing shipped; if it had, set_active back / revert the front-matter pin
- Next drill: when gpt-6-luna gets a shutdown_date, or a new cheap tier ships