Route Every New Model Release Through One Adapter Before It Touches Your Stack

Last month a teammate pasted a leaderboard screenshot into our chat and asked the question I now dread: "this one's cheaper and better — how hard is the swap?" Two years ago I would have said "an afternoon." Then an afternoon swap actually happened, and it took down our chat feature at 11pm. Not because the new model was worse — because its server-sent events carried a heartbeat frame our streaming parser had never seen, the parser choked, and every in-flight conversation died mid-sentence.

The model was fine. Our integration was the problem. Since then, every candidate model in our system enters through exactly one door, and it doesn't get past staging until it survives the same gauntlet our incumbent passes. This post is that gauntlet, written in FastAPI/Python this time, with a pytest suite you can lift directly.

The one-door rule

In our codebase there is a single module — chat/provider.py — where a model identifier is allowed to exist. Routes, the messages table, the billing writer, and the frontend's SSE consumer all depend on one protocol:

# chat/provider.py
from typing import AsyncIterator, Protocol

class Usage(Protocol):
    prompt_tokens: int
    completion_tokens: int

class ChatProvider(Protocol):
    name: str

    async def stream(
        self, messages: list[dict], *, max_tokens: int
    ) -> AsyncIterator[str]:
        """Yield text deltas. Raise ProviderHTTPError on non-2xx."""
        ...

    async def last_usage(self) -> Usage:
        """Token counts for the most recent stream. Drives billing writes."""
        ...

Most newly released open models expose an OpenAI-shaped HTTP API, which means the candidate is one adapter and one config entry away — no route changes, no schema changes:

# chat/openai_shaped.py
import httpx

class OpenAIShapedProvider:
    def __init__(self, name: str, base_url: str, api_key: str, model: str):
        self.name = name
        self._base, self._key, self._model = base_url, api_key, model
        self._usage: Usage | None = None

    async def stream(self, messages, *, max_tokens):
        payload = {
            "model": self._model,
            "messages": messages,
            "max_tokens": max_tokens,
            "stream": True,
            "stream_options": {"include_usage": True},
        }
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream(
                "POST",
                f"{self._base}/chat/completions",
                headers={"Authorization": f"Bearer {self._key}"},
                json=payload,
            ) as resp:
                if resp.status_code != 200:
                    body = (await resp.aread()).decode(errors="replace")
                    raise ProviderHTTPError(resp.status_code, body)
                async for delta in iter_text_deltas(resp):
                    yield delta

    async def last_usage(self):
        assert self._usage is not None, "stream finished without usage frame"
        return self._usage

The interesting part is iter_text_deltas, because this is where candidates actually differ: some send comment lines as keep-alives, some emit a final usage-only chunk with empty choices, some close the connection without a [DONE] marker. That parser is shared, and the gauntlet exists to prove a candidate can live inside it.

The gauntlet: five seam tests, zero opinion questions

Notice what's absent: nothing here asks whether the model writes good poetry. Quality evals come later, on domain prompts, after the integration question is settled. These five tests only ask whether the candidate can survive our plumbing.

# tests/evals/test_provider_seams.py
# run: pytest tests/evals -k seam --provider=candidate
import pytest

pytestmark = pytest.mark.asyncio

async def drain(p, messages, max_tokens=64):
    return "".join([d async for d in p.stream(messages, max_tokens=max_tokens)])

async def test_seam_stream_delivers_text(candidate):
    out = await drain(candidate, [{"role": "user", "content": "Reply with the word ready"}])
    assert out.strip(), "stream completed with zero deltas"

async def test_seam_usage_feeds_billing(candidate):
    await drain(candidate, [{"role": "user", "content": "Name three colors"}])
    usage = await candidate.last_usage()
    assert usage.prompt_tokens > 0 and usage.completion_tokens > 0

async def test_seam_replays_longest_real_thread(candidate, longest_thread_fixture):
    # Pulled from a production messages export (anonymized), not generated.
    await drain(candidate, longest_thread_fixture, max_tokens=128)

async def test_seam_context_overflow_is_a_clean_4xx(candidate, oversized_thread_fixture):
    with pytest.raises(ProviderHTTPError) as err:
        await drain(candidate, oversized_thread_fixture)
    assert 400 <= err.value.status < 500

async def test_seam_client_cancel_midstream_leaves_no_leak(candidate):
    agen = candidate.stream(
        [{"role": "user", "content": "Tell a long story"}], max_tokens=2048
    )
    async for delta in agen:
        break  # simulate the user closing the tab
    await agen.aclose()  # must not raise, must release the connection

Two fixtures do the heavy lifting. longest_thread_fixture is the nastiest conversation I could find in our own history export — the one that broke context windows twice before. oversized_thread_fixture is that same thread padded past the candidate's advertised window, and the test demands the failure arrive as a mapped 4xx, because our UI's retry logic branches on exactly that distinction. A 500 here would trigger blind retries and double our bill; a 413 or 400 triggers a graceful "this conversation is too long" state. This is the class of bug a playground session will never show you.

The result is a small matrix — incumbent column, candidate column, five rows of pass/fail — that I can paste back into the team chat as the actual answer to "how hard is the swap."

Running it without procurement getting involved

The unglamorous blocker in this workflow is never code — it's access. Spinning up a vendor account or reserving GPU time to answer a question that might take forty minutes is a hard sell, so evaluations quietly turn into playground sessions instead.

My current shortcut: Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode offers free access to models together with a free server option, which maps neatly onto this gauntlet — I register the endpoint as another provider config, run the pytest suite there, and haven't provisioned or paid for anything to get the matrix. What sold me beyond the price is that MonkeyCode is developed in the open-source spirit: the pieces are inspectable, so wiring it behind my own provider protocol took an hour, and if I ever outgrow it the same protocol is the exit ramp. That's the same property that makes open-weight releases valuable in the first place — evaluation without a permission slip, and internals you can actually read instead of trusting a black box.

If the candidate passes all five seams, then it earns the expensive part: a quality eval on real domain prompts and a canary slice of traffic.

What each approach actually tells you

Question	Playground session	Seam gauntlet
Can the model produce good text?	Roughly yes	Deliberately not asked
Does its SSE framing break your parser?	Never tested	Tested, incl. cancel mid-stream
Does usage data reach your billing writes?	No	Yes
How does it fail at the context limit?	Unknown	Must be a mapped 4xx
Cost of finding out	New vendor account or GPU time	Free model access + free server
Artifact you can paste into chat	A feeling	A pass/fail matrix

Where this breaks down

• Five seam tests are a gate, not a verdict. A candidate that passes has earned a quality evaluation, not a traffic cutover.
• A free tier answers "is this worth a deeper look" — it does not answer latency, throughput, or availability questions. Don't point production load at it, and don't quote its response times in a capacity plan.
• If your feature leans on one vendor's proprietary surface — a specific tool-calling dialect, a hosted retrieval API — the protocol will leak no matter how clean the adapter is. Either abstract the behavior you need or write the lock-in down as a conscious decision.
• No streaming, no billing, no persistence? Then there's no seam to protect, and the playground is genuinely sufficient.

Pre-budget checklist for any new model

• [ ] Exactly one module in the repo is allowed to name the model
• [ ] All five seam tests pass, including the mid-stream cancel
• [ ] Longest real conversation replays without a context error
• [ ] Context overflow surfaces as the 4xx your retry logic expects
• [ ] Usage counts reconcile against an actual billing-table write
• [ ] Rollback is a config value, not a code change

Release cycles will keep accelerating, and next quarter there will be another screenshot in the team chat. The teams that come out ahead aren't the ones with the fastest hot takes — they're the ones who can turn "how hard is the swap?" into a forty-minute evidence-backed answer.

When you've swapped providers, which seam gave way first — the stream parser, the usage accounting, or the error mapping? Drop the actual failure (status code, chunk shape, whatever you saw) in the comments; I'm curious whether the cancel-mid-stream case bites anyone else as often as it bites us.