Sample production LLM traffic without missing the failures that matter

The production dashboard is green, then support finds a prompt-injection path that none of your sampled traces contain. Your sampler kept plenty of ordinary chats and almost none of the tool-using requests where the defect lived. The model did not suddenly become safer; the evaluation set was looking in the wrong place.

Why a cheap sample can create an expensive blind spot

A production evaluator never sees "traffic" in the abstract. It sees whatever survived a chain of decisions: whether the event was eligible, which unit was sampled, how that unit was classified, whether it passed a selection rule, whether the payload was redacted, whether storage succeeded, and whether an eval worker could read it. A dashboard at the end of that chain cannot tell you which earlier gate removed a case unless every gate leaves evidence.

The first decision is the sampling unit. One HTTP request may contain a complete question and answer. A tool-using agent may span several model calls, tool calls, retries, and a final response. A multi-turn support case may require several requests to judge whether the assistant remembered a constraint. Sampling individual spans from either workflow creates fragments that are cheap to store and impossible to evaluate fairly.

Use the smallest key that still preserves the behavior under test. That may be a distributed trace ID for one request, an application-generated run ID for an agent execution, or a server-issued conversation ID for a multi-turn test. Document the choice. If one conversation can produce a hundred model calls, conversation-level sampling has a much larger storage cost than call-level sampling. That is a real budget decision, not an implementation detail.

The W3C Trace Context specification defines the trace-id carried by traceparent as the identifier for a distributed trace. It also says randomly generated trace IDs are preferred and discusses sampling based on that field. That makes a valid, server-trusted trace ID a useful sampling key for request-scoped evaluation. It does not make every incoming trace ID safe to trust. A public caller can send tracing headers, so validate them and apply your trust-boundary rules before using them to control expensive collection.

The second decision is eligibility. Keep it ahead of every quality or incident rule. A trace that violates consent, regional storage, tenant isolation, or retention policy does not become collectable because the model failed spectacularly. The sampler should be able to log an exclusion reason using non-content metadata without copying the prohibited prompt into an audit stream.

The third decision is classification. A single global probability treats a routine greeting and a tool authorization failure as interchangeable draws. They are not interchangeable to a QA team. Useful strata usually follow product risk and execution shape, such as routine chat, retrieval, tool use, safety block, cancellation, or a known incident. Keep the list small enough that each name has an owner and an exact rule. A stratum named interesting will become a bucket for whatever the current engineer wants to retain.

Selection comes next. A fresh draw from a process-local generator produces a different cohort when an ingestion job is replayed. That makes a model comparison noisy for an avoidable reason. A keyed pseudorandom function can map each unique sampling key to a stable bucket while keeping the assignment unpredictable to callers. With the same bucket namespace and secret, raising a threshold creates a nested cohort: the old selected keys remain selected and new keys are added.

The secret is part of the sampling design, not an authorization control. Generate it from a cryptographically strong source before the outcomes it will sample, keep it in a secrets manager, and record only its version on decision rows. Python's official hmac documentation defines the keyed interface used below. A secret HMAC assignment supports a probability interpretation under explicit assumptions: sampling keys are unique per unit, the secret was generated independently of outcomes, and no later unrecorded stage drops selected rows. Rejection sampling removes modulo bias when mapping 64-bit words into 10,000 buckets.

Finally, selection is not persistence. A trace can be eligible and selected, then disappear because redaction fails, a queue is full, a serializer rejects a tool result, or object storage denies a write. Record these as separate states. selected=true means the policy wanted the case. It does not prove an eval row exists.

For each candidate, the minimum useful decision record is compact:

None of those fields requires the raw prompt. That separation lets the sampling control plane remain auditable even when the content plane has shorter retention or stricter access. If a quota, reservoir, or priority cap runs later, it needs its own decision and conditional probability. The final inclusion probability is not the first-stage value after a second selector intervenes.

Make every inclusion decision reproducible

The implementation below uses basis points, where 10,000 means every eligible case, to avoid fuzzy rate strings and floating-point boundaries. Its values are illustrative policy choices, not production measurements. The important behavior is the ordering: governance eligibility first, forced incident rules second, then a stable bucket compared with the rate for one explicit stratum.

from __future__ import annotations

from dataclasses import dataclass
from hashlib import sha256
from hmac import new as hmac_new
from json import dumps
from types import MappingProxyType
from typing import Literal, Mapping

Stratum = Literal[
    "routine",
    "retrieval",
    "tool_use",
    "safety_block",
    "incomplete_or_error",
]

REQUIRED_STRATA: tuple[Stratum, ...] = (
    "routine",
    "retrieval",
    "tool_use",
    "safety_block",
    "incomplete_or_error",
)

BUCKET_COUNT = 10_000
WORD_SPACE = 1 << 64
ACCEPT_LIMIT = WORD_SPACE - (WORD_SPACE % BUCKET_COUNT)


@dataclass(frozen=True)
class Candidate:
    sampling_key: str
    route: str
    used_retrieval: bool
    used_tool: bool
    safety_blocked: bool
    terminal_state: Literal["completed", "cancelled", "error"]
    failure_kind: str | None
    known_incident: bool
    privacy_eligible: bool


@dataclass(frozen=True)
class SamplingPolicy:
    version: str
    bucket_namespace: str
    bucket_key_version: str
    rates_bps: Mapping[Stratum, int]
    force_failure_kinds: frozenset[str]

    def __post_init__(self) -> None:
        if not self.bucket_namespace:
            raise ValueError("bucket_namespace must not be empty")
        if not self.bucket_key_version:
            raise ValueError("bucket_key_version must not be empty")
        configured = set(self.rates_bps)
        missing = set(REQUIRED_STRATA) - configured
        unknown = configured - set(REQUIRED_STRATA)
        if missing:
            raise ValueError(f"missing sampling rates for: {sorted(missing)}")
        if unknown:
            raise ValueError(f"unknown sampling strata: {sorted(unknown)}")
        for stratum, rate in self.rates_bps.items():
            if isinstance(rate, bool) or not isinstance(rate, int):
                raise TypeError(f"rate for {stratum} must be an integer")
            if rate < 0 or rate > BUCKET_COUNT:
                raise ValueError(f"invalid rate for {stratum}: {rate}")
        object.__setattr__(self, "rates_bps", MappingProxyType(dict(self.rates_bps)))


@dataclass(frozen=True)
class Decision:
    sampling_key: str
    policy_version: str
    bucket_namespace: str
    bucket_key_version: str
    eligible: bool
    stratum: Stratum
    bucket: int | None
    configured_rate_bps: int
    policy_inclusion_probability_bps: int
    selected: bool
    reason: str


def classify(candidate: Candidate) -> Stratum:
    if candidate.terminal_state != "completed":
        return "incomplete_or_error"
    if candidate.safety_blocked:
        return "safety_block"
    if candidate.used_tool:
        return "tool_use"
    if candidate.used_retrieval:
        return "retrieval"
    return "routine"


def stable_bucket(
    bucket_secret: bytes,
    namespace: str,
    sampling_key: str,
) -> int:
    if len(bucket_secret) < 32:
        raise ValueError("bucket_secret must contain at least 32 bytes")

    counter = 0
    while True:
        payload = dumps(
            [namespace, sampling_key, counter],
            ensure_ascii=False,
            separators=(",", ":"),
        ).encode("utf-8")
        digest = hmac_new(bucket_secret, payload, sha256).digest()
        word = int.from_bytes(digest[:8], "big")
        if word < ACCEPT_LIMIT:
            return word % BUCKET_COUNT
        counter += 1


def decide(
    candidate: Candidate,
    policy: SamplingPolicy,
    bucket_secret: bytes,
) -> Decision:
    if not candidate.sampling_key:
        raise ValueError("sampling_key must not be empty")
    if len(bucket_secret) < 32:
        raise ValueError("bucket_secret must contain at least 32 bytes")

    stratum = classify(candidate)
    rate_bps = policy.rates_bps[stratum]

    if not candidate.privacy_eligible:
        return Decision(
            candidate.sampling_key,
            policy.version,
            policy.bucket_namespace,
            policy.bucket_key_version,
            False,
            stratum,
            None,
            rate_bps,
            0,
            False,
            "privacy_ineligible",
        )

    if candidate.known_incident:
        return Decision(
            candidate.sampling_key,
            policy.version,
            policy.bucket_namespace,
            policy.bucket_key_version,
            True,
            stratum,
            None,
            rate_bps,
            BUCKET_COUNT,
            True,
            "known_incident",
        )

    if candidate.failure_kind in policy.force_failure_kinds:
        return Decision(
            candidate.sampling_key,
            policy.version,
            policy.bucket_namespace,
            policy.bucket_key_version,
            True,
            stratum,
            None,
            rate_bps,
            BUCKET_COUNT,
            True,
            f"forced_failure:{candidate.failure_kind}",
        )

    bucket = stable_bucket(
        bucket_secret,
        policy.bucket_namespace,
        candidate.sampling_key,
    )
    return Decision(
        candidate.sampling_key,
        policy.version,
        policy.bucket_namespace,
        policy.bucket_key_version,
        True,
        stratum,
        bucket,
        rate_bps,
        rate_bps,
        bucket < rate_bps,
        "bucket_selected" if bucket < rate_bps else "bucket_rejected",
    )


POLICY = SamplingPolicy(
    version="2026-08-04.1",
    bucket_namespace="llm-eval-trace-v1",
    bucket_key_version="sampling-hmac-key-v1",
    rates_bps={
        "routine": 100,
        "retrieval": 500,
        "tool_use": 1_000,
        "safety_block": 5_000,
        "incomplete_or_error": 10_000,
    },
    force_failure_kinds=frozenset({"tool_denied", "invalid_tool_arguments"}),
)

This is application code, not a hidden feature of an eval vendor. The classifier needs tests because a change in its ordering can move traffic between strata without changing a single configured rate. For example, a cancelled tool call matches both used_tool and an incomplete terminal state. The code deliberately classifies terminal failure first so the investigation does not lose the fact that no complete answer exists.

The bucket namespace and key version are separate from the policy version on purpose. Changing routine from 100 to 200 basis points under the same namespace and secret keeps the original routine cohort and adds keys whose buckets fall between 100 and 199. Changing the algorithm, key format, or sampling unit should use a new namespace. Rotating the secret should use a new key version. Either change starts a new cohort. A rate or classification change still needs a new policy version so historical decisions remain explainable.

Exercise invariants rather than trusting one lucky key. The following tests check repeatability, bucket range and diversity, integer rate validation, the privacy precedence rule, forced incident handling, and the nested-cohort property. They use Python's standard unittest module and run without a network call. The diversity assertion is a compatibility guard against a collapsed bucket function, not a statistical certification of HMAC.

import unittest
from dataclasses import replace

from sampler import (
    BUCKET_COUNT,
    Candidate,
    REQUIRED_STRATA,
    SamplingPolicy,
    decide,
    stable_bucket,
)


TEST_BUCKET_SECRET = bytes(range(32))


def policy(rate_bps: int) -> SamplingPolicy:
    return SamplingPolicy(
        version=f"test-{rate_bps}",
        bucket_namespace="unit-test-v1",
        bucket_key_version="unit-test-key-v1",
        rates_bps={name: rate_bps for name in REQUIRED_STRATA},
        force_failure_kinds=frozenset({"tool_denied"}),
    )


BASE = Candidate(
    sampling_key="server-trace-0001",
    route="support-chat",
    used_retrieval=False,
    used_tool=False,
    safety_blocked=False,
    terminal_state="completed",
    failure_kind=None,
    known_incident=False,
    privacy_eligible=True,
)


class SamplerTests(unittest.TestCase):
    def test_same_key_and_namespace_repeat_the_decision(self) -> None:
        first = decide(BASE, policy(750), TEST_BUCKET_SECRET)
        second = decide(BASE, policy(750), TEST_BUCKET_SECRET)
        self.assertEqual(first.bucket, second.bucket)
        self.assertEqual(first.selected, second.selected)

    def test_bucket_function_uses_the_domain(self) -> None:
        buckets = {
            stable_bucket(
                TEST_BUCKET_SECRET,
                "unit-test-v1",
                f"server-trace-{index:04d}",
            )
            for index in range(2_000)
        }
        self.assertTrue(all(0 <= bucket < BUCKET_COUNT for bucket in buckets))
        self.assertGreater(len(buckets), 1_700)

    def test_boolean_rate_is_rejected(self) -> None:
        invalid_rates = {name: 100 for name in REQUIRED_STRATA}
        invalid_rates["routine"] = True
        with self.assertRaises(TypeError):
            SamplingPolicy(
                version="invalid-rate",
                bucket_namespace="unit-test-v1",
                bucket_key_version="unit-test-key-v1",
                rates_bps=invalid_rates,
                force_failure_kinds=frozenset(),
            )

    def test_privacy_exclusion_wins_over_incident_priority(self) -> None:
        candidate = replace(
            BASE,
            privacy_eligible=False,
            known_incident=True,
        )
        decision = decide(candidate, policy(10_000), TEST_BUCKET_SECRET)
        self.assertFalse(decision.selected)
        self.assertEqual(decision.reason, "privacy_ineligible")

    def test_configured_failure_is_force_included(self) -> None:
        candidate = replace(BASE, failure_kind="tool_denied")
        decision = decide(candidate, policy(0), TEST_BUCKET_SECRET)
        self.assertTrue(decision.selected)
        self.assertEqual(decision.reason, "forced_failure:tool_denied")

    def test_higher_rate_never_removes_an_existing_key(self) -> None:
        for index in range(500):
            candidate = replace(BASE, sampling_key=f"server-trace-{index:04d}")
            low = decide(candidate, policy(200), TEST_BUCKET_SECRET)
            high = decide(candidate, policy(800), TEST_BUCKET_SECRET)
            with self.subTest(sampling_key=candidate.sampling_key):
                if low.selected:
                    self.assertTrue(high.selected)


if __name__ == "__main__":
    unittest.main()

These tests will not tell you whether 200 or 800 basis points is the right business rate. They tell you the sampler honors the policy you chose. Rate selection belongs in a capacity and coverage review using your own traffic mix, eval cost, incident history, and minimum useful cases per stratum.

Work through three failures before choosing rates

A rare tool failure never reaches the evaluator. Consider a support assistant that answers routine account questions and can also call an internal refund tool. Most requests never invoke the tool. A flat sample can therefore look healthy while the tool path has little or no coverage.

The first clue is not a low grader score. It is an incident trace whose decision record says stratum=tool_use, selected=false, and reason=bucket_rejected, combined with a coverage report showing few selected tool cases. If the incident was classified as routine, the rate is not the immediate bug. The classifier failed to recognize tool use. Raising the global rate would spend more on routine chats while leaving the bad classification untouched.

Fix the taxonomy before the percentage. Give actual tool executions their own stratum, distinguish authorization failures from valid tool completions if they have different owners, and force-include narrowly defined failure kinds after the privacy gate. The cost is dataset skew. If every eligible tool_denied case is retained while routine answers are sampled lightly, the raw dataset failure rate will be much higher than production prevalence. That is acceptable for defect discovery. It is not acceptable to present the raw rate as an estimate of customer experience.

Record policy_inclusion_probability_bps with every first-stage decision. An eligible forced case has probability one at that stage. A regular tool-use case has the threshold probability for its stratum, provided the sampling key and HMAC assumptions described above hold. Analysts can then separate a risk set used to find defects from a probability sample used to estimate aggregate quality. If a later quota or cap runs, this first-stage value is not the final inclusion probability.

A replay produces a different set of conversations. A team samples during an ingestion job with a fresh random draw. Monday's prompt version and Tuesday's prompt version are evaluated on whatever each run happens to keep. When the score changes, nobody can tell how much came from the prompt and how much came from the cohort.

Look for missing bucket values and for selection timestamps that change when the same trace is processed again. If the decision record contains only selected=true, there is no way to prove replay stability. A stable keyed bucket fixes that operational problem. Run both variants against the same stored sampling keys, keep the bucket namespace and secret version fixed, and attach each generated answer to its prompt, model, tool, and grader versions.

There is a subtle near-miss here. Stable sampling does not make the model or an LLM grader deterministic. It removes cohort churn from the comparison. Outputs can still vary, tools can return newer data, and model-based judgments can disagree. Freeze external fixtures where the test permits it, preserve actual outputs, and report repeated judgments as repeated judgments rather than pretending the hash controls generation.

The trade-off is persistence. To compare later variants on the same cases, you need a retained, permitted representation of the input and any required context. A stable ID without the corresponding evidence is only a pointer to something you can no longer replay. If policy forbids retaining that evidence, create a redacted incident reproduction or a synthetic counterpart instead of quietly assuming the production case remains available.

A selected streaming trace is missing its ending. The request starts, the sampler marks it for collection, several tokens arrive, and the client disconnects during a tool call. The stored row contains a prompt and a partial assistant message, but the evaluator treats the empty final field as a bad answer. The score is technically faithful to the row and misleading about the cause.

The evidence is in lifecycle fields. The terminal state is cancelled, the response-finished event is absent, or a tool call has no matching result. That is not the same as a completed response that failed a correctness rubric. Put incomplete or transport-error executions in their own stratum and give graders an explicit precondition: only grade answer quality when the run contains the artifacts the rubric needs.

A practical collector retains a small metadata envelope until the run reaches a terminal state. It can then classify the completed shape, make or finalize the sampling decision, redact the allowed content, and persist the row. For large streams, buffering full content until that point may be too expensive. Another design makes an early provisional decision, writes chunks only for provisional selections, and emits a final status that tells the dataset builder whether the row is complete. Either design needs a cleanup path for abandoned buffers.

Delayed classification costs latency and temporary state. Provisional selection can still miss late-arriving failure signals unless the collector has a separate incident path. There is no zero-cost choice. State which late signals matter, how long the envelope may live, and whether an incomplete case is useful as reliability evidence even when it is not gradeable as answer quality.

These three examples also show why route-only strata are weak. Risk can depend on a terminal state, a tool result, a safety decision, or an incident label that arrives after the first model call. Classify at the latest point needed to see the relevant signal, while retaining no more content than the privacy design permits.

Prove the sampler is failing, not the model

Start the investigation before the evaluator. You want counts and joins for candidates, decisions, and stored rows, split by policy version and stratum. A grader score cannot explain a candidate that never received a decision or a selected row that never reached storage.

Emit structured events from those three boundaries. A candidate event identifies a sampling unit's safe metadata and provisional stratum. A decision event records the final stratum and exact policy result. A storage event confirms the eval row was committed. Put policy_version on all three events and join on (sampling_key, policy_version). Keep the raw prompt out of this diagnostic stream.

The following audit reads newline-delimited JSON and fails when candidates have no decisions, decisions select rows that never reach storage, or storage contains a row the policy did not select. It also prints the funnel by stratum so a route with zero selected cases is visible instead of being averaged into an overall total.

from __future__ import annotations

import argparse
import json
from collections import Counter
from pathlib import Path
from typing import Any


def read_events(path: Path) -> list[dict[str, Any]]:
    events: list[dict[str, Any]] = []
    with path.open(encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            try:
                event = json.loads(line)
            except json.JSONDecodeError as error:
                raise ValueError(f"{path}:{line_number}: {error.msg}") from error
            if event.get("event_type") not in {"candidate", "decision", "stored"}:
                raise ValueError(
                    f"{path}:{line_number}: unknown event_type={event.get('event_type')!r}"
                )
            events.append(event)
    return events


UnitKey = tuple[str, str]


def unit_key(event: dict[str, Any]) -> UnitKey:
    sampling_key = event.get("sampling_key")
    policy_version = event.get("policy_version")
    if not isinstance(sampling_key, str) or not sampling_key:
        raise ValueError("every event needs a non-empty sampling_key")
    if not isinstance(policy_version, str) or not policy_version:
        raise ValueError("every event needs a non-empty policy_version")
    return sampling_key, policy_version


def label(key: UnitKey) -> str:
    return f"{key[0]}@{key[1]}"


def audit(events: list[dict[str, Any]]) -> list[str]:
    candidates: dict[UnitKey, dict[str, Any]] = {}
    decisions: dict[UnitKey, dict[str, Any]] = {}
    stored: set[UnitKey] = set()
    duplicate_events: Counter[tuple[str, UnitKey]] = Counter()

    for event in events:
        event_type = str(event["event_type"])
        key = unit_key(event)
        duplicate_events[(event_type, key)] += 1
        if event_type == "candidate":
            candidates[key] = event
        elif event_type == "decision":
            stratum = event.get("stratum")
            if not isinstance(stratum, str) or not stratum:
                raise ValueError(f"decision {label(key)} needs a final stratum")
            decisions[key] = event
        else:
            stored.add(key)

    problems: list[str] = []
    for (event_type, key), count in duplicate_events.items():
        if count > 1:
            problems.append(
                f"duplicate {event_type} events for {label(key)}: {count}"
            )

    missing_decisions = sorted(set(candidates) - set(decisions))
    if missing_decisions:
        problems.append(
            "candidates without decisions: "
            + ", ".join(label(key) for key in missing_decisions[:10])
        )

    orphan_decisions = sorted(set(decisions) - set(candidates))
    if orphan_decisions:
        problems.append(
            "decisions without candidates: "
            + ", ".join(label(key) for key in orphan_decisions[:10])
        )

    selected = {
        key for key, decision in decisions.items() if decision.get("selected") is True
    }
    missing_storage = sorted(selected - stored)
    if missing_storage:
        problems.append(
            "selected rows not stored: "
            + ", ".join(label(key) for key in missing_storage[:10])
        )

    unexpected_storage = sorted(stored - selected)
    if unexpected_storage:
        problems.append(
            "stored rows without a selecting decision: "
            + ", ".join(label(key) for key in unexpected_storage[:10])
        )

    strata = sorted({str(event["stratum"]) for event in decisions.values()})
    print(f"{'stratum':24} {'candidates':>10} {'decisions':>10} {'selected':>10} {'stored':>10}")
    for stratum in strata:
        keys = {
            key
            for key, event in decisions.items()
            if event["stratum"] == stratum
        }
        candidate_keys = keys & set(candidates)
        selected_keys = keys & selected
        stored_keys = selected_keys & stored
        print(
            f"{stratum:24} {len(candidate_keys):10d} {len(keys):10d} "
            f"{len(selected_keys):10d} {len(stored_keys):10d}"
        )

    for key in sorted(set(candidates) & set(decisions)):
        provisional = candidates[key].get("stratum")
        final = decisions[key]["stratum"]
        if provisional and provisional != final:
            print(
                f"RECLASSIFIED: {label(key)}: "
                f"{provisional} -> {final}"
            )

    return problems


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--events", type=Path, required=True)
    args = parser.parse_args()

    problems = audit(read_events(args.events))
    for problem in problems:
        print(f"ERROR: {problem}")
    return 1 if problems else 0


if __name__ == "__main__":
    raise SystemExit(main())

For an illustrative hand-built fixture, not a production measurement, a broken storage leg might print this:

$ python tools/audit_sampling.py --events tests/fixtures/sampling-events-broken.ndjson
stratum                 candidates  decisions   selected     stored
retrieval                       12         12          2          2
routine                         40         40          1          1
tool_use                         8          8          4          3
ERROR: selected rows not stored: server-trace-0047@2026-08-04.1

That error points at collection or persistence, not model quality. If the candidate count for tool_use is zero while application metrics show tool executions, inspect instrumentation or classification. If candidates and decisions exist but selected rows are missing, inspect the queue, redactor, serializer, and storage acknowledgement. If stored rows exist but the eval worker reads fewer, inspect dataset ingestion and query filters.

A traffic-mix change creates a different near-miss. Suppose the overall pass rate falls after a feature sends more requests through retrieval. Check per-stratum results before declaring a regression. If routine and retrieval quality are stable within their own cohorts but retrieval now makes up more of the sample, the aggregate changed because composition changed. Report both the production candidate distribution and the sampled distribution. Do not silently compare two aggregates with different weights.

Grader drift can look identical on the dashboard. Preserve the exact stored outputs, rubric identifier, grader type, grader prompt or code version, and any relevant model identifier used by a model-based grader. The OpenAI grader documentation is one concrete example of multiple grader types with different behavior. Regardless of vendor, rerun the unchanged stored rows under the old and new grader definitions. Label changes with unchanged outputs are grader evidence, not sampling evidence.

A model or prompt regression requires another controlled comparison. Keep the cohort and grader definition fixed, generate or replay outputs under the old and candidate application versions, and compare at row level. If external tools or retrieval data can change, capture their versions or fixture their responses where the test goal permits. Otherwise the comparison includes a data-source change whether the report admits it or not.

One report should therefore expose four independent funnels:

1. Production units by route and stratum, before probability selection.
2. Eligible units and exclusion reasons, without prohibited content.
3. Selected versus stored units by policy version.
4. Gradeable, graded, and disputed rows by grader version.

An overall score belongs after those checks. Before them, it is a number with an unknown denominator.

Roll out a new policy without breaking the trend line

Replacing a flat sampler in one deployment creates an artificial discontinuity. The cohort changes, the stratum mix changes, and forced incidents can lift the apparent failure rate even if the application has not changed. Treat the migration like an instrumentation change.

First, freeze the old definitions. Save the old sampler's effective rate, sampling unit, eligibility rules, known exclusions, and the fields actually present in stored rows. If the old job used an unseeded random draw, say so. You cannot reconstruct membership that was never logged.

Next, run the new classifier in shadow mode. It should emit decision metadata but should not copy extra production payloads. Review the candidate counts, unknown classifications, privacy exclusions, and the policy's expected storage demand. Shadow mode is also where you discover fields that arrive too late for the current collector.

Then evaluate a bounded slice of allowed traffic under both policies. Emit a separate candidate, decision, and storage chain for each policy version while reusing the same server-issued sampling key. Treat (sampling_key, policy_version) as the event key so an old-policy decision can never satisfy the audit for a missing new-policy decision. Store content only under the collection authority already approved for the rollout. The overlap shows whether score movement comes from rows both policies share or from the new cohort. It also reveals a policy that accidentally drops a whole route.

Promote one route or tenant class at a time when your architecture permits it. Keep policy version, bucket namespace, and classifier version on every decision. Do not rename a stratum in place. A rename may look cosmetic in code while changing dashboards, alert filters, and historical joins.

CI should prove the implementation's invariants and the diagnostic fixture, not pretend a small fixture validates production rates. This job follows GitHub's documented workflow syntax and runs the scripts shown above. In a repository with a different CI provider, keep the two commands and translate the surrounding job syntax.

name: production-sampling-contract

on:
  pull_request:
    paths:
      - "sampler.py"
      - "tests/test_sampler.py"
      - "tools/audit_sampling.py"
      - "tests/fixtures/sampling-events-*.ndjson"
      - ".github/workflows/production-sampling-contract.yml"
  workflow_dispatch:

jobs:
  sampling-contract:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v6
      - uses: actions/setup-python@v6
        with:
          python-version: "3.12"
      - name: Check deterministic sampling rules
        run: python -m unittest discover -s tests -p "test_sampler.py" -v
      - name: Check the candidate-to-storage funnel
        run: >-
          python tools/audit_sampling.py
          --events tests/fixtures/sampling-events-valid.ndjson

The valid fixture used by the workflow should include a privacy-ineligible known incident, a forced tool failure, a regular rejection, a selected and stored row, and an incomplete execution. Keep sampling-events-broken.ndjson as a separate test input that proves the audit exits nonzero. The happy-path CI command must never conceal an expected failure with shell logic.

After promotion, preserve an overlap window defined by your approved retention policy. Compare shared rows, old-only rows, and new-only rows. A trend chart should mark the policy change and avoid joining the pre-migration aggregate to the post-migration aggregate as if the denominator stayed constant.

Rate changes are the easiest migration when the bucket namespace and key remain stable. Classification changes are harder because a trace may move from routine to tool_use and receive a different threshold. Sampling-unit changes are hardest because call-level and conversation-level rows are not one-to-one. Start a new series for that change instead of manufacturing continuity.

Operational rollback also needs a version, not an edit to yesterday's config. If storage cost spikes, activate a lower-rate policy version and leave prior decision records intact. If one force-include rule floods the dataset, narrow the definition based on the recorded failure kind. Never delete the reason from historical rows to make the new rule look consistent.

Know the costs and when not to use this design

Risk-aware sampling spends evaluation budget where defects are more likely to matter, but it does not create free coverage. The candidate audit adds a metadata write for traffic that may never be selected. Late classification holds state longer. Forced incidents consume storage and grader calls during outages, exactly when systems are already under pressure. Grouping by conversation can turn one selected key into a large payload.

Put explicit caps after prioritization, then treat the cap as a second selector. Record its decision, the capacity window, and any conditional inclusion probability. If cases within a stratum are randomly chosen at that stage, the final probability is the first-stage probability multiplied by the conditional cap probability. If deterministic priority eviction decides which rows survive, set the final probability to null and exclude those rows from prevalence estimates. A forced case still has first-stage probability one, but it does not have final probability one after an unbounded incident stream hits a cap. Record priority_dropped_due_to_cap with safe metadata and alert on it. Do not claim the retained incidents are complete when the cap fired.

More strata are not automatically better. Each split needs enough eligible cases to inspect, a reason for a different rate, and an owner who can act on failures. A taxonomy with dozens of sparse categories produces unstable charts and complicated weighting. Merge categories that share risk, evaluation method, and response owner. Split only when the distinction changes a decision.

Stable keyed bucketing has its own trap. It repeatedly selects the same keys when the production population is replayed. That is useful for comparisons and bad if you expect periodic rotation to discover new examples from an unchanged key set. Choose deliberately between a fixed comparison cohort and a rotating discovery cohort. A rotating cohort can add a time window to the bucket namespace, but each window then has a different membership and must be labeled as such.

Do not use raw production payload sampling when policy, contract, or regulation prohibits it. A redaction promise written after collection does not authorize the initial copy. Use synthetic cases, opt-in traffic, approved incident reproductions, or on-system aggregation that exports only permitted features. Quality coverage never overrides data governance.

Do not replace a curated release suite with sampled traffic. A fixed suite protects named requirements and known regressions on every release. Production samples discover language, workflows, and failures the curated suite did not anticipate. They solve different problems. Promote valuable production cases into a reviewed regression dataset after redaction, labeling, and ownership rather than depending on chance to select them again.

Do not use a risk-skewed set to estimate prevalence from raw counts. If the question is "What fraction of all eligible conversations fails this rubric?", you need a probability design that supports that estimate, recorded inclusion probabilities, and an analysis that accounts for the design. Forced incidents can remain in a separate census stream. Inverse-probability weighting is not a magic repair when a stratum has zero inclusion probability or too few observations.

Do not overload the W3C sampled trace flag as your eval-policy label. The specification describes it as tracing recording context and notes that downstream components may make their own decisions. Your eval selector has different eligibility, privacy, and grading semantics. Keep its decision in an application-owned record linked by the trace ID.

Do not grade incomplete executions as bad answers unless the requirement is specifically about completion reliability. A cancelled stream may be a client action, a network failure, a timeout, or a server defect. Preserve the terminal evidence, route it to the right reliability metric, and exclude it from rubrics that require a finished answer.

Finally, do not tune rates from one quiet week and forget them. Review unknown strata, cap activations, selected-to-stored loss, incident capture, per-stratum grader coverage, and cost whenever routes or agent behavior change. OpenAI's dataset guidance recommends expanding evaluation data as edge cases and blind spots appear. The same discipline applies to a vendor-neutral pipeline: an incident should leave behind a reviewed test, not merely a temporary sampling exception.