Writing a runtime¶

A runtime is how Wind Tunnel talks to your agent platform. Implement two small Protocols — AgentRuntime and AgentHandle — and every scenario, perturbation, evaluator, and report in the framework works against your platform unchanged.

The contracts live in windtunnel/spi/agent_runtime.py. They are typing.Protocols: no base class to inherit, no framework registration — structural typing. If your class has the methods, it's a runtime.

For harness certification, a runtime may additionally implement the optional ReferenceCapableAgentRuntime protocol. It declares one inference-substitution seam for the runtime service so wt selftest can drive golden and poison model decisions through the otherwise live stack.

The contract at a glance¶

from windtunnel.spi.agent_runtime import AgentConfig, AgentHandle, Message, Response


class MyRuntime:
    def provision(self, config: AgentConfig, mcps: list | None = None) -> AgentHandle:
        """Stand up a live agent wired to the given (already-started) MCP handles."""


class MyHandle:
    def send(self, messages: list[Message], session_id: str) -> Response:
        """Drive one turn. Returns a flat {"content", "tool_calls"} dict
        (or an OpenAI-shaped chat completion — both accepted, see below)."""

    def reset_state(self) -> None:
        """Wipe cross-run state. Cheap, called before every run. Idempotent."""

    def teardown(self) -> None:
        """Release everything. Called once per batch. Idempotent, must not raise."""

A conformance suite (tests/test_runtime_conformance.py) runs the same scenarios through the in-memory reference runtime; point it at yours to verify the contract before debugging real failures.

`provision(config, mcps)` — once per batch¶

Called once before a batch of runs. Receives:

config: AgentConfig — the agent's identity and knobs:

Field	Meaning
`agent_id`, `variant_id`	identity; lands in every `Trace`
`system_prompt`	override the platform's default system prompt
`persona_doc`	path to an operating-notes doc (e.g. AGENTS.md)
`skills`	optional skill/persona `.md` files
`mcp_servers`	declarative MCP references (`MCPSpec(name, url)`)
`model`	`ModelSpec(name, quant)` — which model to run
`sampling`	`SamplingConfig(temperature, top_p, tool_choice, max_tokens)`

Your runtime interprets these for your platform. Unsupported fields should raise loudly, not silently no-op — a bench that silently ignores temperature=0 produces lies, not results.

mcps — already-started MCPHandle objects (the runner starts them before calling you). Your job is to make their handle.url reachable from wherever your agent runs and register their tools however your platform mounts tools. Two recurring gotchas from the reference drivers:
Network translation: the mock runs on the host; your agent runs in a container. localhost in handle.url must become host.docker.internal (or your network's equivalent) before registration.
Tool-mount timing: platforms that read tool registrations at startup need a restart (or explicit re-sync) after registration — and anything holding a session against the restarted service may need its own restart. Gate on the tools actually being callable (probe with a real request), not on the service being up.

Return an AgentHandle. Provisioning is the expensive step — containers, auth, registration — which is exactly why it's separated from reset_state().

Optional `provision_reference()` — certify the harness¶

from windtunnel.spi import AgentConfig, ReferenceCase

class MyRuntime:
    def provision_reference(self, config: AgentConfig, case: ReferenceCase, mcps=None):
        self.inference_substitute.install(case.decisions)
        return self._provision_normal_agent(config, mcps)

Implement this only if the runtime can substitute decisions at its genuine model-inference seam while leaving its normal agent loop, tools, fixtures, probes, and evidence path live. The seam is runtime-wide; scenarios do not declare endpoints or service-specific configuration. Wind Tunnel calls the method once per reference case, and handle.teardown() must remove the substitution so cases cannot contaminate each other.

The built-in in_memory runtime is deliberately unsupported because it shortcuts the loop this capability is meant to certify. See reference self-tests for the complete contract, verdicts, and probe-wiring timing.

`send(messages, session_id)` — one turn¶

Drive one turn of the agent loop and return a response dict in either of two accepted shapes (the runner normalizes both):

flat message — the forward-looking minimal contract: {"content": "...", "tool_calls": [...]} (a {"message": {...}} wrapper is also tolerated). If you're writing a new runtime, return this.
OpenAI chat-completions — choices[0].message with role, content, tool_calls, finish_reason. Kept for OpenAI-compat so a runtime can hand back an upstream completions payload untouched.

The two rules that matter:

Surface intermediate tool calls. If your platform runs the whole agent loop server-side and only hands back the final message, the trajectory layer (must_call) silently fails for every tool-using scenario. Use whatever streaming/events API exposes the intermediate steps and reconstruct the full picture. (This single mistake cost the reference implementation a rewrite — it is the most common way to build a runtime that looks fine and scores garbage.)
Don't normalize tool-call shapes. Record tool_calls exactly as your platform emits them (OpenAI wire shape, flattened shape, whatever). The Trace preserves them faithfully; evaluators handle both. Normalizing at the runtime layer destroys evidence — drift in tool-call shape is itself a failure mode worth catching.

Multi-turn scenarios call send() repeatedly with the same session_id; your platform's session affinity must honor it.

Also worth capturing if your platform exposes it: the rendered prompt (the exact chat-template output per turn). Turn.rendered_prompt and its auto-computed hash let you diff what the model actually saw across runs — the fastest way to catch template regressions.

`reset_state()` vs `teardown()`¶

Two lifecycle levels, deliberately distinct:

reset_state() — between runs. Cheap and frequent. Wipe sessions, conversation history, memory stores, tool-call logs — anything that lets run N leak into run N+1. The runner calls it after provision() and before every run, including the first run of the first scenario; drivers may therefore rely on reset_state() as the seed point. Treat a failed wipe as fatal. Cross-run contamination is the classic source of false passes (the reference bench was once burned by a full-text-search index that survived the wipe and answered scenario N+1's question from scenario N's data — wipe the indexes too).
teardown() — end of batch. Expensive and rare. Stop containers, kill tunnels, revoke grants, drop registrations. Idempotent and non-raising: it runs in finally paths.

Auth and identity¶

The runner doesn't know about auth — credentials are the runtime's problem. The reference pattern: a dev-mode identity (Bearer dev:<agent_id>) behind a flag in the platform, plus a stub for any external identity provider, so the bench is hermetic. Whatever you do, keep real credentials in env vars, never in runtime code or scenario definitions.

The paved path: `http_inject`¶

If your agent process can expose two HTTP routes, prefer the built-in http_inject runtime over writing a custom driver. Implement the Contract C routes (POST /wt/inject and POST /wt/reset) described in the inject-protocol design, then run Wind Tunnel with wt run --runtime http_inject. The runtime sends only the newest user turn, validates the strict response envelope, and converts ordered {name, arguments} tool-call objects into the OpenAI-shaped SPI response that Wind Tunnel traces already understand.

Runtime configuration is deliberately narrow:

Setting	Default	Meaning
`WT_INJECT_URL`	`http://127.0.0.1:8647`	Base URL for the Contract C endpoint.
`WT_INJECT_TIMEOUT_S`	`120.0`	Per-request agent deadline sent as `timeout_s`; the driver adds a fixed five-second transport grace.

Contract C v1 has no route for dynamic tool registration. The endpoint owns its own tool wiring and must report the complete ordered tool-call transcript in every /wt/inject response, even when the list is empty.

A minimal skeleton¶

import httpx

from windtunnel.spi.agent_runtime import AgentConfig


class HttpAgentHandle:
    def __init__(self, base_url: str, api_key: str, timeout: float = 300.0):
        self._client = httpx.Client(base_url=base_url, timeout=timeout,
                                    headers={"Authorization": f"Bearer {api_key}"})

    def send(self, messages, session_id):
        run = self._client.post("/v1/runs", json={
            "input": messages, "session_id": session_id,
        }).json()
        # Drain the events stream; reconstruct an OpenAI-shaped response,
        # collecting intermediate tool_calls — not just the final message.
        return self._drain_events(run["run_id"])

    def reset_state(self):
        resp = self._client.post("/admin/reset", json={"scope": "all"})
        resp.raise_for_status()   # a failed wipe must be fatal

    def teardown(self):
        try:
            self._client.close()
        except Exception:
            pass                  # teardown must not raise


class HttpRuntime:
    def __init__(self, base_url: str, api_key: str):
        self._base_url, self._api_key = base_url, api_key

    def provision(self, config: AgentConfig, mcps=None) -> HttpAgentHandle:
        for handle in (mcps or []):
            self._register_tools(handle.url)   # incl. localhost translation
        return HttpAgentHandle(self._base_url, self._api_key)

Registering it with the `wt` CLI¶

wt run --runtime <name> finds a runtime through a RuntimePlugin (windtunnel/spi/runtime_plugin.py): an object with build(runtime_name, label, soul_path) -> AgentRuntime, registered under the windtunnel.runtimes entry-point group or passed as a module:attr path. Everything else on it is optional and discovered by capability:

class HttpPlugin:
    # How many scenario jobs may run at once, each provisioning its own
    # handle from the one built runtime. Omit it (= 1) unless concurrent
    # sessions are truly isolated; None declares no limit.
    max_concurrency = 4

    def build(self, runtime_name, label, soul_path):
        return HttpRuntime(base_url=..., api_key=...)

    # Which sweeps collide: two `wt run` invocations with the same key take
    # turns. Default: the runtime name. Key by the backend when one name can
    # reach several (here, a URL read from the environment).
    def lock_key(self, runtime_name):
        return f"{runtime_name}:{os.environ.get('MY_AGENT_URL', 'http://127.0.0.1:9000')}"

    # The model this runtime answers with, as an opaque label. Scopes ladder
    # evidence to the model and is checked against [tool.windtunnel.ladder.models].
    # Return None when unknown; never guess.
    def model_label(self, runtime_name):
        return os.environ.get("MY_AGENT_MODEL")

    def pre_run(self, runtime, scenarios, runtime_name):   # once, before the sweep
        ...

    def post_run(self, runtime, scenarios, runtime_name):  # once, even on abort
        ...

max_concurrency is what makes wt run --scheduler concurrent safe: the CLI never runs more jobs at once than the plugin declares, whatever --max-concurrency asks for. The default of 1 is deliberate — a runtime that binds fixed ports, or whose reset_state() wipes a backend shared by every session, must not be driven concurrently. Any finite value also makes each sweep hold a machine-wide lock on lock_key from before build() until after post_run(), so a second sweep queues instead of colliding (see iterating on an agent).

Token cost comes from the handle, not the plugin: a send() response that carries an OpenAI-style usage object (input_tokens/output_tokens or prompt_tokens/completion_tokens) is summed into the trace's usage; a response without one records the run's usage as unknown (see 0005: Experiment ladder).

Per-call usage (optional). If your usage object also reports a cache split, it is picked up automatically — wt normalizes every send's raw usage dict into one entry on trace.model_calls, in turn order:

{"conversation": "<this run's session_id>", "prompt_tokens": 1500, "cached_tokens": 1300, "completion_tokens": 30}

prompt_tokens is the TOTAL prompt tokens for that call, including any served from a prompt cache — the same meaning as usage.input_tokens. A field is null, never guessed, when that one call's usage dict didn't report it. Accepted spellings, checked in order, first match wins:

Field	Spellings
`cached_tokens`	`cached_tokens`, `prompt_tokens_details.cached_tokens` (nested), `cache_read_input_tokens`, `cacheRead`
`prompt_tokens`	`prompt_tokens`, `input_tokens`, `input`
`completion_tokens`	`completion_tokens`, `output_tokens`, `output`

One wire shape reports input as uncached-only, keeping the cache read in a separate cacheRead field rather than nesting it inside the prompt total — when a usage dict has both, prompt_tokens is their sum, so it always means "including cached" like every other spelling.

Nothing is required beyond what your usage object already reports: without a cache split, calls still get prompt_tokens/completion_tokens entries (cached_tokens: null), and a runtime that reports no usage at all leaves trace.model_calls as null. Per-call usage feeds the pricing cache-read rate and the prompt-cache-miss check — see Per-call usage and the prompt-cache-miss check.

Several inference calls in one send. One send() can cover several model calls under the hood — an agent loop's tool call, tool result, call again, all inside the one turn your handle returns. Reporting only that turn's aggregate usage would dilute a miss on one of those calls into the turn's overall ratio. Nest a "calls" list under usage instead, one usage dict per inference call, in order:

{
  "usage": {
    "prompt_tokens": 10000, "cached_tokens": 9500, "completion_tokens": 60,
    "calls": [
      {"prompt_tokens": 1000, "cached_tokens": 0, "completion_tokens": 10},
      {"prompt_tokens": 4000, "cached_tokens": 500, "completion_tokens": 20},
      {"prompt_tokens": 5000, "cached_tokens": 4900, "completion_tokens": 30}
    ]
  }
}

When usage.calls is a non-empty list, each entry becomes its own trace.model_calls entry (normalized the same way as a per-send usage dict) instead of the one aggregated entry — so the prompt-cache-miss check still catches a single bad call even when the send's own aggregate looks fine. Without calls, wt falls back to one entry per send, as above. An entry may also carry its own "conversation" string, overriding this run's session_id default — for a call that starts its own side conversation (e.g. a forked review call), so its own first call is judged as a first call of its own conversation, not folded into the run's.

Checklist before trusting your runtime¶

[ ] Conformance tests pass (test_runtime_conformance.py pointed at your runtime).
[ ] If provision_reference() is implemented, golden and poison cases each receive fresh handles and teardown removes the inference substitution.
[ ] A must_call scenario passes — proves intermediate tool calls surface.
[ ] Two scenarios run back-to-back with a deliberately stateful first scenario — proves reset_state() actually isolates runs. Automate this with windtunnel.api.run_reset_canary(), the packaged version of the check: it seeds a random nonce into one session, resets, and probes a fresh session for it. A recalled nonce proves a leak; a clean run is evidence, not proof, of isolation — pair it with a StateProbe for stateful backends (see the reset canary section of the inject-protocol design doc for the full asymmetry rationale).

Two ways to run it:
- wt doctor --runtime <name> — the bring-up command. Run it once against a freshly stood-up stack; it needs a live model behind the runtime (recall mode: seed, reset, ask a fresh session to recall).
- From pytest, in CI without a live model, pass probe_recall=False and a StateProbe — this is the hermetic path: seeding still uses send() (a stub model is fine, it only needs to ingest), but the check is reset → scan the probe's post-reset snapshot for the nonce. No probe turns, no send() after reset.
```
from windtunnel.api import run_reset_canary

def test_reset_canary_hermetic():
    result = run_reset_canary(
        runtime, config, probe_recall=False, state_probe=my_state_probe,
    )
    assert not result.leaked, result.detail
```
  - [ ] sampling.temperature=0 twice produces (near-)identical traces — proves the sampling config actually reaches the model. - [ ] Kill the bench mid-run; rerun — proves teardown()/provision() recover from dirty state.

Writing a runtime¶

The contract at a glance¶

provision(config, mcps) — once per batch¶

Optional provision_reference() — certify the harness¶

send(messages, session_id) — one turn¶

reset_state() vs teardown()¶