Case Study: A Cassette Proxy That Records a Model Once and Replays It in CI

Your integration tests should not need a network, an API key, or a budget to prove that your prompt-building code still assembles the right payload. The cheapest way to get there is to call the model exactly once, store the response under a hash of the request, and replay that response forever after. This is a case study of that small tool: what it does, the code that implements it, where it breaks, and who should walk away from it.

The project is deliberately tiny — one file, standard library only, three modes — because a record/replay layer that you cannot read in ten minutes is a layer you will not debug at 2 a.m. Treat the timings below as things you measure on your own hardware, not as published benchmarks.

Background: the test suite that quietly became an HTTP client

A prompt-assembly module usually starts life as pure functions: you build messages, you template a system prompt, you trim history to a token budget. Then someone adds one end-to-end test that calls the real endpoint, and your unit suite becomes an HTTP client with a credit card attached.

The symptoms are predictable. CI fails when the provider has a bad afternoon, pull requests from forks cannot run because secrets are withheld, and someone eventually adds @pytest.mark.skip to the only test that would have caught a broken payload. None of that is a model problem; it is a boundary problem, and it has a boring, well-understood fix.

Goal: three modes, one file, no dependencies

The tool needs to do four things and nothing else: compute a stable cache key from a request body, forward the request upstream in record mode, persist the response, and serve it in replay mode. Missing cassettes must fail loudly rather than silently falling through to the network, because a silent fallback reintroduces exactly the flakiness you are trying to delete.

• record — call upstream, save the response, return it.
• replay — read from disk only; a miss returns HTTP 409 with the cache key.
• off — pass everything through, for the days you actually want live model behaviour.

The artifact: a stdlib cassette proxy

The following file runs as-is on Python 3.10+ against any OpenAI-compatible chat endpoint. It buffers responses, so streaming is out of scope for this version; that limitation is discussed below.

# cassette_proxy.py - stdlib only. CASSETTE_MODE = off | record | replay
import hashlib, json, os, pathlib, sys, urllib.error, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

UPSTREAM = os.environ.get("UPSTREAM_BASE", "https://api.example.com/v1").rstrip("/")
MODE = os.environ.get("CASSETTE_MODE", "replay")
STORE = pathlib.Path(os.environ.get("CASSETTE_DIR", "cassettes"))
PORT = int(os.environ.get("PORT", "8089"))
API_KEY = os.environ.get("UPSTREAM_API_KEY", "")
VOLATILE = {"stream", "user", "metadata", "request_id"}  # never part of the key


def canonical(payload: dict) -> str:
    body = {k: v for k, v in payload.items() if k not in VOLATILE}
    return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def key_for(payload: dict) -> str:
    return hashlib.sha256(canonical(payload).encode("utf-8")).hexdigest()[:32]


def load(key: str):
    path = STORE / f"{key}.json"
    return json.loads(path.read_text("utf-8"))["response"] if path.exists() else None


def save(key: str, payload: dict, response: dict) -> None:
    STORE.mkdir(parents=True, exist_ok=True)
    record = {"key": key, "request": payload, "response": response}
    (STORE / f"{key}.json").write_text(json.dumps(record, indent=2, ensure_ascii=False))


def call_upstream(raw: bytes, path: str) -> dict:
    req = urllib.request.Request(
        f"{UPSTREAM}{path}", data=raw, method="POST",
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}"},
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as err:
        if err.code == 429:
            raise SystemExit("upstream rate limit hit; stop recording and resume later")
        raise


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

    def log_message(self, fmt, *args):
        print(f"[{MODE}] {fmt % args}", file=sys.stderr)

    def send_json(self, status: int, body: dict) -> None:
        blob = json.dumps(body).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(blob)))
        self.end_headers()
        self.wfile.write(blob)

    def do_GET(self):
        if self.path == "/healthz":
            return self.send_json(200, {"mode": MODE})
        self.send_json(404, {"error": "not_found"})

    def do_POST(self):
        raw = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        payload = json.loads(raw or b"{}")
        key = key_for(payload)
        if MODE == "replay":
            hit = load(key)
            print(f"{key} {'HIT' if hit else 'MISS'}", file=sys.stderr)
            return self.send_json(200, hit) if hit else self.send_json(409, {"error": "cassette_miss", "key": key})
        response = call_upstream(raw, self.path)
        if MODE == "record":
            save(key, payload, response)
        self.send_json(200, response)


if __name__ == "__main__":
    print(f"cassette proxy on :{PORT} mode={MODE} upstream={UPSTREAM}", file=sys.stderr)
    ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()

Why the key is a hash of the canonical body

Hashing the whole body minus a small volatile set means any meaningful change — a new system prompt line, a different temperature, an edited user message — produces a different key, and therefore a cassette miss. That is the behaviour you want: a miss is a diff you can review, not a runtime surprise.

Strip only fields that genuinely carry no semantic weight for your tests, such as user, metadata, and request IDs. Everything else belongs in the key, including sampling parameters, because temperature=0 and temperature=0.7 are different experiments even when they look alike.

Recording once, against a free tier

Record mode needs an OpenAI-compatible upstream and a key. Record mode is also where a free model tier is genuinely useful: recording is a one-time, low-volume, human-supervised act, so a promotional allowance covers it comfortably. MonkeyCode's free model access and its free server option are what this walkthrough assumes for the recording step and for hosting the replay service; point UPSTREAM_BASE at whatever compatible endpoint you actually hold credentials for.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

The allowance is a promotion, not a permanent contract, so read the live page before you size anything around it — the advertised figure at the time of writing was a 10M-token allowance, and promotional numbers rotate. Do not build a plan that only works at the current figure.

# record: run once, watch the log, commit the cassettes
CASSETTE_MODE=record UPSTREAM_BASE="$PROVIDER_BASE_URL" UPSTREAM_API_KEY="$PROVIDER_KEY" \
  python cassette_proxy.py &

CASSETTE_MODE=record pytest tests/test_end_to_end.py -q

git add cassettes && git commit -m "record cassettes for changelog summarizer"

Wiring replay into pytest without secrets

Point your client at the proxy and give it a dummy key; the fixture starts the server, waits for /healthz, and tears it down afterwards. CI then runs with no provider secret at all, which also means fork pull requests can run the full suite.

# conftest.py
import os, subprocess, sys, time, urllib.request
import pytest

@pytest.fixture(scope="session", autouse=True)
def cassette_server():
    env = {**os.environ, "CASSETTE_MODE": "replay", "PORT": "8089"}
    proc = subprocess.Popen([sys.executable, "cassette_proxy.py"], env=env)
    for _ in range(100):
        try:
            urllib.request.urlopen("http://127.0.0.1:8089/healthz", timeout=0.5)
            break
        except OSError:
            time.sleep(0.1)
    yield "http://127.0.0.1:8089/v1"
    proc.terminate()
    proc.wait(timeout=10)

def test_summary_payload(cassette_server):
    client = make_client(base_url=cassette_server, api_key="cassette")
    out = summarize_commits(client, ["fix: guard null", "feat: retry cap"])
    assert out.startswith("Two commits")

Hosting replay where CI can reach it

If your CI runners are ephemeral and you would rather not start a process per job, deploy the same file in replay mode and bake the cassettes/ directory into the image or mount it from the repository. The free server option is a reasonable host for a read-only replay service precisely because the workload is trivial: it serves JSON files and never calls anything.

Check two things before you rely on it: whether long-running idle processes stay warm on your plan, and whether the filesystem persists across restarts. If persistence is unclear, ship the cassettes inside the artifact; that removes the question entirely and keeps replay deterministic.

Situation	Mode	Reasoning
New prompt, first run	record	Pay once, then freeze the behaviour
Every pull request	replay	No secret, no network, no rate limit
Validating model quality	off	A cassette would hide real drift
Upstream model retired	record	Re-record deliberately, review the diff
Debugging a miss	replay	The 409 body carries the key to inspect

What to measure instead of trusting a claim

The proxy prints one line per request with the key and HIT or MISS, which is enough to compute the only three numbers that matter. Measure wall-clock time of the suite in each mode, the count of cassette misses on a clean checkout, and the number of outbound requests your CI job makes. A non-zero miss count on a clean checkout is a bug, not a warmup.

# compare modes on your machine; write your own numbers down
for m in replay off; do
  start=$(date +%s%N)
  CASSETTE_MODE=$m pytest -q >/dev/null
  echo "$m: $(( ($(date +%s%N) - start) / 1000000 )) ms"
done

Limitations you will hit within a month

• Cassettes are snapshots. When you change a prompt, every affected test misses at once; expect to re-record in batches and review those diffs like code.
• Streaming responses are not supported here. Buffer them, or put the proxy in front of a non-streaming test-only code path.
• Recorded payloads contain whatever you sent. Scrub secrets and personal data before committing, and treat the cassettes/ directory as production-adjacent data.
• The proxy reads the Authorization header only from the environment and never stores it, but it does log request lines; keep the volume out of shared logs if prompts are sensitive.
• Model deprecations arrive without warning. A cassette will happily keep replaying behaviour from a model you can no longer call.

Who should not use this

Skip this approach if your tests exist to measure output quality, because replay guarantees you will never notice a regression in the model itself. Skip it if your product is a streaming chat interface where the token-by-token path is the thing under test. And skip it for a single throwaway script; the ceremony only pays off once you have a suite you intend to keep running.

Lessons learned

1. Fail closed. A 409 on a missing cassette caught two accidental payload changes in the first week that a network fallback would have masked.
2. Hash the request, not the test name. Test names get renamed; request bodies are the real contract.
3. Keep the artifact boring. Standard library and one file meant the reviewer could read the whole thing before approving the CI change.
4. Separate recording from reviewing. Record in one commit, inspect the cassette diff in another, and never let a re-record ride along with a feature change.

If you want to try the recording step without a paid key, MonkeyCode's free model access plus the free server option is enough to run this workflow end to end; start by recording a single endpoint, then convert the suite one test at a time.