diff --git a/examples/model_drift/README.md b/examples/model_drift/README.md new file mode 100644 index 0000000..1959c01 --- /dev/null +++ b/examples/model_drift/README.md @@ -0,0 +1,192 @@ +# Model-drift check + +Did a replacement model **drift** from the one it retires? + +You are about to swap a soon-to-be-discontinued model for a newer one. Before you flip the +switch you want a crisp, reproducible answer to one question: on the prompts you actually +care about, does the replacement answer *equivalently* to the outgoing model — or does it +quietly drift? + +This example wires that check on top of DProvenanceKit. It records the **old** model's +answers as a *golden* run and the **new** model's answers as a *candidate* run (one +`CRITICAL` `generation` event per prompt, carrying `{prompt_id, model, response}`), then +gates the candidate against the golden with a `RegressionGate`. A per-prompt answer that +falls below *your* equivalence threshold surfaces as a critical regression — i.e. drift — +and the check fails with exit code `1` (parity exits `0`), so it drops straight into CI. + +## The honest framing + +DProvenanceKit is the **record + gate substrate**. It does not call a model, own your +prompts, or decide what "the same answer" means. **You** own the three things that make the +check meaningful: + +- **The prompt set** — what "the answers you care about" actually are. The check is only as + good as the prompts you pick. +- **The API key / provider access** — the kit never calls a model for you. You bring the + `ModelClient`. +- **The drift threshold and the equivalence notion** — lexical similarity? an LLM judge? + your own rule? How similar is "still the same answer" for *your* use case. + +## The timing catch — capture the golden before the old model is gone + +The golden baseline is the **old model's** answers, so you must capture them **while the old +model is still callable.** Once it is discontinued you can no longer produce the baseline. + +By default a single run records *both* models, so it needs both live at once — fine for a +pre-cutover go/no-go. To keep gating **after** the old model is retired, capture the golden +once and reuse it: + +```bash +# While the outgoing model is still callable — record ONLY its answers as the golden. +python -m examples.model_drift --old gpt-5.4-2026-03-05 --record-golden \ + --golden-db golden.sqlite --prompts my_prompts.jsonl --live + +# Any time later (even after gpt-5.4 is gone) — gate a candidate against the SAVED golden. +# The old model is never called; only the candidate runs. +python -m examples.model_drift --old gpt-5.4-2026-03-05 --new gpt-5.4-pro-2026-03-05 \ + --golden-db golden.sqlite --prompts my_prompts.jsonl --live +``` + +`record_baseline(...)` and `run_drift_check(..., golden_db=...)` are the Python equivalents. + +## Run it (zero setup, offline) + +The package ships a deterministic `FakeModelClient` — answers are a pure function of the +prompt (no network, no key, no randomness), so the whole flow runs anywhere. It can +simulate **parity** (both models answer identically) or **drift** (the newer model diverges +on a subset of prompts). + +```bash +# Parity — the replacement matches the outgoing model. Exits 0. +python -m examples.model_drift --old old-model --new new-model --fake parity + +# Drift — the replacement diverges on some prompts. Prints which ones, exits 1. +python -m examples.model_drift --old old-model --new new-model --fake drift +``` + +Example drift output: + +``` +Model drift check: OLD=old-model NEW=new-model (judge=lexical, threshold=0.80) + prompt score verdict + greeting 0.66 DRIFT + refund 1.00 ok + ... +Result: DRIFT — 4 of 8 prompts drifted: greeting, password, summary, email +``` + +## Run it live against OpenAI + +Say you are retiring `gpt-5.4` (GPT-5.4 Thinking) for `gpt-5.4-pro` (GPT-5.4 Pro). + +**Pin the dated snapshots for the go/no-go.** A drift baseline must be reproducible: baseline +against the floating `gpt-5.4` alias and it can move under you, so you can no longer tell "the +candidate drifted" from "my baseline drifted." The dated snapshot also freezes the exact +behavior you are about to lose. + +```bash +pip install "dprovenancekit[openai]" +export OPENAI_API_KEY=sk-... + +python -m examples.model_drift \ + --old gpt-5.4-2026-03-05 --new gpt-5.4-pro-2026-03-05 \ + --prompts my_prompts.jsonl \ + --live --judge llm --judge-model gpt-5.4-pro-2026-03-05 \ + --threshold 0.85 +``` + +**Prefer `--judge llm` over the lexical default for a base→pro swap.** A "pro" model tends to +reword and elaborate far more than it changes meaning, so lexical similarity flags a pile of +false drift on wording alone; the LLM judge scores *semantic* equivalence instead. Pick a +neutral, pinned grader with `--judge-model` (ideally not one of the two under test, to avoid a +model favouring its own style). The judge fails **closed**: any reply that is not a clean +`[0, 1]` score — an error string, an out-of-range number — reads as drift, so a flaky judge +can never silently pass a regression. It costs one extra call per prompt. + +**Keep watching the floating alias after cutover (drift surveillance).** Once you depend on the +moving `gpt-5.4-pro` alias, OpenAI can update it under you. Gate the saved dated golden against +the *floating* candidate on a schedule to catch that: + +```bash +python -m examples.model_drift \ + --old gpt-5.4-2026-03-05 --new gpt-5.4-pro \ + --golden-db golden.sqlite --prompts my_prompts.jsonl \ + --live --judge llm --judge-model gpt-5.4-pro-2026-03-05 --threshold 0.85 +``` + +`--live` uses `OpenAIModelClient`, which imports `openai` **lazily** (only when constructed), +so the module and the whole `--fake` path import fine with `openai` absent. Temperature is 0 +(the default) so a re-run of the golden reproduces it. + +## Bring your own prompt set + +Prompts are a JSONL file — one `{"id": ..., "prompt": ...}` per line +(`prompts.sample.jsonl` is a starter set): + +```jsonl +{"id": "refund_policy", "prompt": "What is our refund window for digital goods?"} +{"id": "escalation", "prompt": "When should a support agent escalate to a human?"} +``` + +```bash +python -m examples.model_drift --old old-model --new new-model \ + --prompts my_prompts.jsonl --fake drift +``` + +Keep the `id`s stable across runs — they key each prompt 1:1 between the golden and +candidate, so the gate compares like with like. + +## Bring your own equivalence notion + +A `DriftEvaluator` scores two answers in `[0, 1]` (`1.0` = equivalent). Two ship here: + +- `LexicalSimilarityEvaluator` — stdlib-only surface similarity via + `difflib.SequenceMatcher`. Deterministic, dependency-free, the default. Judges *wording*, + not *meaning*. +- `LLMJudgeEvaluator(model_client, judge_model)` — asks a model to rate equivalence, so a + reworded-but-equivalent answer can still count as equivalent. + +Write your own by implementing `score(old_response, new_response) -> float` and an +`identifier`, then hand it to `run_drift_check(..., evaluator=your_evaluator)`. The helper +`as_equivalence_evaluator` adapts any `DriftEvaluator` into the kit's +`AnyEquivalenceEvaluator`, keyed on the `generation` events so the gate compares the two +runs' response payloads with your rule. + +## From Python + +```python +from examples.model_drift.harness import Prompt, run_drift_check +from examples.model_drift.providers import FakeModelClient +from examples.model_drift.evaluators import LexicalSimilarityEvaluator + +result = run_drift_check( + old_model="old-model", + new_model="new-model", + prompts=[Prompt("capital", "What is the capital of France?")], + client=FakeModelClient(mode="drift"), + evaluator=LexicalSimilarityEvaluator(), + threshold=0.8, + db_path="drift.sqlite", # durable path keeps the golden baseline +) +print(result.passed, result.drifted_prompts) +``` + +## How the threshold maps onto the gate + +`RegressionGate`'s `semantic_threshold` is compared to the *combined* weighted alignment +score, not the raw evaluator output. Under strict-audit weights (type `0.5` / payload `0.5`) +with a matching `type_identifier`, `combined = 0.5 + 0.5 * response_similarity`. The harness +therefore builds a profile with `semantic_threshold = 0.5 + 0.5 * threshold`, so your +`--threshold` *is* the effective response-similarity cutoff: an answer scoring below it lands +below the profile threshold, its `CRITICAL` step is flagged changed-beyond-equivalence, and +the gate fails. See `harness.py::_drift_profile` for the derivation. + +## Files + +| File | What it is | +| --- | --- | +| `providers.py` | `ModelClient` protocol; deterministic `FakeModelClient`; lazy-import `OpenAIModelClient`. | +| `evaluators.py` | `DriftEvaluator` protocol; `LexicalSimilarityEvaluator`; `LLMJudgeEvaluator`; `as_equivalence_evaluator` adapter. | +| `harness.py` | `run_drift_check(...)` — records both runs and gates the candidate. Returns a `DriftCheckResult`. | +| `__main__.py` | The CLI (`python -m examples.model_drift`). | +| `prompts.sample.jsonl` | A starter prompt set. | diff --git a/examples/model_drift/__init__.py b/examples/model_drift/__init__.py new file mode 100644 index 0000000..1167c2d --- /dev/null +++ b/examples/model_drift/__init__.py @@ -0,0 +1,30 @@ +"""Model-drift check — did a replacement model DRIFT from the one it retires? + +You are about to swap a soon-to-be-discontinued model for a newer one. Before you flip +the switch, you want to know: on the prompts you actually care about, does the replacement +answer *equivalently* to the outgoing model — or does it quietly drift? + +What DProvenanceKit gives you, and what it does NOT: + + • The kit is the RECORD + GATE substrate. It records the OLD model's answers as a + GOLDEN run and the NEW model's answers as a CANDIDATE run (one CRITICAL + ``generation`` event per prompt, carrying ``{prompt_id, model, response}``), then + gates the candidate against the golden with a :class:`RegressionGate`. A per-prompt + answer that falls below your equivalence threshold surfaces as a critical + regression — i.e. drift — and the check fails. + + • YOU own the three things that make the check meaningful: + - the PROMPT SET (what "the answers you care about" actually are), + - the API KEY / provider access (the kit never calls a model for you), + - the DRIFT THRESHOLD and the equivalence notion (lexical? an LLM judge? your own + rule) — how similar is "still the same answer" for your use case. + + • The TIMING CATCH: the golden baseline is the OLD model's answers, so you must + capture them WHILE THE OLD MODEL IS STILL CALLABLE. Once it is discontinued you can + no longer produce the baseline — record and save it before the cutover, then gate + every candidate against that saved baseline. + +This package is deliberately provider-agnostic. It ships a deterministic ``FakeModelClient`` +so the whole flow runs with zero setup (no network, no key), and an ``OpenAIModelClient`` +that imports ``openai`` lazily (``pip install "dprovenancekit[openai]"``) for live checks. +""" diff --git a/examples/model_drift/__main__.py b/examples/model_drift/__main__.py new file mode 100644 index 0000000..234f7ee --- /dev/null +++ b/examples/model_drift/__main__.py @@ -0,0 +1,219 @@ +"""CLI for the model-drift check. + + python -m examples.model_drift --old MODEL --new MODEL \\ + [--prompts FILE.jsonl] [--fake parity|drift | --live] \\ + [--judge lexical|llm] [--judge-model MODEL] [--threshold 0.8] [--db PATH] + +Defaults to ``--fake parity`` so it runs with zero setup (no network, no API key). Prints a +per-prompt drift report and exits 1 when the new model drifts, 0 on parity — so it drops +straight into a CI gate. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from typing import List, Optional, Sequence + +from .evaluators import ( + DriftEvaluator, + LexicalSimilarityEvaluator, + LLMJudgeEvaluator, +) +from .harness import ( + DEFAULT_THRESHOLD, + DriftCheckResult, + Prompt, + record_baseline, + run_drift_check, +) +from .providers import ( + FakeModelClient, + ModelClient, + OpenAIModelClient, +) + +# A small built-in prompt set so the demo runs without a --prompts file. +_SAMPLE_PROMPTS: List[Prompt] = [ + Prompt("greeting", "How should I greet a new customer in a support chat?"), + Prompt("refund", "What is a reasonable refund window for a digital product?"), + Prompt("units", "How many meters are in a kilometer?"), + Prompt("capital", "What is the capital of France?"), + Prompt("password", "What makes a password strong?"), + Prompt("summary", "Summarize the benefits of unit testing in one sentence."), + Prompt("email", "Write a one-line out-of-office reply."), + Prompt("advice", "Give one tip for staying focused while working from home."), +] + + +def _load_prompts(path: str) -> List[Prompt]: + """Load prompts from a JSONL file (one ``{"id": ..., "prompt": ...}`` per line).""" + prompts: List[Prompt] = [] + with open(path, "r", encoding="utf-8") as handle: + for lineno, line in enumerate(handle, start=1): + line = line.strip() + if not line: + continue + record = json.loads(line) + try: + prompts.append(Prompt(id=str(record["id"]), text=str(record["prompt"]))) + except KeyError as exc: # pragma: no cover - user input validation + raise ValueError( + f"{path}:{lineno}: each line needs 'id' and 'prompt' keys" + ) from exc + if not prompts: + raise ValueError(f"{path}: no prompts found") + return prompts + + +def _build_client(args: argparse.Namespace) -> ModelClient: + if args.live: + return OpenAIModelClient() # imports openai lazily; reads OPENAI_API_KEY + return FakeModelClient(mode=args.fake) + + +def _build_evaluator(args: argparse.Namespace, client: ModelClient) -> DriftEvaluator: + if args.judge == "llm": + judge_model = args.judge_model or args.new + return LLMJudgeEvaluator(client, judge_model) + return LexicalSimilarityEvaluator() + + +def _print_report(result: DriftCheckResult, judge: str) -> None: + print( + f"Model drift check: OLD={result.old_model} NEW={result.new_model} " + f"(judge={judge}, threshold={result.threshold:.2f})" + ) + print(f" {'prompt':<12} {'score':>6} verdict") + for entry in result.per_prompt: + verdict = "DRIFT" if entry.drifted else "ok" + print(f" {entry.prompt_id:<12} {entry.score:>6.2f} {verdict}") + print() + drifted = result.drifted_prompts + if result.passed: + print(f"Result: PARITY — all {len(result.per_prompt)} prompts within threshold.") + else: + print( + f"Result: DRIFT — {len(drifted)} of {len(result.per_prompt)} prompts drifted: " + f"{', '.join(drifted)}" + ) + + +def main(argv: Optional[Sequence[str]] = None) -> int: + parser = argparse.ArgumentParser( + prog="python -m examples.model_drift", + description=( + "Check whether a replacement model drifts from a soon-to-be-discontinued " + "model on a fixed prompt set. Exits 1 on drift, 0 on parity." + ), + ) + parser.add_argument("--old", required=True, help="the outgoing model (golden baseline)") + parser.add_argument( + "--new", help="the replacement model (candidate); required unless --record-golden" + ) + parser.add_argument( + "--prompts", + help="JSONL prompt file (one {'id','prompt'} per line); defaults to a built-in set", + ) + + source = parser.add_mutually_exclusive_group() + source.add_argument( + "--fake", + choices=["parity", "drift"], + help="use the deterministic offline client in this mode (default: parity)", + ) + source.add_argument( + "--live", + action="store_true", + help='call OpenAI live (needs: pip install "dprovenancekit[openai]" + OPENAI_API_KEY)', + ) + + parser.add_argument( + "--judge", + choices=["lexical", "llm"], + default="lexical", + help="equivalence evaluator (default: lexical; llm needs a live provider)", + ) + parser.add_argument( + "--judge-model", + help="model the llm judge uses (default: the --new model)", + ) + parser.add_argument( + "--threshold", + type=float, + default=DEFAULT_THRESHOLD, + help=f"minimum per-prompt equivalence to count as no-drift (default: {DEFAULT_THRESHOLD})", + ) + parser.add_argument("--db", help="SQLite path to record the two runs (default: temp file)") + parser.add_argument( + "--golden-db", + help="reuse a golden baseline recorded here (skip calling --old); with " + "--record-golden, write the baseline here instead", + ) + parser.add_argument( + "--record-golden", + action="store_true", + help="record ONLY --old's answers as the golden baseline (into --golden-db) and " + "exit; run this while the outgoing model is still callable", + ) + args = parser.parse_args(argv) + + # Default source is the zero-setup offline fake in parity mode. + if not args.live and args.fake is None: + args.fake = "parity" + + # Argument checks argparse can't express on its own. + if args.record_golden and args.golden_db is None and args.db is None: + print( + "error: --record-golden needs --golden-db (or --db) to write the baseline to", + file=sys.stderr, + ) + return 2 + if not args.record_golden and args.new is None: + print("error: --new is required (unless --record-golden)", file=sys.stderr) + return 2 + + try: + prompts = _load_prompts(args.prompts) if args.prompts else list(_SAMPLE_PROMPTS) + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + try: + client = _build_client(args) + except ImportError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + # Record-only: capture the outgoing model's golden baseline and stop. Do this while the + # old model is still callable; gate candidates against it later with --golden-db. + if args.record_golden: + golden_dest = args.golden_db or args.db + assert golden_dest is not None # guarded above + run_id = record_baseline( + model=args.old, prompts=prompts, client=client, db_path=golden_dest + ) + print(f"Recorded golden baseline for {args.old} -> {golden_dest} (run {run_id})") + return 0 + + assert args.new is not None # guarded above + evaluator = _build_evaluator(args, client) + + result = run_drift_check( + old_model=args.old, + new_model=args.new, + prompts=prompts, + client=client, + evaluator=evaluator, + threshold=args.threshold, + db_path=args.db, + golden_db=args.golden_db, + ) + + _print_report(result, judge=args.judge) + return 0 if result.passed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/model_drift/evaluators.py b/examples/model_drift/evaluators.py new file mode 100644 index 0000000..080c54b --- /dev/null +++ b/examples/model_drift/evaluators.py @@ -0,0 +1,140 @@ +"""Drift evaluators — how similar is "still the same answer"? + +A :class:`DriftEvaluator` scores two answers (old model's, new model's) in ``[0, 1]``: +``1.0`` = equivalent, lower = more drift. YOU own this notion of equivalence. + + • :class:`LexicalSimilarityEvaluator` — stdlib-only surface similarity via + :func:`difflib.SequenceMatcher`. No dependencies, deterministic, good default. + + • :class:`LLMJudgeEvaluator` — asks a model to rate equivalence. It delegates to a + :class:`~examples.model_drift.providers.ModelClient`, so the vendor SDK is imported + lazily by that client only when you actually construct a live provider. + +:func:`as_equivalence_evaluator` adapts any :class:`DriftEvaluator` into the kit's +:class:`~dprovenancekit.AnyEquivalenceEvaluator`, keyed on the ``generation`` events so the +gate compares the two runs' response payloads with your evaluator. +""" + +from __future__ import annotations + +import re +from difflib import SequenceMatcher +from typing import TYPE_CHECKING, Any + +try: # Python 3.8+: typing.Protocol + from typing import Protocol +except ImportError: # pragma: no cover - Protocol always present on supported versions + from typing_extensions import Protocol # type: ignore[assignment] + +from dprovenancekit import AnyEquivalenceEvaluator + +if TYPE_CHECKING: # avoid importing providers at runtime (keeps the module graph acyclic) + from .providers import ModelClient + + +class DriftEvaluator(Protocol): + """Scores the equivalence of two answers in ``[0, 1]`` (1.0 = equivalent).""" + + identifier: str + + def score(self, old_response: str, new_response: str) -> float: + """Return how equivalent ``new_response`` is to ``old_response`` (1.0 = same).""" + ... + + +def _clamp01(value: float) -> float: + # NaN is treated as drift (0.0), never a silent pass. + if value != value: # NaN + return 0.0 + return 0.0 if value < 0.0 else 1.0 if value > 1.0 else value + + +# ── Lexical (stdlib only) ──────────────────────────────────────────────────────── + + +class LexicalSimilarityEvaluator: + """Surface-form similarity via :func:`difflib.SequenceMatcher` (stdlib only). + + ``ratio()`` is ``1.0`` for identical text and falls toward ``0.0`` as the two strings + diverge. Cheap and dependency-free — a sensible default, though it judges *wording*, + not *meaning* (use :class:`LLMJudgeEvaluator` when a reworded-but-equivalent answer + should still count as equivalent). + """ + + identifier = "lexical_seqmatch_v1" + + def score(self, old_response: str, new_response: str) -> float: + return SequenceMatcher(None, old_response, new_response).ratio() + + +# ── LLM-as-judge (delegates to a ModelClient) ──────────────────────────────────── + + +_JUDGE_INSTRUCTIONS = ( + "You are grading whether two answers to the same question are EQUIVALENT in meaning. " + "Reply with a single number between 0 and 1: 1.0 = fully equivalent, 0.0 = unrelated " + "or contradictory. Reply with ONLY the number.\n\n" + "ANSWER A:\n{old}\n\nANSWER B:\n{new}\n\nEquivalence score:" +) + +_NUMBER_RE = re.compile(r"[-+]?\d*\.?\d+") + + +def _parse_score(raw: str) -> float: + """Extract a valid ``[0, 1]`` equivalence score from a judge reply. + + Fails CLOSED. Anything we cannot read as a single in-range score — an error string, an + out-of-range number (``"Error 503"``, ``"9 out of 10"``), or prose with no ``[0, 1]`` + value — scores ``0.0`` (treated as drift) rather than silently passing the check. A + number outside ``[0, 1]`` is NOT clamped up to a pass; it is discarded as "not a score". + When several in-range numbers appear (an off-spec reply), the lowest is used — + conservative toward flagging drift. + """ + in_range = [ + value + for value in (float(m.group()) for m in _NUMBER_RE.finditer(raw or "")) + if 0.0 <= value <= 1.0 + ] + return min(in_range) if in_range else 0.0 + + +class LLMJudgeEvaluator: + """Ask a model to rate the equivalence of two answers. + + Intended for LIVE providers: pass a real :class:`ModelClient` and the model to judge + with. With the offline :class:`FakeModelClient` the reply is not a number, so the score + falls back to ``0.0`` — use the lexical evaluator for offline demos. + """ + + def __init__(self, model_client: "ModelClient", judge_model: str) -> None: + self._client = model_client + self._judge_model = judge_model + self.identifier = f"llm_judge:{judge_model}" + + def score(self, old_response: str, new_response: str) -> float: + prompt = _JUDGE_INSTRUCTIONS.format(old=old_response, new=new_response) + return _parse_score(self._client.complete(self._judge_model, prompt)) + + +# ── Adapter into the kit's equivalence evaluator ───────────────────────────────── + + +def as_equivalence_evaluator(evaluator: DriftEvaluator) -> AnyEquivalenceEvaluator: + """Adapt a :class:`DriftEvaluator` into the kit's :class:`AnyEquivalenceEvaluator`. + + The kit hands the callback the two event PAYLOADS (the ``generation`` events). We + compare the same prompt's answers only — if two payloads carry different ``prompt_id`` + they are not the same step, so they score ``0.0`` (never equivalent). + """ + + def _score(old_payload: Any, new_payload: Any) -> float: + if getattr(old_payload, "prompt_id", None) != getattr(new_payload, "prompt_id", None): + return 0.0 + old_response = getattr(old_payload, "response", "") + new_response = getattr(new_payload, "response", "") + return _clamp01(float(evaluator.score(old_response, new_response))) + + return AnyEquivalenceEvaluator( + evaluator_identifier=evaluator.identifier, + evaluator=_score, + ) diff --git a/examples/model_drift/harness.py b/examples/model_drift/harness.py new file mode 100644 index 0000000..3f2c6b1 --- /dev/null +++ b/examples/model_drift/harness.py @@ -0,0 +1,319 @@ +"""The drift-check harness — record two runs, gate the candidate against the golden. + +:func:`run_drift_check` is the whole story in one call: + + 1. Record the OLD model's answers as a GOLDEN run and the NEW model's answers as a + CANDIDATE run — one CRITICAL ``generation`` event per prompt, carrying + ``{prompt_id, model, response}``, in the same prompt order — into a SQLite store. + 2. Gate the candidate against the golden with the kit's :class:`RegressionGate`, using + your :class:`DriftEvaluator` to decide when two answers are equivalent. A per-prompt + answer below the threshold surfaces as a CRITICAL regression (= drift) and fails. + 3. Return a structured :class:`DriftCheckResult`: overall pass/fail, a per-prompt score, + and exactly which prompts drifted. + +Only the PUBLIC DProvenanceKit API is used. +""" + +from __future__ import annotations + +import os +import tempfile +import uuid +from dataclasses import dataclass +from typing import Dict, List, Optional, Sequence, cast + +from dprovenancekit import ( + AlignmentMode, + AlignmentProfile, + AlignmentStrategy, + DProvenanceKit, + RegressionGate, + RegressionReport, + SQLiteTraceStore, + TraceableEvent, + TracePriority, + TraceRun, +) + +from .evaluators import DriftEvaluator, as_equivalence_evaluator +from .providers import ModelClient + +DEFAULT_THRESHOLD = 0.8 + + +# ── The recorded event: one generation per prompt ──────────────────────────────── + + +@dataclass(frozen=True) +class GenerationEvent(TraceableEvent): + """One model answer to one prompt — the unit the drift check records and gates.""" + + prompt_id: str + model: str + response: str + + @property + def type_identifier(self) -> str: + # Fold the prompt id into the type so the aligner binds each prompt 1:1 across the + # golden and candidate runs, instead of greedily matching by response text. + return f"generation:{self.prompt_id}" + + @property + def priority(self) -> TracePriority: + # CRITICAL so a changed answer escalates to a real regression: the engine only + # raises severity on CRITICAL steps, and that HIGH severity is what "drift" means. + return TracePriority.CRITICAL + + def to_dict(self) -> dict: + return {"prompt_id": self.prompt_id, "model": self.model, "response": self.response} + + @classmethod + def from_dict(cls, data: dict) -> "GenerationEvent": + return cls( + prompt_id=data["prompt_id"], + model=data.get("model", ""), + response=data.get("response", ""), + ) + + +# ── Inputs and results ─────────────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class Prompt: + """A fixed evaluation prompt. ``id`` keys it across the two runs; ``text`` is asked.""" + + id: str + text: str + + +@dataclass(frozen=True) +class PromptDrift: + """The verdict for one prompt: the two answers and their equivalence score.""" + + prompt_id: str + prompt: str + old_response: str + new_response: str + score: float + drifted: bool + + +@dataclass(frozen=True) +class DriftCheckResult: + """The outcome of a drift check. ``passed`` is the gate verdict (True = no drift).""" + + passed: bool + threshold: float + old_model: str + new_model: str + per_prompt: List[PromptDrift] + report: RegressionReport # the kit's RegressionGate report, for the full diagnostic + + @property + def drifted_prompts(self) -> List[str]: + return [p.prompt_id for p in self.per_prompt if p.drifted] + + +# ── Internals ──────────────────────────────────────────────────────────────────── + + +def _clamp01(value: float) -> float: + # NaN is treated as drift (0.0) so a custom evaluator's NaN can never read as a pass. + if value != value: # NaN + return 0.0 + return 0.0 if value < 0.0 else 1.0 if value > 1.0 else value + + +def _drift_profile(threshold: float) -> AlignmentProfile: + """A strict-audit-shaped profile whose EFFECTIVE response-similarity cutoff is + ``threshold``. + + ``semantic_threshold`` is compared to the COMBINED weighted score, not the raw + evaluator output. Under strict-audit weights (type 0.5 / payload 0.5) with a matching + ``type_identifier``, combined = ``0.5 + 0.5 * response_similarity``. To make the answer + cutoff equal ``threshold`` we set ``semantic_threshold = 0.5 + 0.5 * threshold``: an + answer scoring below ``threshold`` lands below it and is flagged as drift. + """ + return AlignmentProfile( + strategy=AlignmentStrategy.STRICT_AUDIT, + version=1, + type_weight=0.5, + payload_weight=0.5, + structural_weight=0.0, + temporal_weight=0.0, + semantic_threshold=0.5 + 0.5 * threshold, + max_ambiguous_candidates=1, + ambiguity_delta_threshold=0.0, + alignment_mode=AlignmentMode.LINEAR, + ) + + +def _record_model_run( + kit: DProvenanceKit, + store: SQLiteTraceStore, + model: str, + prompts: Sequence[Prompt], + client: ModelClient, +) -> uuid.UUID: + """Ask ``model`` every prompt and record one CRITICAL generation event per answer.""" + with kit.run(context_id=model, store=store) as run: + # Fixed engine name for both runs: the only thing that differs golden-vs-candidate + # is the answer text, so the gate isolates response drift. + with kit.with_engine("llm"): + for prompt in prompts: + response = client.complete(model, prompt.text) + kit.record( + GenerationEvent(prompt_id=prompt.id, model=model, response=response) + ) + return run.run_id + + +def _responses_by_prompt(run: TraceRun) -> Dict[str, str]: + out: Dict[str, str] = {} + for event in run.events: + payload = cast(GenerationEvent, event.payload) + out[payload.prompt_id] = payload.response + return out + + +def _latest_run_by_context(store: SQLiteTraceStore, context_id: str) -> Optional[TraceRun]: + """Return the newest recorded run with this ``context_id`` (or ``None``). + + ``list_run_metadata`` yields rows newest-first, so the first context match is the newest. + """ + for row in store.list_run_metadata(): + if row.context_id == context_id: + return store.get_run(uuid.UUID(row.run_id)) + return None + + +# ── The public calls ───────────────────────────────────────────────────────────── + + +def record_baseline( + *, + model: str, + prompts: Sequence[Prompt], + client: ModelClient, + db_path: str, +) -> uuid.UUID: + """Record ONLY ``model``'s answers as a golden run into ``db_path``. + + Run this while the outgoing model is still callable. The db becomes the durable baseline + you gate future candidates against via ``run_drift_check(..., golden_db=db_path)`` — so + the check keeps working after the old model is retired. Returns the golden run id. + """ + with SQLiteTraceStore(GenerationEvent, db_path, start_writer=False) as store: + kit = DProvenanceKit(GenerationEvent) + return _record_model_run(kit, store, model, prompts, client) + + +def run_drift_check( + *, + old_model: str, + new_model: str, + prompts: Sequence[Prompt], + client: ModelClient, + evaluator: DriftEvaluator, + threshold: float = DEFAULT_THRESHOLD, + db_path: Optional[str] = None, + golden_db: Optional[str] = None, +) -> DriftCheckResult: + """Record both models on ``prompts`` and gate the new model for drift. + + Args: + old_model: the soon-to-be-discontinued model (its answers are the GOLDEN baseline). + new_model: the replacement model (the CANDIDATE). + prompts: the fixed prompt set — you own it; the check is only as good as it is. + client: your :class:`ModelClient` (offline fake, OpenAI, or your own provider). + evaluator: your :class:`DriftEvaluator` — what "still the same answer" means. + threshold: minimum per-prompt equivalence in ``[0, 1]`` to count as no-drift. + db_path: where to record the runs. ``None`` uses a throwaway temp db. + golden_db: reuse a golden baseline recorded earlier with :func:`record_baseline` + (keyed on ``old_model``) instead of calling the old model — this is what lets you + gate after the old model is retired. ``None`` records the old model now. + + Returns: + A :class:`DriftCheckResult` with the pass/fail verdict, per-prompt scores, and the + list of drifted prompt ids. + """ + threshold = _clamp01(threshold) + + # The golden (old model). Reuse a saved baseline when given — that is what lets you gate + # after the old model is retired; otherwise record it now, beside the candidate. + golden: Optional[TraceRun] = None + if golden_db is not None: + with SQLiteTraceStore(GenerationEvent, golden_db, start_writer=False) as gstore: + golden = _latest_run_by_context(gstore, old_model) + if golden is None: + raise ValueError( + f"no golden run for model '{old_model}' in {golden_db}; record one with " + "record_baseline(...) while the old model is still callable" + ) + + created_temp = db_path is None + if db_path is None: + handle, db_path = tempfile.mkstemp(prefix="model_drift_", suffix=".sqlite") + os.close(handle) + + try: + # Record the candidate (and the golden too, unless a saved one was reused) into a + # trace store. get_run flushes, then fetches a detached TraceRun. + with SQLiteTraceStore(GenerationEvent, db_path, start_writer=False) as store: + kit = DProvenanceKit(GenerationEvent) + if golden is None: + golden = store.get_run( + _record_model_run(kit, store, old_model, prompts, client) + ) + candidate = store.get_run( + _record_model_run(kit, store, new_model, prompts, client) + ) + + if golden is None or candidate is None: # pragma: no cover - defensive + raise RuntimeError("failed to read back the recorded runs from the trace store") + + # 2. Gate the candidate against the golden with our answer-equivalence evaluator. + # A per-prompt answer below `threshold` falls below the profile's semantic + # threshold, flags that CRITICAL step as changed-beyond-equivalence, and fails. + gate = RegressionGate( + profile=_drift_profile(threshold), + evaluator=as_equivalence_evaluator(evaluator), + ) + report = gate.check(golden, candidate) + + # 3. Per-prompt breakdown, read back from the recorded runs. + old_by_prompt = _responses_by_prompt(golden) + new_by_prompt = _responses_by_prompt(candidate) + per_prompt: List[PromptDrift] = [] + for prompt in prompts: + old_response = old_by_prompt.get(prompt.id, "") + new_response = new_by_prompt.get(prompt.id, "") + score = _clamp01(float(evaluator.score(old_response, new_response))) + per_prompt.append( + PromptDrift( + prompt_id=prompt.id, + prompt=prompt.text, + old_response=old_response, + new_response=new_response, + score=score, + drifted=score < threshold, + ) + ) + + return DriftCheckResult( + passed=report.passed, + threshold=threshold, + old_model=old_model, + new_model=new_model, + per_prompt=per_prompt, + report=report, + ) + finally: + if created_temp: + # Best-effort cleanup of the throwaway db (and its WAL/SHM sidecars). + for suffix in ("", "-wal", "-shm"): + try: + os.remove(db_path + suffix) + except OSError: + pass diff --git a/examples/model_drift/prompts.sample.jsonl b/examples/model_drift/prompts.sample.jsonl new file mode 100644 index 0000000..bbcaf9f --- /dev/null +++ b/examples/model_drift/prompts.sample.jsonl @@ -0,0 +1,8 @@ +{"id": "greeting", "prompt": "How should I greet a new customer in a support chat?"} +{"id": "refund", "prompt": "What is a reasonable refund window for a digital product?"} +{"id": "units", "prompt": "How many meters are in a kilometer?"} +{"id": "capital", "prompt": "What is the capital of France?"} +{"id": "password", "prompt": "What makes a password strong?"} +{"id": "summary", "prompt": "Summarize the benefits of unit testing in one sentence."} +{"id": "email", "prompt": "Write a one-line out-of-office reply."} +{"id": "advice", "prompt": "Give one tip for staying focused while working from home."} diff --git a/examples/model_drift/providers.py b/examples/model_drift/providers.py new file mode 100644 index 0000000..1d7b022 --- /dev/null +++ b/examples/model_drift/providers.py @@ -0,0 +1,164 @@ +"""Model providers — the thing that turns a prompt into an answer. + +A :class:`ModelClient` is any object with ``complete(model, prompt) -> str``. The kit never +calls a model for you; you own the provider (and its API key). + +Two clients ship here: + + • :class:`FakeModelClient` — deterministic, offline. Derives each answer from the + prompt text via a pure rule (no network, no randomness), so the whole drift flow + runs with zero setup. It can simulate PARITY (both models answer identically) or + DRIFT (the newer model diverges on a subset of prompts) for demos, tests and CI. + + • :class:`OpenAIModelClient` — a live client. It imports ``openai`` LAZILY, inside + ``__init__``, so *this module imports fine even when ``openai`` is not installed*. + Install it with ``pip install "dprovenancekit[openai]"`` to run a live check. +""" + +from __future__ import annotations + +import hashlib +from typing import Callable, List, Optional + +try: # Python 3.8+: typing.Protocol + from typing import Protocol +except ImportError: # pragma: no cover - Protocol always present on supported versions + from typing_extensions import Protocol # type: ignore[assignment] + + +class ModelClient(Protocol): + """Anything that can answer a prompt with a given model. + + Implement this to plug in your own provider (Anthropic, a local model, an internal + gateway, …). ``complete`` must be deterministic enough to baseline — set temperature + to 0 for live providers so a re-run of the golden reproduces it. + """ + + def complete(self, model: str, prompt: str) -> str: + """Return the model's answer to ``prompt``.""" + ... + + +# ── Deterministic offline client (demos / tests / CI) ──────────────────────────── + + +def _default_drift_selector(prompt: str) -> bool: + """Which prompts drift in ``drift`` mode: a stable, content-derived ~half of them.""" + digest = hashlib.sha1(prompt.strip().lower().encode("utf-8")).hexdigest() + return int(digest, 16) % 2 == 0 + + +def _topic(prompt: str) -> str: + words = prompt.strip().rstrip("?.!").split() + return " ".join(words[:4]) if words else "the question" + + +# A small fixed vocabulary the drift answers are drawn from. Two different model names hash +# to two different word sequences, so their answers diverge substantially (low similarity). +_LEXICON = ( + "policy update depends context reverse affirm deny caveat revised current legacy " + "nuance threshold escalate default guidance evidence tradeoff baseline divergent " + "clarify restrict permit outdated" +).split() + + +class FakeModelClient: + """A deterministic, offline stand-in for a real model. + + The answer is a pure function of the prompt (and, on the drifting subset in ``drift`` + mode, of the model name), so runs are perfectly reproducible. + + Modes: + * ``"parity"`` — the answer depends ONLY on the prompt, so every model returns the + same text: a replacement can never drift. + * ``"drift"`` — on the prompts selected by ``drifts_on`` the answer also depends on + the model name, so two different models produce different text (drift); all other + prompts stay in parity. + + Pass ``drifts_on`` to control exactly which prompts drift (handy in tests); it defaults + to a stable hash of the prompt that drifts roughly half of any prompt set. + """ + + def __init__( + self, + *, + mode: str = "parity", + drifts_on: Optional[Callable[[str], bool]] = None, + ) -> None: + if mode not in ("parity", "drift"): + raise ValueError(f"mode must be 'parity' or 'drift', got {mode!r}") + self.mode = mode + self._drifts_on = drifts_on or _default_drift_selector + + def complete(self, model: str, prompt: str) -> str: + if self.mode == "drift" and self._drifts_on(prompt): + return self._drifted_answer(model, prompt) + return self._base_answer(prompt) + + @staticmethod + def _base_answer(prompt: str) -> str: + topic = _topic(prompt) + return ( + f"Regarding {topic}: the established answer is that the standard, well-sourced " + f"guidance applies, with the usual caveats noted." + ) + + @staticmethod + def _drifted_answer(model: str, prompt: str) -> str: + # A model-derived word sequence, so two different models produce genuinely + # different answers to the same prompt (not just a different name tag). The digest + # of (model, prompt) picks distinct words from the lexicon; distinct model names + # give distinct digests and therefore substantially different text. + topic = _topic(prompt) + digest = hashlib.sha256(f"{model}::{prompt}".encode("utf-8")).digest() + picks: List[str] = [] + seen = set() + for byte in digest: + word = _LEXICON[byte % len(_LEXICON)] + if word not in seen: + seen.add(word) + picks.append(word) + if len(picks) >= 12: + break + return f"On {topic} the position is now: " + ", ".join(picks) + "." + + +# ── Live client (opt-in; lazy vendor import) ───────────────────────────────────── + + +class OpenAIModelClient: + """A live :class:`ModelClient` backed by the OpenAI Python SDK. + + The ``openai`` import happens INSIDE ``__init__`` (never at module top), so importing + this module — and using :class:`FakeModelClient` — works with ``openai`` absent and no + API key. Constructing this client is what requires the dependency. + + The key is read from the ``OPENAI_API_KEY`` environment variable unless you pass one + explicitly. Temperature defaults to 0 so a golden baseline is reproducible. + """ + + def __init__( + self, + *, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + temperature: float = 0.0, + ) -> None: + try: + from openai import OpenAI + except ImportError as exc: # pragma: no cover - exercised only without openai + raise ImportError( + "OpenAIModelClient requires the 'openai' package. Install it with:\n" + ' pip install "dprovenancekit[openai]"' + ) from exc + + self._temperature = temperature + self._client = OpenAI(api_key=api_key, base_url=base_url) + + def complete(self, model: str, prompt: str) -> str: + response = self._client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + temperature=self._temperature, + ) + return response.choices[0].message.content or "" diff --git a/pyproject.toml b/pyproject.toml index 78aa920..0023f9f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,9 @@ dev = ["pytest>=7.0", "fastapi", "httpx", "IPython", "pytest-asyncio", "langchai # needed to wire DProvenanceKit into the corresponding framework. langchain = ["langchain-core>=0.2"] openai-agents = ["openai-agents>=0.1"] +# The OpenAI Python SDK, used by the examples/model_drift live client. The core stays pure +# standard library; this extra is only needed to run a drift check against a live OpenAI model. +openai = ["openai>=1.0"] llama-index = ["llama-index-core"] # The adapter is a crewai event-bus listener (crewai.events.BaseEventListener), verified # against the public `crewai.events` API stabilized in the 1.x line. It does NOT use diff --git a/tests/test_model_drift.py b/tests/test_model_drift.py new file mode 100644 index 0000000..13c30d6 --- /dev/null +++ b/tests/test_model_drift.py @@ -0,0 +1,277 @@ +"""Deterministic tests for the ``examples/model_drift`` package. + +Every test runs offline with the package's :class:`FakeModelClient` — no network, no API +key, no randomness — so they are safe on every PR. They pin the end-to-end story: + + * parity → the gate PASSES and reports no drift, + * drift → the gate FAILS and names exactly the prompts that drifted, + * the example imports and runs with ``openai`` NOT installed (the lazy-import contract), + * :class:`LexicalSimilarityEvaluator` scores 1.0 for identical text and lower — and + monotonically — as the text diverges. +""" + +from __future__ import annotations + +import importlib +import os +import sys + +import pytest + +# Make the ``examples`` namespace package importable when the suite is run with a bare +# ``pytest`` (``python -m pytest`` already puts the repo root on sys.path). +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from examples.model_drift.evaluators import LexicalSimilarityEvaluator # noqa: E402 +from examples.model_drift.harness import Prompt, run_drift_check # noqa: E402 +from examples.model_drift.providers import FakeModelClient # noqa: E402 + +# A tiny fixed prompt set. ``drifts_on`` keys on the prompt TEXT (what the client receives), +# so we can make exactly one prompt drift and assert on it. +_PROMPTS = [ + Prompt("capital", "What is the capital of France?"), + Prompt("meters", "How many meters are in a kilometer?"), + Prompt("focus", "Give one tip for staying focused."), +] + + +def _drifts_on_focus(text: str) -> bool: + return "focused" in text.lower() + + +# ── parity → PASS, no drift ────────────────────────────────────────────────────── + + +def test_parity_passes_with_no_drift(): + result = run_drift_check( + old_model="old-model", + new_model="new-model", + prompts=_PROMPTS, + client=FakeModelClient(mode="parity"), + evaluator=LexicalSimilarityEvaluator(), + threshold=0.8, + ) + + assert result.passed is True + assert result.report.passed is True + assert result.drifted_prompts == [] + # In parity mode every model returns the same text, so every score is a perfect 1.0. + assert all(entry.score == 1.0 and not entry.drifted for entry in result.per_prompt) + + +# ── drift → FAIL, names the drifted prompts ────────────────────────────────────── + + +def test_drift_fails_and_reports_drifted_prompts(): + result = run_drift_check( + old_model="old-model", + new_model="new-model", + prompts=_PROMPTS, + client=FakeModelClient(mode="drift", drifts_on=_drifts_on_focus), + evaluator=LexicalSimilarityEvaluator(), + threshold=0.8, + ) + + assert result.passed is False + assert result.report.passed is False + # The kit's gate escalates a changed CRITICAL step to a HIGH regression. + assert result.report.regression_level.value == "high" + # Only the one drifting prompt is flagged; the parity prompts stay clean. + assert result.drifted_prompts == ["focus"] + by_id = {entry.prompt_id: entry for entry in result.per_prompt} + assert by_id["focus"].drifted and by_id["focus"].score < 0.8 + assert by_id["capital"].score == 1.0 and not by_id["capital"].drifted + assert by_id["meters"].score == 1.0 and not by_id["meters"].drifted + + +def test_cli_exit_codes_mirror_the_gate(): + # The CLI returns a shell exit code: 0 on parity, 1 on drift (like ``dpk gate``). + from examples.model_drift.__main__ import main + + parity_code = main(["--old", "old-model", "--new", "new-model", "--fake", "parity"]) + drift_code = main(["--old", "old-model", "--new", "new-model", "--fake", "drift"]) + + assert parity_code == 0 + assert drift_code == 1 + + +# ── imports + runs with openai NOT installed (the lazy-import contract) ─────────── + + +def test_example_imports_and_runs_without_openai(monkeypatch): + # Simulate ``openai`` being absent: setting the entry to None makes ``import openai`` + # raise ImportError, regardless of whether the SDK is installed in this environment. + monkeypatch.setitem(sys.modules, "openai", None) + monkeypatch.delitem(sys.modules, "examples.model_drift.providers", raising=False) + + # The module imports fine with openai absent — nothing imports it at module top. + providers = importlib.import_module("examples.model_drift.providers") + + # The deterministic fake path runs with no SDK and no key. + client = providers.FakeModelClient(mode="parity") + assert client.complete("any-model", "hello?") == client.complete("other-model", "hello?") + + # Constructing the live client is the only thing that needs the SDK, and it fails with a + # clear, actionable error rather than an import-time crash. + with pytest.raises(ImportError, match="openai"): + providers.OpenAIModelClient() + + +# ── LexicalSimilarityEvaluator: 1.0 for identical, lower + monotonic as it diverges ── + + +def test_lexical_similarity_is_one_for_identical_and_monotonic_on_divergence(): + evaluator = LexicalSimilarityEvaluator() + base = "The refund window for digital goods is thirty days." + + identical = evaluator.score(base, base) + minor = evaluator.score(base, "The refund window for digital goods is thirty-one days.") + major = evaluator.score(base, "The refund window for digital goods is fourteen days total.") + unrelated = evaluator.score(base, "Photosynthesis converts sunlight into chemical energy.") + + # Identical text is a perfect match. + assert identical == 1.0 + # Every score is a valid similarity in [0, 1]. + for value in (identical, minor, major, unrelated): + assert 0.0 <= value <= 1.0 + # More divergence => a lower score (monotonic-ish). + assert identical > minor > major > unrelated + + +# ── LLM judge fails CLOSED on unparseable / out-of-range replies (regression) ────── + + +@pytest.mark.parametrize( + "reply, expected", + [ + ("1.0", 1.0), + ("0.85", 0.85), + ("0", 0.0), + (" 0.5 ", 0.5), + ("Error 503: service unavailable", 0.0), + ("HTTP 429 Too Many Requests", 0.0), + ("I'd rate these a 9 out of 10", 0.0), + ("5 days vs 30 days, not equivalent", 0.0), + ("", 0.0), + ("totally different", 0.0), + ], +) +def test_llm_judge_parse_fails_closed(reply, expected): + # A judge reply we can't read as a single [0, 1] score must score 0.0 (drift), never + # get clamped up into a silent pass. Regression test for the fail-open parser bug. + from examples.model_drift.evaluators import _parse_score + + assert _parse_score(reply) == expected + + +def test_llm_judge_error_string_reads_as_drift(): + # A provider that returns an ERROR STRING (not a number) must fail closed to drift, so a + # broken judge can't silently pass a real regression. + from examples.model_drift.evaluators import LLMJudgeEvaluator + + class _ErrorClient: + def complete(self, model: str, prompt: str) -> str: + return "Error 503: service unavailable" + + judge = LLMJudgeEvaluator(_ErrorClient(), judge_model="judge-x") + assert judge.score("5 days", "30 days") == 0.0 + + +def test_clamp01_treats_nan_as_drift(): + # A custom evaluator returning NaN must read as drift, not slip through the clamp. + from examples.model_drift.evaluators import _clamp01 + + assert _clamp01(float("nan")) == 0.0 + + +# ── Saved golden baseline: capture once, gate later without the old model ────────── + + +def test_saved_golden_is_reused_without_calling_the_old_model(tmp_path): + from examples.model_drift.harness import record_baseline + + golden_db = str(tmp_path / "golden.sqlite") + + # 1. Capture the OLD model's baseline while it is still callable. + record_baseline( + model="old-model", + prompts=_PROMPTS, + client=FakeModelClient(mode="parity"), + db_path=golden_db, + ) + + # 2. A client that answers the NEW model but blows up if the retired OLD model is called + # — proving the gate reuses the saved golden instead of re-recording it. + class _RetiredOldModelClient: + def __init__(self) -> None: + self._fake = FakeModelClient(mode="parity") + + def complete(self, model: str, prompt: str) -> str: + if model == "old-model": + raise AssertionError("the retired old model must not be called") + return self._fake.complete(model, prompt) + + result = run_drift_check( + old_model="old-model", + new_model="new-model", + prompts=_PROMPTS, + client=_RetiredOldModelClient(), + evaluator=LexicalSimilarityEvaluator(), + threshold=0.8, + golden_db=golden_db, + ) + + # Parity answers depend only on the prompt, so the reused golden and the fresh candidate + # match: no drift. + assert result.passed is True + assert result.drifted_prompts == [] + + +def test_run_drift_check_errors_when_golden_db_has_no_baseline(tmp_path): + from dprovenancekit import SQLiteTraceStore + + from examples.model_drift.harness import GenerationEvent + + empty_db = str(tmp_path / "empty.sqlite") + with SQLiteTraceStore(GenerationEvent, empty_db, start_writer=False): + pass # a store file with no runs recorded + + with pytest.raises(ValueError, match="no golden run"): + run_drift_check( + old_model="old-model", + new_model="new-model", + prompts=_PROMPTS, + client=FakeModelClient(mode="parity"), + evaluator=LexicalSimilarityEvaluator(), + threshold=0.8, + golden_db=empty_db, + ) + + +def test_cli_record_golden_then_gate_reuses_it(tmp_path): + from examples.model_drift.__main__ import main + + golden_db = str(tmp_path / "golden.sqlite") + + assert ( + main( + ["--old", "old-model", "--record-golden", "--golden-db", golden_db, "--fake", "parity"] + ) + == 0 + ) + # Gate a candidate against the saved golden — parity → exit 0. + assert ( + main( + ["--old", "old-model", "--new", "new-model", "--golden-db", golden_db, "--fake", "parity"] + ) + == 0 + ) + + +def test_cli_usage_errors_return_2(): + from examples.model_drift.__main__ import main + + # --record-golden with nowhere to write the baseline. + assert main(["--old", "old-model", "--record-golden", "--fake", "parity"]) == 2 + # A gate run with no --new. + assert main(["--old", "old-model", "--fake", "parity"]) == 2