Prompt Regression Testing in CI: A Golden Set That Blocks Bad Prompt Changes

"""The prompt test suite: checks every golden case, summarises a run and compares it with the baseline."""
import json
import re
from concurrent.futures import ThreadPoolExecutor

import harness

GOLDEN = "golden.jsonl"
SAMPLES = 3          # draws per case: production runs at temperature 0.7, so one draw proves little
TOLERANCE = 0.05     # how far the overall pass rate may fall before the gate fails


def load_golden(path=GOLDEN):
    return [json.loads(line) for line in open(path, encoding="utf-8") if line.strip()]


def parse_json(text):
    """The JSON object in the reply (a ```json fence around it is fine), or None."""
    m = re.search(r"\{.*\}", text or "", re.S)
    if not m:
        return None
    try:
        obj = json.loads(m.group(0))
    except json.JSONDecodeError:
        return None
    return obj if isinstance(obj, dict) else None


# ---------- Step 1: deterministic checks ----------
def check_case(case, output):
    """Run every assert of the case on the model's output; return the failure messages (empty = passed).
    Types: json; equals (ignores case and surrounding spaces); contains, contains_any, not_contains
    (case-insensitive; "{ticket}" means the case's ticket); max_words; judge (asks judge() the assert's
    question about the field; fails unless the answer equals "expect"; skipped while judge() returns None)."""
    # TODO (Step 1): parse_json(output) (None -> ["output is not a JSON object"]); then for each assert other than
    # "json", read the field (not a string -> "<field> is missing") and add a message for every assert that fails.
    # For "judge", call judge(a["question"], value) and fail only if it returns something other than None and a["expect"].
    raise NotImplementedError("Step 1: write check_case()")


# ---------- Step 2: a model as the judge ----------
JUDGE_PROMPT = """You check replies written by a customer support assistant.

Reply to check:
<<<
{text}
>>>

Question: {question}
Answer only if the reply itself does it. Warnings such as "never share your password" and statements that a
team will review a request do not count. Answer with one word: YES or NO."""


def judge(question, text):
    """"yes" or "no": the judge model's answer to a yes/no question about text; "unclear" if the answer is
    neither."""
    return None


def run_suite(template, cases, samples=SAMPLES):
    """Every case `samples` times. Returns {case id: {"passes": n, "samples": samples, "failures": [...]}}."""
    jobs = [(c, s) for c in cases for s in range(samples)]
    with ThreadPoolExecutor(8) as pool:
        outputs = list(pool.map(lambda j: harness.generate(template, j[0], j[1]), jobs))
        checked = list(pool.map(lambda j: check_case(j[0][0], j[1]), zip(jobs, outputs)))
    results = {c["id"]: {"passes": 0, "samples": samples, "failures": []} for c in cases}
    for (c, _), fails in zip(jobs, checked):
        if fails:
            results[c["id"]]["failures"] += fails
        else:
            results[c["id"]]["passes"] += 1
    return results


# ---------- Step 3: summarise a run ----------
def summarize(results, cases):
    """{"status": {case id: "pass" (every sample passed) | "flaky" (some did) | "fail" (none did)},
    "pass_rate": share of cases with status pass, "critical_pass_rate": the same over cases tagged critical
    (1.0 if there are none)}, rates rounded to 3 places."""
    raise NotImplementedError("summarize() arrives in Step 3")


# ---------- Step 4: the gate ----------
def compare(baseline, current, cases, tolerance=TOLERANCE):
    """Why the current run must not ship (an empty list means it may). Blocking reasons:
    - a critical case whose baseline status was "pass" is not "pass" now (name the case id);
    - the pass rate fell by more than `tolerance` below the baseline's (give both rates)."""
    raise NotImplementedError("compare() arrives in Step 4")