"""
observe — a local, append-only activity/cost ledger for :func:`llm.chat`.
Every :func:`best_engine_ai_helper.llm.chat` call already emits a small event
dict to any observer registered via :func:`llm.add_observer`; nothing
consumes it by default. This module is that consumer: call :func:`enable`
once (the CLI's ``activity`` command and the MCP server do this at startup;
the FastAPI app does NOT — ``GET /api/activity`` reads the ledger directly
instead, see :func:`api.activity`, so it works even under a plain
``uvicorn`` launch that never called ``enable()``) and every subsequent
call — local (always free) or cloud (a real paid
request through OpenAI/Mistral/OpenRouter/Together/Azure/Anthropic/Gemini,
see :mod:`best_engine_ai_helper.engine`'s ``mode: cloud``) — is appended to a
SQLite database at ``~/.best-engine-ai-helper/usage.db``.
The point is the "one company, several users" case: a shared machine or a
small internal server fields calls from more than one person, and someone
needs to answer "who is calling what, how often, and at what cost" without
standing up a separate telemetry stack. This ledger is local-only (no
network call, no third-party service) and off by default.
Cost is a **best-effort estimate**, not a provider-verified bill. Most
providers report real token usage per call; when that field is present it is
used directly, otherwise cost is estimated from character counts via a fixed
chars-per-token ratio (see :data:`_CHARS_PER_TOKEN`) against the paid price
table, read through :func:`cloud_catalog.load_cloud_catalog` so that a
``catalog update --cloud`` refresh reaches the bill and not only the ranking.
Local backends (Ollama, vLLM) always cost ``0.0`` — there is no paid API call
to price. A cloud-backend call for a model the table does not price gets
``cost_usd: None`` (unknown), never a fabricated number.
What a local ``0.0`` means is "nothing was billed", not "nothing was spent":
electricity, disk and the slower model's time are real and this ledger has no
way to measure them. :meth:`Ledger.counterfactual` answers the question that
*is* answerable from what is recorded — what the same workload would have cost
at a provider's list price.
Author
------
Warith Harchaoui <warithmetics@deraison.ai>
"""
from __future__ import annotations
import contextvars
import getpass
import os
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from . import llm as _llm
# ---------------------------------------------------------------------------
# Storage location
# ---------------------------------------------------------------------------
_USER_DIR = Path.home() / ".best-engine-ai-helper"
_DEFAULT_DB_PATH = _USER_DIR / "usage.db"
# No price-file path lives here. It used to, and having two modules each hold
# their own `_PRICING_PATH` is how the ledger ended up reading a narrower table
# than the ranker: one of the two learned to overlay the refresh cache and the
# other did not. `cloud_catalog` owns where the prices are and what counts as
# one; this module asks it (see `_load_pricing`).
# Backends that never incur a per-call charge: local inference, whatever the
# hardware. Any other backend name is treated as a paid API.
# `lightweight` is free in money and expensive in time: it runs locally off this
# machine's own disk, so there is no API call to price.
_FREE_BACKENDS = frozenset({"ollama", "vllm", "langchain", "lightweight"})
# Rough English-text approximation (OpenAI's own rule of thumb: ~4 characters
# per token). A real count from the provider's `usage` field is always
# preferred once available; this is the fallback until then.
_CHARS_PER_TOKEN = 4.0
# ---------------------------------------------------------------------------
# User attribution
# ---------------------------------------------------------------------------
# Scoped override for a single call or a `with` block (e.g. a web server
# handling one request per user); falls back to a process-wide override env
# var, then the OS login name -- the sensible default for a shared machine
# where each person has their own account.
_CURRENT_USER: contextvars.ContextVar[str | None] = contextvars.ContextVar(
"best_engine_current_user", default=None
)
[docs]
def current_user() -> str:
"""
Resolve the identity to attribute the next ledger entry to.
Precedence: :func:`as_user` scope > ``BEST_ENGINE_USER`` env var > OS
login name > ``"unknown"`` (only when even the OS refuses to say).
Returns
-------
str
The resolved user identity. Never empty.
Examples
--------
>>> isinstance(current_user(), str) and current_user() != ""
True
"""
scoped = _CURRENT_USER.get()
if scoped:
return scoped
env = os.environ.get("BEST_ENGINE_USER")
if env:
return env
try:
return getpass.getuser()
except Exception: # noqa: BLE001 — some sandboxes have no login name at all
return "unknown"
[docs]
class as_user:
"""
Context manager: attribute every ledger entry inside the block to ``name``.
Parameters
----------
name : str
Identity to record for calls made inside this block.
Examples
--------
>>> with as_user("alice"):
... current_user()
'alice'
>>> current_user() != "alice"
True
"""
def __init__(self, name: str) -> None:
self._name = name
self._token: contextvars.Token[str | None] | None = None
def __enter__(self) -> None:
self._token = _CURRENT_USER.set(self._name)
def __exit__(self, *exc_info: object) -> None:
if self._token is not None:
_CURRENT_USER.reset(self._token)
# ---------------------------------------------------------------------------
# Cost estimation
# ---------------------------------------------------------------------------
def _load_pricing() -> dict[str, dict[str, float]]:
"""
The paid price table, as ``{model_id: {input_per_1m, output_per_1m}}`` (USD).
Read through :func:`cloud_catalog.load_cloud_catalog`, not off
``pricing.yaml`` directly. The two are not the same table: the loader
overlays ``~/.best-engine-ai-helper/pricing_cache.yaml``, which is where
``catalog update --cloud`` writes the list prices it fetched and the models
it adopted. Reading the seed alone meant the ledger could not price the
very models a refresh had just fetched a price for — it recorded
``cost_usd: None`` ("nobody priced this") about a figure sitting in the
cache — while ``report --cloud``, two modules away, ranked them on it. One
table now, so a refreshed price reaches the bill as well as the ranking.
Returns
-------
dict
Empty dict when no catalog is readable — an unpriced catalog degrades
to "cost unknown" for every cloud call, never a crash.
Examples
--------
>>> table = _load_pricing()
>>> all({"input_per_1m", "output_per_1m"} <= set(v) for v in table.values())
True
"""
from . import cloud_catalog as _cloud_catalog
table: dict[str, dict[str, float]] = {}
for entry in _cloud_catalog.load_cloud_catalog():
model_id = str(entry.get("id") or "")
input_per_1m, output_per_1m = entry.get("input_per_1m"), entry.get("output_per_1m")
# A half-priced entry is as unusable as an unpriced one, and leaving it
# in with a missing side would bill its input at zero.
if not model_id or input_per_1m is None or output_per_1m is None:
continue
table[model_id] = {
"input_per_1m": float(input_per_1m),
"output_per_1m": float(output_per_1m),
}
return table
[docs]
def estimate_cost_usd(event: dict[str, Any]) -> float | None:
"""
Estimate the USD cost of one :func:`llm.chat` event.
Parameters
----------
event : dict
One event dict as emitted by :func:`llm.chat` (``backend``, ``model``,
``in_chars``, ``out_chars``, and — when the transport's provider
reports them — ``in_tokens``/``out_tokens``).
Returns
-------
float or None
``0.0`` for a local backend (always free). For a paid backend: the
cost from ``pricing.yaml`` using the provider's own reported token
counts when the event carries them (``in_tokens``/``out_tokens``,
exact), falling back to the character-count heuristic otherwise
(approximate — see :data:`_CHARS_PER_TOKEN`). None when the model is
not in the table — an unpriced model must never silently show as free.
Examples
--------
>>> estimate_cost_usd({"backend": "ollama", "model": "qwen3:8b",
... "in_chars": 100, "out_chars": 100})
0.0
>>> estimate_cost_usd({"backend": "openai", "model": "not-in-table",
... "in_chars": 100, "out_chars": 100}) is None
True
"""
backend = event.get("backend", "")
if backend in _FREE_BACKENDS:
return 0.0
pricing = _load_pricing().get(event.get("model", ""))
if not pricing:
return None
# Real provider-reported counts are exact; the char-count heuristic is a
# fallback for backends/events that don't carry them (e.g. LangChain, or
# any event recorded before a transport started reporting usage).
in_tokens = event.get("in_tokens")
out_tokens = event.get("out_tokens")
if in_tokens is None:
in_tokens = float(event.get("in_chars", 0)) / _CHARS_PER_TOKEN
if out_tokens is None:
out_tokens = float(event.get("out_chars", 0)) / _CHARS_PER_TOKEN
cost: float = float(in_tokens) / 1_000_000 * float(pricing.get("input_per_1m", 0.0)) + float(
out_tokens
) / 1_000_000 * float(pricing.get("output_per_1m", 0.0))
return round(cost, 6)
# ---------------------------------------------------------------------------
# Ledger
# ---------------------------------------------------------------------------
[docs]
class Ledger:
"""
Append-only SQLite sink for :func:`llm.chat` observer events.
Parameters
----------
db_path : str or Path or None
Where to store the database. Defaults to
``~/.best-engine-ai-helper/usage.db``; ``:memory:`` is accepted for
tests. The parent directory is created if missing.
"""
def __init__(self, db_path: str | Path | None = None) -> None:
self.db_path = str(db_path) if db_path is not None else str(_DEFAULT_DB_PATH)
if self.db_path != ":memory:":
Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
# `check_same_thread=False`: a FastAPI app may call `record` from a
# different thread than the one that constructed the Ledger; SQLite
# itself serializes writes, so this is safe for the low write volume
# a per-call ledger sees.
self._conn = sqlite3.connect(self.db_path, check_same_thread=False)
self._ensure_schema()
def _ensure_schema(self) -> None:
"""Create the ``calls`` table if it does not already exist."""
self._conn.execute(
"""
CREATE TABLE IF NOT EXISTS calls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
user TEXT NOT NULL,
backend TEXT NOT NULL,
model TEXT NOT NULL,
kind TEXT NOT NULL,
in_chars INTEGER NOT NULL,
images INTEGER NOT NULL,
out_chars INTEGER NOT NULL,
latency_ms REAL NOT NULL,
ok INTEGER NOT NULL,
error TEXT,
cost_usd REAL
)
"""
)
self._conn.commit()
[docs]
def record(self, event: dict[str, Any]) -> None:
"""
Persist one :func:`llm.chat` event, enriched with user and cost.
Parameters
----------
event : dict
One event dict as emitted by :func:`llm.chat`.
"""
self._conn.execute(
"""
INSERT INTO calls
(ts, user, backend, model, kind, in_chars, images, out_chars,
latency_ms, ok, error, cost_usd)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
datetime.now(timezone.utc).isoformat(),
current_user(),
event.get("backend", ""),
event.get("model", ""),
event.get("kind", ""),
event.get("in_chars", 0),
event.get("images", 0),
event.get("out_chars", 0),
event.get("latency_ms", 0.0),
1 if event.get("ok") else 0,
event.get("error"),
estimate_cost_usd(event),
),
)
self._conn.commit()
[docs]
def summary(self) -> dict[str, Any]:
"""
Aggregate the ledger into the figures a usage dashboard needs.
Returns
-------
dict
``total_calls``, ``total_cost_usd`` (``0.0`` for an empty ledger,
None if any recorded call has an unpriced model — an unknown
component must never be silently dropped from the total),
``error_rate`` (0-1), ``by_user`` / ``by_model`` (call count +
cost per key, sorted by call count descending), and
``recent_errors`` (last 10 failed calls: ts, user, model, error).
"""
cur = self._conn.cursor()
total_calls = cur.execute("SELECT COUNT(*) FROM calls").fetchone()[0]
total_cost = cur.execute(
"SELECT SUM(cost_usd) FROM calls WHERE cost_usd IS NOT NULL"
).fetchone()[0]
has_unpriced_paid_call = cur.execute(
"SELECT COUNT(*) FROM calls WHERE cost_usd IS NULL"
).fetchone()[0]
errors = cur.execute("SELECT COUNT(*) FROM calls WHERE ok = 0").fetchone()[0]
def _grouped(column: str) -> list[dict[str, Any]]:
rows = cur.execute(
f"SELECT {column}, COUNT(*), SUM(cost_usd) FROM calls " # noqa: S608 — column is one of two fixed literals below, never user input
f"GROUP BY {column} ORDER BY COUNT(*) DESC"
).fetchall()
return [{column: key, "calls": count, "cost_usd": cost} for key, count, cost in rows]
recent_errors = cur.execute(
"SELECT ts, user, model, error FROM calls WHERE ok = 0 ORDER BY id DESC LIMIT 10"
).fetchall()
if total_calls == 0:
total_cost_usd: float | None = 0.0
elif has_unpriced_paid_call:
total_cost_usd = None
else:
total_cost_usd = total_cost
return {
"total_calls": total_calls,
"total_cost_usd": total_cost_usd,
"error_rate": round(errors / total_calls, 4) if total_calls else 0.0,
"by_user": _grouped("user"),
"by_model": _grouped("model"),
"recent_errors": [
{"ts": ts, "user": user, "model": model, "error": error}
for ts, user, model, error in recent_errors
],
}
[docs]
def counterfactual(self) -> dict[str, Any]:
"""
What the **free** calls in this ledger would have cost at list price.
The field around this project is full of local-versus-cloud break-even
calculators: type in a guessed monthly token volume, read back a
payback period. The guess is the weak part — nobody knows their volume
before they have run the workload, and the number that comes out
inherits whatever was typed in.
This ledger already holds the workload, measured. Crossing it with the
same price table :func:`estimate_cost_usd` bills paid calls against
turns the question around: not "what might you spend", but **"what did
the calls you already made avoid"**. Two sources, both already here,
and no invented volume.
Three things it deliberately does not claim:
* **It is not a saving.** Running locally costs electricity, disk and
the time a slower model takes. This figure is an avoided *bill*, not
a profit, and the ledger has no business inventing a watt.
* **It is not the same answer.** A paid model given the same prompts
would not emit the same tokens, so the output volume is this
machine's, carried across as the closest thing to a workload anyone
has measured.
* **The token counts are approximate.** A local backend reports no
``usage`` block, so the characters recorded per call go through
:data:`_CHARS_PER_TOKEN` exactly as they do for an unreported paid
call.
Returns
-------
dict
``free_calls``, ``in_tokens``, ``out_tokens`` (both estimated),
and ``models``: one row per priced catalog entry
(``model``, ``provider``, ``cost_usd``), cheapest first. Empty
``models`` when nothing in the catalog carries a price.
Examples
--------
>>> led = Ledger(":memory:")
>>> led.record({"backend": "ollama", "model": "qwen3:8b", "kind": "llm",
... "in_chars": 4000, "out_chars": 4000, "latency_ms": 1.0,
... "ok": True})
>>> what_if = led.counterfactual()
>>> what_if["free_calls"], what_if["in_tokens"], what_if["out_tokens"]
(1, 1000, 1000)
>>> cheapest = what_if["models"][0]
>>> cheapest["cost_usd"] >= 0.0
True
>>> led.close()
An empty ledger has nothing to price, and says so with zeros rather
than with a catalog of zero-cost rows:
>>> empty = Ledger(":memory:")
>>> empty.counterfactual()["models"]
[]
>>> empty.close()
"""
row = self._conn.execute(
"SELECT COUNT(*), SUM(in_chars), SUM(out_chars) FROM calls "
"WHERE ok = 1 AND cost_usd = 0.0"
).fetchone()
free_calls = int(row[0] or 0)
in_tokens = int((row[1] or 0) / _CHARS_PER_TOKEN)
out_tokens = int((row[2] or 0) / _CHARS_PER_TOKEN)
models: list[dict[str, Any]] = []
if free_calls:
from . import cloud_catalog as _cloud_catalog
for entry in _cloud_catalog.load_cloud_catalog():
input_per_1m, output_per_1m = entry.get("input_per_1m"), entry.get("output_per_1m")
if input_per_1m is None or output_per_1m is None:
continue
models.append(
{
"model": str(entry.get("id") or "?"),
"provider": str(entry.get("provider") or "?"),
"cost_usd": round(
in_tokens / 1_000_000 * float(input_per_1m)
+ out_tokens / 1_000_000 * float(output_per_1m),
6,
),
}
)
models.sort(key=lambda m: (m["cost_usd"], m["model"]))
return {
"free_calls": free_calls,
"in_tokens": in_tokens,
"out_tokens": out_tokens,
"models": models,
}
[docs]
def close(self) -> None:
"""Close the underlying SQLite connection."""
self._conn.close()
# ---------------------------------------------------------------------------
# Module-level enable/disable
# ---------------------------------------------------------------------------
_active_ledger: Ledger | None = None
[docs]
def enable(db_path: str | Path | None = None) -> Ledger:
"""
Start recording every :func:`llm.chat` call to a ledger.
Idempotent: calling this again returns the already-active ledger rather
than registering a second observer.
Parameters
----------
db_path : str or Path or None
Forwarded to :class:`Ledger`. Ignored if a ledger is already active.
Returns
-------
Ledger
The active ledger, for direct querying (:meth:`Ledger.summary`).
"""
global _active_ledger
if _active_ledger is None:
_active_ledger = Ledger(db_path)
_llm.add_observer(_active_ledger.record)
return _active_ledger
[docs]
def disable() -> None:
"""
Stop recording and close the active ledger, if any.
Clears every registered :mod:`llm` observer, not only this module's —
today the ledger is the only observer type in the suite, so this is
equivalent in practice, but a caller with its own observers registered
separately would need to re-register them after calling this.
"""
global _active_ledger
_llm.clear_observers()
if _active_ledger is not None:
_active_ledger.close()
_active_ledger = None
[docs]
def is_enabled() -> bool:
"""Return whether a ledger is currently recording."""
return _active_ledger is not None
[docs]
def active_ledger() -> Ledger | None:
"""Return the active ledger, or None if :func:`enable` was never called."""
return _active_ledger