Skip to content

ontocast.onto.run_meter

Per-run token meter with enforceable limits.

A :class:RunMeter counts tokens for one run as they are spent, across every concurrent content unit, and refuses the next provider call once a limit is reached: a cap on a token count, or a caller's limit policy. The meter knows nothing about money; a caller that prices tokens does so in its limit.

The meter is bound to the running task with :func:use_run_meter. asyncio tasks (and LangGraph nodes) copy the context they start in, so every unit of a fan-out records into the same object.

A refusal is an exception, not a state field: :class:BudgetExceededError propagates out of the unit loops and best-effort steps, the fan-out re-raises it once every sibling unit has finished, and it leaves the graph run (astream / ainvoke) carrying the meter's snapshot. The meter itself also stays readable after the run, so a caller can settle from meter.snapshot() either way.

Deliberately a leaf module with no OntoCast imports, so :mod:ontocast.tool.llm can import it without reaching back into :mod:ontocast.tool.

Attributes

TOKEN_FIELDS = tuple(TokenTotals.model_fields) module-attribute

logger = logging.getLogger(__name__) module-attribute

Type Aliases

RunLimit = Callable[[Mapping[str, TokenTotals]], str | None]

Classes

BudgetExceededError

Bases: RuntimeError

A run reached a token cap or its meter's limit.

Raised before the provider call that would have exceeded it, so that call is never made. Calls already in flight finish and are recorded, so the final usage can pass the limit by what they spent.

Attributes:

Name Type Description
snapshot

The meter's state when the call was refused.
Source code in ontocast/onto/run_meter.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class BudgetExceededError(RuntimeError):
    """A run reached a token cap or its meter's ``limit``.

    Raised before the provider call that would have exceeded it, so that call is
    never made. Calls already in flight finish and are recorded, so the final
    usage can pass the limit by what they spent.

    Attributes:
        snapshot: The meter's state when the call was refused.
    """

    def __init__(self, message: str, snapshot: RunMeterSnapshot) -> None:
        super().__init__(message)
        self.snapshot = snapshot

Attributes

snapshot = snapshot instance-attribute

Methods:

__init__(message, snapshot)
Source code in ontocast/onto/run_meter.py 
102
103
104
def __init__(self, message: str, snapshot: RunMeterSnapshot) -> None:
    super().__init__(message)
    self.snapshot = snapshot

RunMeter

Thread- and task-safe token meter for one run.

check(), called before every provider call, raises :class:BudgetExceededError once any field summed over models reaches its token_caps entry, or once limit returns a reason. After the first refusal every later check() refuses too.

Parameters:

Name Type Description Default
limit RunLimit | None

The caller's policy, given a copy of the counts per model before each provider call; returns a reason to stop, or None. Called outside the meter's lock, so it may read the meter; an exception it raises propagates to the call it was guarding.

 None
token_caps dict[str, int] | None

Limits on token counts summed over models, keyed by field name.

 None
on_usage Callable[[RunMeterSnapshot], None] | None

Called with a snapshot after usage is recorded, at most once every min_interval_s seconds, and on :meth:flush. Called outside the lock; an exception it raises is logged, not raised.

 None
min_interval_s float

Minimum seconds between two on_usage calls.

 0.0
clock Callable[[], float]

Monotonic time source for the throttle.

 monotonic

Raises:

Type Description
ValueError

On an unknown field name in token_caps.
Source code in ontocast/onto/run_meter.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
class RunMeter:
    """Thread- and task-safe token meter for one run.

    ``check()``, called before every provider call, raises
    :class:`BudgetExceededError` once any field summed over models reaches its
    ``token_caps`` entry, or once ``limit`` returns a reason. After the first
    refusal every later ``check()`` refuses too.

    Args:
        limit: The caller's policy, given a copy of the counts per model before
            each provider call; returns a reason to stop, or None. Called
            outside the meter's lock, so it may read the meter; an exception it
            raises propagates to the call it was guarding.
        token_caps: Limits on token counts summed over models, keyed by field
            name.
        on_usage: Called with a snapshot after usage is recorded, at most once
            every ``min_interval_s`` seconds, and on :meth:`flush`. Called
            outside the lock; an exception it raises is logged, not raised.
        min_interval_s: Minimum seconds between two ``on_usage`` calls.
        clock: Monotonic time source for the throttle.

    Raises:
        ValueError: On an unknown field name in ``token_caps``.
    """

    def __init__(
        self,
        *,
        limit: RunLimit | None = None,
        token_caps: dict[str, int] | None = None,
        on_usage: Callable[[RunMeterSnapshot], None] | None = None,
        min_interval_s: float = 0.0,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        for name in token_caps or {}:
            if name not in TokenTotals.model_fields:
                raise ValueError(
                    f"Unknown token field {name!r}; expected one of {TOKEN_FIELDS}"
                )
        self.limit = limit
        self.token_caps: dict[str, int] = dict(token_caps or {})
        self.on_usage = on_usage
        self.min_interval_s = min_interval_s
        self.usage_by_model: dict[str, TokenTotals] = {}
        self._clock = clock
        self._lock = threading.Lock()
        self._exceeded: str | None = None
        self._last_emit: float | None = None

    @classmethod
    def from_token_limits(
        cls,
        *,
        max_input_tokens: int | None = None,
        max_output_tokens: int | None = None,
    ) -> RunMeter | None:
        """A meter capping ``input`` and ``output``, or None if unlimited.

        Args:
            max_input_tokens: Cap on provider input tokens over all models.
            max_output_tokens: Cap on provider output tokens over all models.
        """
        caps: dict[str, int] = {}
        if max_input_tokens is not None:
            caps["input"] = max_input_tokens
        if max_output_tokens is not None:
            caps["output"] = max_output_tokens
        return cls(token_caps=caps) if caps else None

    @property
    def exceeded(self) -> str | None:
        """Why the meter refused a call, or None if it has not."""
        return self._exceeded

    def record(self, model: str, usage: TokenTotals) -> None:
        """Add ``usage`` to ``model``'s totals and notify ``on_usage`` if due."""
        with self._lock:
            self.usage_by_model.setdefault(model, TokenTotals()).add(usage)
        self._maybe_emit()

    def totals(self) -> TokenTotals:
        """Counts summed over every model."""
        with self._lock:
            return self._totals_locked()

    def usage(self) -> dict[str, TokenTotals]:
        """A copy of the counts per model."""
        with self._lock:
            return self._usage_copy_locked()

    def check(self) -> None:
        """Raise :class:`BudgetExceededError` if a limit has been reached."""
        with self._lock:
            reason = self._exceeded or self._capped_locked()
            usage = self._usage_copy_locked() if reason is None else None
        if reason is None and self.limit is not None and usage is not None:
            reason = self.limit(usage)
        if reason is None:
            return
        with self._lock:
            if self._exceeded is None:
                self._exceeded = reason
            reason = self._exceeded
            snapshot = self._snapshot_locked()
        raise BudgetExceededError(f"Run budget exceeded: {reason}", snapshot)

    def snapshot(self) -> RunMeterSnapshot:
        """Current state as a JSON-serialisable dict."""
        with self._lock:
            return self._snapshot_locked()

    def flush(self) -> None:
        """Call ``on_usage`` now, whatever the throttle."""
        self._maybe_emit(force=True)

    def _maybe_emit(self, *, force: bool = False) -> None:
        callback = self.on_usage
        if callback is None:
            return
        with self._lock:
            now = self._clock()
            due = (
                force
                or self._last_emit is None
                or now - self._last_emit >= self.min_interval_s
            )
            if not due:
                return
            self._last_emit = now
            snapshot = self._snapshot_locked()
        try:
            callback(snapshot)
        except Exception:
            logger.exception("RunMeter on_usage callback failed")

    def _totals_locked(self) -> TokenTotals:
        total = TokenTotals()
        for usage in self.usage_by_model.values():
            total.add(usage)
        return total

    def _usage_copy_locked(self) -> dict[str, TokenTotals]:
        return {
            model: usage.model_copy() for model, usage in self.usage_by_model.items()
        }

    def _capped_locked(self) -> str | None:
        totals = self._totals_locked()
        for name, cap in self.token_caps.items():
            used = getattr(totals, name)
            if used >= cap:
                return f"{name} {used} reached its cap of {cap}"
        return None

    def _snapshot_locked(self) -> RunMeterSnapshot:
        totals = self._totals_locked()
        return RunMeterSnapshot(
            usage_by_model={
                model: usage.model_dump()
                for model, usage in self.usage_by_model.items()
            },
            totals=totals.model_dump(),
            token_caps=dict(self.token_caps),
            exceeded=self._exceeded,
        )

Attributes

exceeded property

Why the meter refused a call, or None if it has not.

limit = limit instance-attribute
min_interval_s = min_interval_s instance-attribute
on_usage = on_usage instance-attribute
token_caps = dict(token_caps or {}) instance-attribute
usage_by_model = {} instance-attribute

Methods:

__init__(*, limit=None, token_caps=None, on_usage=None, min_interval_s=0.0, clock=time.monotonic)
Source code in ontocast/onto/run_meter.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def __init__(
    self,
    *,
    limit: RunLimit | None = None,
    token_caps: dict[str, int] | None = None,
    on_usage: Callable[[RunMeterSnapshot], None] | None = None,
    min_interval_s: float = 0.0,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    for name in token_caps or {}:
        if name not in TokenTotals.model_fields:
            raise ValueError(
                f"Unknown token field {name!r}; expected one of {TOKEN_FIELDS}"
            )
    self.limit = limit
    self.token_caps: dict[str, int] = dict(token_caps or {})
    self.on_usage = on_usage
    self.min_interval_s = min_interval_s
    self.usage_by_model: dict[str, TokenTotals] = {}
    self._clock = clock
    self._lock = threading.Lock()
    self._exceeded: str | None = None
    self._last_emit: float | None = None
check()

Raise :class:BudgetExceededError if a limit has been reached.

Source code in ontocast/onto/run_meter.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def check(self) -> None:
    """Raise :class:`BudgetExceededError` if a limit has been reached."""
    with self._lock:
        reason = self._exceeded or self._capped_locked()
        usage = self._usage_copy_locked() if reason is None else None
    if reason is None and self.limit is not None and usage is not None:
        reason = self.limit(usage)
    if reason is None:
        return
    with self._lock:
        if self._exceeded is None:
            self._exceeded = reason
        reason = self._exceeded
        snapshot = self._snapshot_locked()
    raise BudgetExceededError(f"Run budget exceeded: {reason}", snapshot)
flush()

Call on_usage now, whatever the throttle.

Source code in ontocast/onto/run_meter.py 
218
219
220
def flush(self) -> None:
    """Call ``on_usage`` now, whatever the throttle."""
    self._maybe_emit(force=True)
from_token_limits(*, max_input_tokens=None, max_output_tokens=None) classmethod

A meter capping input and output, or None if unlimited.

Parameters:

Name Type Description Default
max_input_tokens int | None

Cap on provider input tokens over all models.

 None
max_output_tokens int | None

Cap on provider output tokens over all models.

 None
Source code in ontocast/onto/run_meter.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@classmethod
def from_token_limits(
    cls,
    *,
    max_input_tokens: int | None = None,
    max_output_tokens: int | None = None,
) -> RunMeter | None:
    """A meter capping ``input`` and ``output``, or None if unlimited.

    Args:
        max_input_tokens: Cap on provider input tokens over all models.
        max_output_tokens: Cap on provider output tokens over all models.
    """
    caps: dict[str, int] = {}
    if max_input_tokens is not None:
        caps["input"] = max_input_tokens
    if max_output_tokens is not None:
        caps["output"] = max_output_tokens
    return cls(token_caps=caps) if caps else None
record(model, usage)

Add usage to model's totals and notify on_usage if due.

Source code in ontocast/onto/run_meter.py 
181
182
183
184
185
def record(self, model: str, usage: TokenTotals) -> None:
    """Add ``usage`` to ``model``'s totals and notify ``on_usage`` if due."""
    with self._lock:
        self.usage_by_model.setdefault(model, TokenTotals()).add(usage)
    self._maybe_emit()
snapshot()

Current state as a JSON-serialisable dict.

Source code in ontocast/onto/run_meter.py 
213
214
215
216
def snapshot(self) -> RunMeterSnapshot:
    """Current state as a JSON-serialisable dict."""
    with self._lock:
        return self._snapshot_locked()
totals()

Counts summed over every model.

Source code in ontocast/onto/run_meter.py 
187
188
189
190
def totals(self) -> TokenTotals:
    """Counts summed over every model."""
    with self._lock:
        return self._totals_locked()
usage()

A copy of the counts per model.

Source code in ontocast/onto/run_meter.py 
192
193
194
195
def usage(self) -> dict[str, TokenTotals]:
    """A copy of the counts per model."""
    with self._lock:
        return self._usage_copy_locked()

RunMeterSnapshot

Bases: TypedDict

JSON-serialisable state of a :class:RunMeter.

Source code in ontocast/onto/run_meter.py 
82
83
84
85
86
87
88
class RunMeterSnapshot(TypedDict):
    """JSON-serialisable state of a :class:`RunMeter`."""

    usage_by_model: dict[str, dict[str, int]]
    totals: dict[str, int]
    token_caps: dict[str, int]
    exceeded: str | None

Attributes

exceeded instance-attribute
token_caps instance-attribute
totals instance-attribute
usage_by_model instance-attribute

TokenTotals

Bases: BaseModel

Token counts for one model, split by where each response came from.

input to reasoning count provider calls made in this run, as the provider reports them: cache_read and cache_write are counted inside input, reasoning inside output. replayed_* count responses served by the OntoCast disk cache, for which no provider call was made.

Source code in ontocast/onto/run_meter.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class TokenTotals(BaseModel):
    """Token counts for one model, split by where each response came from.

    ``input`` to ``reasoning`` count provider calls made in this run, as the
    provider reports them: ``cache_read`` and ``cache_write`` are counted inside
    ``input``, ``reasoning`` inside ``output``. ``replayed_*`` count responses
    served by the OntoCast disk cache, for which no provider call was made.
    """

    input: int = Field(
        default=0, description="Input tokens of provider calls, cache reads included."
    )
    cache_read: int = Field(
        default=0, description="Input tokens served from the provider's prompt cache."
    )
    cache_write: int = Field(
        default=0, description="Input tokens written to the provider's prompt cache."
    )
    output: int = Field(
        default=0, description="Output tokens of provider calls, reasoning included."
    )
    reasoning: int = Field(default=0, description="Reasoning tokens of provider calls.")
    replayed_input: int = Field(
        default=0, description="Input tokens of responses replayed from disk cache."
    )
    replayed_output: int = Field(
        default=0, description="Output tokens of responses replayed from disk cache."
    )
    calls: int = Field(default=0, description="Provider calls that returned.")
    replays: int = Field(default=0, description="Responses replayed from disk cache.")

    def add(self, other: TokenTotals) -> None:
        """Add every count of ``other`` to this one."""
        for name in TokenTotals.model_fields:
            setattr(self, name, getattr(self, name) + getattr(other, name))

Attributes

cache_read = Field(default=0, description="Input tokens served from the provider's prompt cache.") class-attribute instance-attribute
cache_write = Field(default=0, description="Input tokens written to the provider's prompt cache.") class-attribute instance-attribute
calls = Field(default=0, description='Provider calls that returned.') class-attribute instance-attribute
input = Field(default=0, description='Input tokens of provider calls, cache reads included.') class-attribute instance-attribute
output = Field(default=0, description='Output tokens of provider calls, reasoning included.') class-attribute instance-attribute
reasoning = Field(default=0, description='Reasoning tokens of provider calls.') class-attribute instance-attribute
replayed_input = Field(default=0, description='Input tokens of responses replayed from disk cache.') class-attribute instance-attribute
replayed_output = Field(default=0, description='Output tokens of responses replayed from disk cache.') class-attribute instance-attribute
replays = Field(default=0, description='Responses replayed from disk cache.') class-attribute instance-attribute

Methods:

add(other)

Add every count of other to this one.

Source code in ontocast/onto/run_meter.py 
68
69
70
71
def add(self, other: TokenTotals) -> None:
    """Add every count of ``other`` to this one."""
    for name in TokenTotals.model_fields:
        setattr(self, name, getattr(self, name) + getattr(other, name))

Functions:

active_run_meter()

The meter bound to the running task, if any.

Source code in ontocast/onto/run_meter.py 
281
282
283
def active_run_meter() -> RunMeter | None:
    """The meter bound to the running task, if any."""
    return _active_run_meter.get()

run_meter_scope(*, max_input_tokens=None, max_output_tokens=None)

Bind a token-capped meter for one document unless one is already bound.

An already-bound meter is yielded unchanged and the caps are ignored: the caller that bound it owns the limits. With no meter bound and no cap set, yields None and meters nothing.

Parameters:

Name Type Description Default
max_input_tokens int | None

Cap on provider input tokens for the document.

 None
max_output_tokens int | None

Cap on provider output tokens for the document.

 None
Source code in ontocast/onto/run_meter.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
@contextmanager
def run_meter_scope(
    *,
    max_input_tokens: int | None = None,
    max_output_tokens: int | None = None,
) -> Iterator[RunMeter | None]:
    """Bind a token-capped meter for one document unless one is already bound.

    An already-bound meter is yielded unchanged and the caps are ignored: the
    caller that bound it owns the limits. With no meter bound and no cap set,
    yields None and meters nothing.

    Args:
        max_input_tokens: Cap on provider input tokens for the document.
        max_output_tokens: Cap on provider output tokens for the document.
    """
    outer = _active_run_meter.get()
    if outer is not None:
        yield outer
        return
    meter = RunMeter.from_token_limits(
        max_input_tokens=max_input_tokens, max_output_tokens=max_output_tokens
    )
    if meter is None:
        yield None
        return
    with use_run_meter(meter):
        yield meter

use_run_meter(meter)

Meter every provider call made inside this block with meter.

On exit the meter is flushed. If the meter refused a call inside the block and the block itself raised nothing, :class:BudgetExceededError is raised here, so a refusal that a best-effort step swallowed still stops the run.

Source code in ontocast/onto/run_meter.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
@contextmanager
def use_run_meter(meter: RunMeter) -> Iterator[RunMeter]:
    """Meter every provider call made inside this block with ``meter``.

    On exit the meter is flushed. If the meter refused a call inside the block
    and the block itself raised nothing, :class:`BudgetExceededError` is raised
    here, so a refusal that a best-effort step swallowed still stops the run.
    """
    token = _active_run_meter.set(meter)
    try:
        yield meter
    finally:
        _active_run_meter.reset(token)
        meter.flush()
    if meter.exceeded is not None:
        raise BudgetExceededError(
            f"Run budget exceeded: {meter.exceeded}", meter.snapshot()
        )