Streaming is a metering bypass until you make it not one
Meter SSE responses by tapping the stream with a pass-through scanner that extracts provider-reported usage at flush time — and make your protocol translators preserve usage and termination guarantees.
If your proxy enforces token budgets by reading the usage object from upstream JSON responses, every streamed request is a budget bypass. An SSE response is a byte stream you’re forwarding chunk-by-chunk to the client; there is no response body to parse after the fact, and by the time the stream ends, your handler has usually already returned. In Cookey — a gateway whose entire pitch is “grant apps budget-capped access to your API keys” — this would have meant any app could evade its token budget by adding "stream": true. The comment in the fix names the stakes plainly: without stream metering, “streamed requests would never count against token budgets.”
The problem has a second layer when the gateway translates protocols. Cookey normalizes every LLM provider to OpenAI chat-completions shape, so the Anthropic adapter rewrites Anthropic’s SSE events into OpenAI chunks on the fly. The first version of that translator dropped usage data entirely (Anthropic reports it in message_start and message_delta events that have no OpenAI equivalent), overwrote earlier system messages instead of concatenating them, and could lose the final SSE line if the upstream closed without a trailing newline — which also meant clients might never receive data: [DONE].
Adapters translate, the pipeline meters
The design separates two responsibilities that are tempting to entangle. Adapters translate wire formats and must emit gateway-canonical SSE in which usage appears as a usage object on some data: chunk — for providers that don’t natively do this, the adapter synthesizes that chunk. The pipeline meters any canonical stream with one generic wrapper that never modifies bytes: clients receive exactly what the adapter produced, and metering is a read-only tap. That split means one metering implementation covers every current and future provider, and adapters can be tested purely as format translators.
The wrapper also accepts an honest limitation: usage is recorded after the stream completes, not enforced during it. A streamed response that blows past the remaining budget finishes streaming; the next request gets denied. Mid-stream enforcement would mean killing streams on estimated counts — worse behavior in exchange for tighter accounting on a boundary that budgets (which cap totals, not instants) don’t need.
A read-only tap
The tap is ~60 lines (apps/proxy/src/server/gateway/stream-usage.ts): a TransformStream that enqueues every chunk untouched, then scans a line buffer for usage:
export function createUsageScanningStream(
source: ReadableStream<Uint8Array>,
usageSpec: UsageSpec,
onComplete: (usage: ExtractedUsage) => Promise<void>,
): ReadableStream<Uint8Array> {
const decoder = new TextDecoder();
let buffer = "";
let latest: ExtractedUsage = {};
const scanLine = (line: string) => {
const trimmed = line.trim();
if (!trimmed.startsWith("data:")) return;
const payload = trimmed.slice(5).trim();
if (!payload || payload === "[DONE]") return;
try {
const parsed = JSON.parse(payload);
const usage = extractUsage(usageSpec, parsed);
if (usage.totalTokens !== undefined || usage.inputTokens !== undefined /* … */) {
latest = { ...latest, ...usage };
}
} catch {
// Partial or non-JSON chunk — ignore; passthrough is unaffected
}
};
return source.pipeThrough(
new TransformStream<Uint8Array, Uint8Array>({
transform(chunk, controller) {
controller.enqueue(chunk); // client gets bytes untouched
buffer += decoder.decode(chunk, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() ?? ""; // keep the partial line
for (const line of lines) scanLine(line);
},
async flush() {
if (buffer) scanLine(buffer); // final line may lack a newline
if (latest.totalTokens === undefined) {
const input = latest.inputTokens ?? 0;
const output = latest.outputTokens ?? 0;
if (input + output > 0) latest.totalTokens = input + output;
}
await onComplete(latest).catch(() => {
// Usage recording must never break the client's stream
});
},
}),
);
}This sits in the pipeline right where the upstream response body is handed back for streaming — you wrap the ReadableStream and return the wrapped one; onComplete calls the same recordTokenUsage the non-streaming path uses, so both paths hit identical counters.
The adapter side of the contract is visible in the Anthropic translator’s fix (commit 5e955af, apps/proxy/src/server/adapters/anthropic-messages.ts). It accumulates input_tokens from message_start and output_tokens from message_delta, then emits an OpenAI stream_options-style final usage chunk before [DONE]:
const emitDone = (controller) => {
if (doneEmitted) return;
doneEmitted = true;
if (inputTokens > 0 || outputTokens > 0) {
controller.enqueue(encoder.encode(`data: ${JSON.stringify({
id: messageId, object: "chat.completion.chunk", model,
choices: [],
usage: {
prompt_tokens: inputTokens,
completion_tokens: outputTokens,
total_tokens: inputTokens + outputTokens,
},
})}\n\n`));
}
controller.enqueue(encoder.encode("data: [DONE]\n\n"));
};
// …
flush(controller) {
// A final SSE line without a trailing newline must not be dropped,
// and clients must always see a terminating [DONE].
if (buffer) processLine(buffer, controller);
emitDone(controller);
},The flush handler and the idempotent emitDone are the difference between a translator that works in demos and one that works against real upstreams: providers do close streams without trailing newlines, and clients (and downstream scanners) do hang waiting for a [DONE] that a naive translator only emits when the upstream happens to send message_stop.
Degrading without touching the stream
Each layer degrades independently and never at the client’s expense. A provider that reports no usage in its stream yields an empty extraction — the request was still admitted against request-count quotas and pre-checked against the existing token budget, so the gap is one response’s tokens uncounted, not an open valve. Malformed or partial JSON in a chunk is skipped by the scanner without disturbing passthrough. If usage recording fails (database blip), the .catch in flush swallows it: a metering write must never truncate a client’s stream. A client that disconnects mid-stream cancels the pipe before flush — that response’s tokens go unrecorded, an accepted loss consistent with “record on completion.” Note the asymmetry with admission-time checks, which fail closed; metering fails open, because by flush time the tokens are already spent and the only question is bookkeeping.
Streaming stops being a policy dimension
Once the tap is in place, "stream": true stops being a policy dimension — budgets, dashboards, and spend projections mean the same thing for streamed and buffered traffic, and you never have to choose between offering streaming and enforcing caps. Adding a provider imposes one metering obligation, stated in the adapter contract: emit usage as a canonical chunk, synthesizing it if the provider won’t. And your streaming bugs become legible: if budget dashboards show zeros for one provider, the fault is that provider’s adapter not honoring the contract — one file to fix, with the scanner and every other adapter unaffected.