Scrubbing PII From Recorded LLM Test Fixtures Before Committing Them

A recorded cassette from a real LLM call is a verbatim copy of what a user typed and what the model said back about it. Header-level redaction, which is what most secret-scanning setups do, does not touch either. The data you need to remove is inside a JSON body, and on a streaming endpoint it is spread across a few hundred separate fragments.

Where the PII actually is

Record one call to a chat completions endpoint and enumerate what lands on disk. There are five places, and generic tooling reliably catches only the first.

• Authorization headers and API keys. The one thing every tool handles. VCR-style libraries all have a header filter.
• The request body’s message array. messages[].content holds whatever the user wrote — names, addresses, medical detail, an entire pasted email thread. This is the main event and it is inside a JSON string inside a YAML file.
• The system prompt. Frequently contains customer names, internal account identifiers, retrieved documents, and often the prompt itself is something you would rather not publish.
• The response body. The model repeats back what it was given, especially on summarisation and extraction tasks. A scrubber that only handles the request leaves the same data in the reply.
• Metadata. Request ids, organisation ids, and the user field that many APIs accept for abuse tracking, which is frequently a real account id.

Scrubbing at record time

Scrub as the cassette is written, so unredacted data never reaches disk at all. In vcrpy the hooks are filter_headers, filter_query_parameters, filter_post_data_parameters, before_record_request and before_record_response, all documented in vcrpy’s advanced usage guide.

One option on that list is not optional here. decode_compressed_response=True makes vcrpy decompress gzip and deflate bodies before recording. Without it the response body on disk is compressed bytes, your regex matches nothing, every test passes, and the cassette contains the user’s data in a form that is trivially recoverable. A scrubber that silently matches nothing is the worst outcome available, because it produces confidence.

# tests/conftest.py
import json
import re
import vcr

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
PHONE = re.compile(r"\+?\d[\d ()-]{7,}\d")
CARD  = re.compile(r"\b(?:\d[ -]*?){13,19}\b")

REDACTIONS = ((EMAIL, "[email protected]"), (PHONE, "+10000000000"), (CARD, "4111111111111111"))


def scrub_text(text: str) -> str:
    for pattern, replacement in REDACTIONS:
        text = pattern.sub(replacement, text)
    return text


def scrub_json_strings(node):
    """Walk a decoded JSON document and scrub every string leaf."""
    if isinstance(node, str):
        return scrub_text(node)
    if isinstance(node, list):
        return [scrub_json_strings(v) for v in node]
    if isinstance(node, dict):
        return {k: scrub_json_strings(v) for k, v in node.items()}
    return node


def scrub_body(raw):
    if raw is None:
        return raw
    text = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else raw
    try:
        document = json.loads(text)
    except ValueError:
        return scrub_text(text)          # not JSON: fall back to plain text
    document.pop("user", None)           # drop the abuse-tracking account id
    return json.dumps(scrub_json_strings(document))


def before_record_request(request):
    if request.body:
        body = scrub_body(request.body)
        request.body = body.encode() if isinstance(request.body, bytes) else body
    return request


def before_record_response(response):
    response["body"]["string"] = scrub_body(response["body"]["string"])
    response["headers"].pop("Set-Cookie", None)
    return response


llm_vcr = vcr.VCR(
    cassette_library_dir="tests/cassettes",
    record_mode="once",
    decode_compressed_response=True,
    filter_headers=["authorization", "x-api-key", "openai-organization", "cookie"],
    filter_query_parameters=["api_key"],
    before_record_request=before_record_request,
    before_record_response=before_record_response,
)

Two notes on the shape of that code. The JSON walk scrubs every string leaf rather than a list of known paths, because message arrays nest differently across providers and a path-based scrubber breaks silently when a provider adds a field. And scrub_body falls back to plain-text scrubbing when the body is not JSON, which is what makes it safe to point at an error page or an HTML response.

Streamed bodies break naive scrubbers

A recorded streaming completion is not a JSON document. It is a sequence of server-sent event frames, each a data: line containing its own small JSON object, terminated by a sentinel. The scrubber above will hit the json.loads failure path and fall back to plain text, which mostly works for regexes but silently loses the ability to drop fields.

More importantly, tokenisation defeats regexes across frame boundaries. An email address arrives as several tokens in several frames, so no single frame contains a string the email pattern matches, and the plain-text pass finds nothing at all in a body that plainly contains an email when reassembled. Handle streamed bodies by reassembling first:

SSE_DATA = re.compile(r"^data:\s*(.*)$", re.MULTILINE)


def scrub_sse_body(text: str) -> str:
    """Reassemble the streamed text, scrub it, and drop the original deltas.

    Frame-by-frame scrubbing cannot work: an email address is split across
    several token deltas, so no individual frame matches the pattern.
    """
    pieces = []
    for match in SSE_DATA.finditer(text):
        payload = match.group(1).strip()
        if payload in ("", "[DONE]"):
            continue
        try:
            chunk = json.loads(payload)
        except ValueError:
            continue
        for choice in chunk.get("choices", []):
            pieces.append(choice.get("delta", {}).get("content") or "")

    assembled = scrub_text("".join(pieces))

    # Re-emit as a single frame. Tests that assert on assembled content still
    # pass; tests that count frames must be rewritten, and should be, because
    # frame boundaries are not a stable property of a provider.
    return (
        "data: "
        + json.dumps({"choices": [{"delta": {"content": assembled}, "index": 0}]})
        + "\n\ndata: [DONE]\n\n"
    )

Collapsing to one frame changes what the cassette can test, and that trade is worth making explicit: you lose the ability to replay realistic chunk timing, and you gain a fixture that can be committed. If chunk-level behaviour is what you are testing, use a synthetic cassette written by hand rather than a recording of real traffic.

The pre-commit backstop

The record-time scrubber will be bypassed. Somebody will record with a different fixture, a colleague will copy a cassette in from a debugging session, a new endpoint will not go through the configured VCR instance. So the second layer assumes the first was misconfigured and checks the file on its way into the repository.

1. Write a checker that scans staged cassettes and exits non-zero on a hit. It must not fix anything. A hook that silently rewrites the file trains people to ignore it, and the point is to find out that the recording path is broken.

   #!/usr/bin/env python3
   """Fail if a cassette contains anything that looks like personal data."""
   import re
   import sys
   
   PATTERNS = {
       "email": re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),
       "phone": re.compile(r"\+?\d[\d ()-]{7,}\d"),
       "api key": re.compile(r"\b(sk|rk)-[A-Za-z0-9]{20,}\b"),
       "bearer": re.compile(r"(?i)bearer\s+[A-Za-z0-9._-]{20,}"),
   }
   ALLOWED = {"[email protected]", "+10000000000", "4111111111111111"}
   
   failures = 0
   for path in sys.argv[1:]:
       text = open(path, encoding="utf-8", errors="replace").read()
       for label, pattern in PATTERNS.items():
           for match in pattern.finditer(text):
               if match.group(0).strip() in ALLOWED:
                   continue
               print(path + ": possible " + label + ": " + match.group(0)[:40])
               failures += 1
   
   if failures:
       print("\n" + str(failures) + " finding(s). Re-record through llm_vcr, do not edit by hand.")
   sys.exit(1 if failures else 0)

2. Register it as a local hook scoped to the cassette directory, using the local-hook form documented by the pre-commit project. Scoping by files keeps it off the rest of the repository, where these patterns produce constant false positives.

   # .pre-commit-config.yaml
   repos:
     - repo: local
       hooks:
         - id: cassette-pii
           name: no personal data in recorded cassettes
           entry: tools/check_cassettes.py
           language: python
           files: ^tests/cassettes/.*\.ya?ml$
           pass_filenames: true

3. Run the same check in CI. Local hooks are skipped with one flag and are absent entirely on a fresh clone. The CI run is the one that is actually enforced, and it should scan the whole cassette directory rather than only the diff, so that a file added before the hook existed is still caught.
4. Add the allow-list deliberately. Redacted placeholders must be recognisable, or the hook fires on its own output. Use obviously fake values from reserved ranges — example.com is reserved for documentation, and the test card numbers published by payment processors are designed to be inert.

If it is already committed

Scrubbing a file in a new commit does not remove the data. The old blob is still in the repository, still in every clone, and still reachable by hash — and if the repository was ever public or mirrored, it is reachable by anyone who fetched it.

Treat it as a disclosure rather than a formatting problem. Rotate any credential that appeared, immediately, before touching the history — rewriting takes time and rotation does not. Rewrite the history with a tool built for it, such as git-filter-repo, force-push, and require every clone to be re-created, since a stale clone will happily reintroduce the old objects on the next push. Then follow whatever your incident process says about personal data, because at that point it is one.

The cheaper position is not to record from real traffic at all where you can avoid it. A fixture built from synthetic inputs carries none of this risk and tests the same code paths for most assertions; see testing without the model for how far that gets you.