cendor-tokenguard — budget

Stop runaway LLM bills, and get per-feature / per-user cost attribution for free. One decorator caps a unit of work; one context manager tags its spend. No dashboard, no account, no infrastructure.

pip install cendor-tokenguard

npm i @cendor/tokenguard

Quickstart

from cendor.core import instrument
from cendor.tokenguard import budget, track, report

client = instrument(openai_client)

@budget(usd=0.50, on_exceed="downgrade", downgrade={"gpt-4o": "gpt-4o-mini"})
def answer(q: str) -> str:
    with track(feature="support_bot", user_id="alice"):
        resp = client.chat.completions.create(model="gpt-4o", messages=[{"role": "user", "content": q}])
        return resp.choices[0].message.content

for row in report(group_by=["feature", "user_id"]):
    print(row["tags"], row["usd"], row["tokens"], row["calls"])

import { instrument } from '@cendor/core';
import { budget, track, report } from '@cendor/tokenguard';

const client = instrument(openaiClient);

const answer = budget({ usd: 0.50, onExceed: 'downgrade', downgrade: { 'gpt-4o': 'gpt-4o-mini' } })(
  (q: string) => track({ feature: 'support_bot', userId: 'alice' }, async () => {
    const resp = await client.chat.completions.create({
      model: 'gpt-4o', messages: [{ role: 'user', content: q }] });
    return resp.choices[0].message.content;
  }));

for (const row of report(['feature', 'userId'])) {
  console.log(row.tags, row.usd.toString(), row.tokens, row.calls);
}

See it in the stack. The connected support-agent recipe (budget + context + audit) is in the Cookbook.

Core concepts

How it enforces

tokenguard subscribes to core’s event bus — it never patches a client. Once your client is instrumented, @budget enforces a cap and track(...) attributes spend by tag, with no per-call wiring. Enforcement happens at two moments: a pre-flight projection (before a call runs) and post-flight accounting (after a call returns, reading the real Usage/Money off the emitted LLMCall).

Hard cap vs runaway guard — pick by intent

This is the distinction that trips people up:

• Hard cap that must never be exceeded → "block" or "downgrade" (both pre-flight): they project the next call’s cost and refuse or reroute it before it runs, so spend stays at or under the cap.
• Cheap runaway-loop stop → "raise" (post-flight): the breaker trips only after a call returns and pushes spend over the cap, so the breaching call has already run and been billed. Spend therefore overshoots by one call. "raise" stops the next iteration, not the breaching one.

Reasoning models

A reasoning model’s hidden thinking can’t be predicted pre-flight, so no projection bounds a single call in advance. Two mechanisms cover them: the cumulative gate ("raise"/"block") enforces on the recorded usage, which already includes reasoning (OpenAI folds it into completion_tokens, Anthropic into output_tokens); and "clamp" hands the provider its own ceiling so one call is capped server-side. Reasoning tokens are billed at the output rate, so cost is exact either way.

Cost attribution

track(**tags) attributes ambient spend (feature / user_id / session_id …) via contextvars, across nested and async calls. report(group_by=[…]) aggregates per tag, and report().assert_under(usd=…, **tags) turns cost into a test assertion.

Threads don’t inherit track/budget. They ride contextvars, so an asyncio task inherits them but a plain threading.Thread you start does not — carry them across with contextvars.copy_context():

import contextvars, threading
ctx = contextvars.copy_context()          # captures the active budget + tags
threading.Thread(target=lambda: ctx.run(worker)).start()

Streaming timing

A streamed call is accounted once its stream is drained (consumed), not when it’s launched — the LLMCall is emitted only when the chunk iterator is exhausted or closed. So a loop that launches many streams before draining them can overspend under post-flight modes. Drain (or close) each stream before starting the next, or gate spend with a pre-flight mode ("block"/"downgrade"/"clamp"), which is evaluated before the call runs.

Unpriced models — a USD blind spot

A call whose model has no price records $0, so a USD cap can’t enforce against it. tokenguard warns once per model (UnpricedModelWarning) and counts these in unpriced_calls(). A token cap is unaffected — tokens are counted regardless of price — so prefer a tokens= cap (or add a rate via core.prices) for a model that isn’t in the table.

Functions & classes

budget()

A decorator and context manager that caps a unit of work. Budgets nest — the tightest applicable cap wins, and an inner downgrade/clamp never masks an outer hard cap.

budget(usd=None, tokens=None, on_exceed="raise", scope=None,
       downgrade=None, output_reserve=256, reasoning_reserve=0)

budget({ usd, tokens, onExceed = 'raise', scope,
         downgrade, outputReserve = 256, reasoningReserve = 0 })(fn)   // decorator form
await withBudget({ usd: 0.25, onExceed: 'block' }, () => { /* ... */ });  // scoped form

on_exceed modes

Config is validated eagerly: a missing cap, an unknown on_exceed, "downgrade" without a map/usd cap, or "clamp" without a tokens= cap raises ValueError — no silent no-op budgets.

track() & estimate()

with track(feature="support", user_id="alice"):   # tag ambient spend (contextvars)
    ...
estimate(model, messages, max_output_tokens=0)     # price a call WITHOUT making it -> Money

await track({ feature: 'support', userId: 'alice' }, async () => {  // tag ambient spend (ALS)
  /* ... */
});
estimate(model, messages, 0);                      // price a call WITHOUT making it -> Money

report()

Aggregates recorded spend into rows of {tags, usd, tokens, input_tokens, output_tokens, reasoning_tokens, calls, unpriced_calls}. reasoning_tokens is the portion of output_tokens spent reasoning (a subset, not added into tokens); unpriced_calls is how many of the group’s calls had no price. report().assert_under(usd=…, **tags) turns cost into a test assertion.

Introspection & config

QueueSink — low-latency durable logging

The bus fans out to subscribers inline, so a durable sink (SQLite/OTel/file) adds its write latency to every model call. On a long or high-throughput run that’s a latency cliff. Wrap the sink in sinks.QueueSink to move that I/O onto a background thread — write() enqueues and returns immediately, and a single worker drains it into the inner sink in order:

from cendor.tokenguard import use_sink
from cendor.tokenguard.sinks import QueueSink, SQLiteSink

sink = QueueSink(SQLiteSink("spend.db"))     # durable logging, off the hot path
use_sink(sink)
# … the run: model calls no longer pay the sink's I/O latency …
sink.flush()     # block until the queue is drained (e.g. at a checkpoint)
sink.close()     # flush + stop the worker + close the inner sink (or use `with QueueSink(...)`)

import { useSink } from '@cendor/tokenguard';
import { QueueSink, SQLiteSink } from '@cendor/tokenguard/sinks';

const sink = new QueueSink(new SQLiteSink('spend.db'));  // durable logging, off the hot path
useSink(sink);
// … the run: model calls no longer pay the sink's I/O latency …
await sink.flush();  // resolve once the queue is drained (e.g. at a checkpoint)
await sink.close();  // flush + stop the drain loop + close the inner sink

• Ordering preserved (single FIFO worker); max_queue=N applies back-pressure when full (a row is never silently dropped) — None (default) is unbounded.
• Durability is opt-in at shutdown: the worker is a daemon thread, so call flush()/close() before exit or a hard crash can drop still-queued rows. flush()/close() are the optional core.protocols.Sink lifecycle methods.

How it works

%%{init: {"flowchart": {"htmlLabels": false}} }%%
graph TD
    CALL["instrumented LLM call"]
    PRE{"pre-flight:<br/>would it breach the cap?"}
    BLOCK["block: raise BudgetExceeded<br/>(the call never runs)"]
    DOWN["downgrade: reroute<br/>to a cheaper model"]
    RUN["the call runs"]
    BUS["bus: LLMCall<br/>with usage + cost"]
    REC["record spend by tags<br/>track(feature, user_id)"]
    POST{"over cap now?"}
    STOP["raise / truncate<br/>(stops the next call)"]
    REP["report(group_by)<br/>assert_under()"]

    CALL --> PRE
    PRE -->|block| BLOCK
    PRE -->|downgrade| DOWN --> RUN
    PRE -->|"within budget"| RUN
    RUN --> BUS --> REC --> POST
    POST -->|yes| STOP
    REC --> REP

    classDef tg fill:#8B5CF6,color:#ffffff,stroke:#7C3AED;
    classDef co fill:#94A3BB,color:#0F172A,stroke:#64748B;
    classDef stop fill:#F43F5E,color:#ffffff,stroke:#E11D48;
    class DOWN,REC,REP tg;
    class BUS co;
    class BLOCK,STOP stop;

• Post-flight accounting: the bus subscriber reads actual Usage/Money off each emitted LLMCall, records a row keyed by the active tags, and decrements the active budget(s).
• Pre-flight enforcement: a core interceptor estimates the next call and, with "block"/"downgrade"/"clamp", refuses / reroutes / caps it before it runs.
• Bounded memory: the in-memory spend buffer is FIFO-capped (default 100k rows); attach a sink for durable, complete history.

Plugs into the stack

Wrap-around. It rides the call you already make — you don’t change the call itself. Once the client is instrumented, @budget enforces and track records automatically. In a managed-runtime setup, enforce a coarser budget at your entrypoint and ingest actual spend from the runtime’s gen_ai.* spans via core.otel.ingest.

Honest limits

• "raise" overshoots by one call — it’s post-flight. For a true ceiling, use "block".
• Streaming is accounted on drain, so fanning out many undrained streams can overspend under post-flight modes; use a pre-flight mode or drain each stream in turn.
• Unpriced models are a USD blind spot (they record $0) — prefer a tokens= cap or add a rate. tokenguard warns once per model and counts them in unpriced_calls().
• State is in-process and module-global — ideal for a single worker. For multi-process, put durable spend through a sink rather than the in-memory aggregate.