tegwick
09aa1f3604
Add moonshotai/kimi-k3 as OpenRouter basemodel default after live smoke, USD→EUR cost conversion, append-only usage ledger, and CLI run/cost/spend week commands with token and euro reporting.
473 lines
15 KiB
Python
473 lines
15 KiB
Python
"""Append-only usage ledger for operator spend reporting.
|
||
|
||
Separate from :class:`~llm_connect.quality.QualityLedger`, which stores adaptive
|
||
routing quality signals. This ledger is for tokens + estimated cost accounting.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import contextvars
|
||
import json
|
||
import os
|
||
import threading
|
||
from contextlib import contextmanager
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Iterator, TextIO
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from llm_connect.costs import CostEstimate, estimate_cost
|
||
from llm_connect.fx import FxRate
|
||
from llm_connect.models import LLMResponse
|
||
from llm_connect.rates import ModelRateRegistry
|
||
|
||
|
||
ENV_USAGE_LEDGER = "LLM_CONNECT_USAGE_LEDGER"
|
||
ENV_TZ = "LLM_CONNECT_TZ"
|
||
DEFAULT_TZ = "Europe/Berlin"
|
||
|
||
_PATH_LOCKS: dict[Path, threading.Lock] = {}
|
||
_PATH_LOCKS_GUARD = threading.Lock()
|
||
_AUTO_RECORD_DISABLED: contextvars.ContextVar[bool] = contextvars.ContextVar(
|
||
"llm_connect_auto_record_disabled",
|
||
default=False,
|
||
)
|
||
|
||
|
||
@contextmanager
|
||
def suppress_auto_usage_record() -> Iterator[None]:
|
||
"""Disable env-based auto recording (e.g. while CLI records explicitly)."""
|
||
token = _AUTO_RECORD_DISABLED.set(True)
|
||
try:
|
||
yield
|
||
finally:
|
||
_AUTO_RECORD_DISABLED.reset(token)
|
||
|
||
|
||
def _utc_now() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _normalise_datetime(value: datetime | str) -> datetime:
|
||
if isinstance(value, datetime):
|
||
dt = value
|
||
elif isinstance(value, str):
|
||
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||
else:
|
||
raise TypeError(f"Expected datetime or ISO string, got {type(value).__name__}")
|
||
|
||
if dt.tzinfo is None:
|
||
return dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(timezone.utc)
|
||
|
||
|
||
def _serialise_datetime(value: datetime) -> str:
|
||
return _normalise_datetime(value).isoformat().replace("+00:00", "Z")
|
||
|
||
|
||
def _validate_non_negative_int(name: str, value: int) -> None:
|
||
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
|
||
raise ValueError(f"{name} must be a non-negative integer")
|
||
|
||
|
||
def _path_lock(path: Path) -> threading.Lock:
|
||
resolved = path.resolve()
|
||
with _PATH_LOCKS_GUARD:
|
||
lock = _PATH_LOCKS.get(resolved)
|
||
if lock is None:
|
||
lock = threading.Lock()
|
||
_PATH_LOCKS[resolved] = lock
|
||
return lock
|
||
|
||
|
||
def _lock_file(handle: TextIO) -> None:
|
||
if os.name == "nt":
|
||
import msvcrt
|
||
|
||
msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
|
||
else:
|
||
import fcntl
|
||
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
|
||
|
||
|
||
def _unlock_file(handle: TextIO) -> None:
|
||
if os.name == "nt":
|
||
import msvcrt
|
||
|
||
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
|
||
else:
|
||
import fcntl
|
||
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
||
|
||
|
||
@contextmanager
|
||
def _locked_file(path: Path, mode: str) -> Iterator[TextIO]:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
local_lock = _path_lock(path)
|
||
with local_lock:
|
||
with path.open(mode, encoding="utf-8") as handle:
|
||
_lock_file(handle)
|
||
try:
|
||
yield handle
|
||
finally:
|
||
_unlock_file(handle)
|
||
|
||
|
||
def default_usage_ledger_path(*, env: dict[str, str] | None = None) -> Path:
|
||
"""Resolve the default usage ledger path.
|
||
|
||
Order: ``LLM_CONNECT_USAGE_LEDGER`` → ``$XDG_DATA_HOME/llm-connect/usage.jsonl``
|
||
→ ``~/.local/share/llm-connect/usage.jsonl``.
|
||
"""
|
||
environ = env if env is not None else os.environ
|
||
explicit = environ.get(ENV_USAGE_LEDGER)
|
||
if explicit and str(explicit).strip():
|
||
return Path(str(explicit).strip()).expanduser()
|
||
xdg = environ.get("XDG_DATA_HOME")
|
||
if xdg and str(xdg).strip():
|
||
return Path(str(xdg).strip()).expanduser() / "llm-connect" / "usage.jsonl"
|
||
return Path.home() / ".local" / "share" / "llm-connect" / "usage.jsonl"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class UsageEvent:
|
||
"""One recorded LLM call for spend accounting."""
|
||
|
||
provider: str
|
||
model_id: str
|
||
prompt_tokens: int
|
||
completion_tokens: int
|
||
total_tokens: int
|
||
cost_usd: float | None
|
||
cost_eur: float | None
|
||
cost_source: str
|
||
source: str
|
||
fx_source: str | None = None
|
||
recorded_at: datetime = field(default_factory=_utc_now)
|
||
tags: dict[str, Any] = field(default_factory=dict)
|
||
|
||
def __post_init__(self) -> None:
|
||
for name in ("provider", "model_id", "source", "cost_source"):
|
||
if not str(getattr(self, name)).strip():
|
||
raise ValueError(f"{name} must be a non-empty string")
|
||
for name in ("prompt_tokens", "completion_tokens", "total_tokens"):
|
||
_validate_non_negative_int(name, getattr(self, name))
|
||
if self.cost_usd is not None and (
|
||
not isinstance(self.cost_usd, (int, float)) or float(self.cost_usd) < 0
|
||
):
|
||
raise ValueError("cost_usd must be a non-negative number or None")
|
||
if self.cost_eur is not None and (
|
||
not isinstance(self.cost_eur, (int, float)) or float(self.cost_eur) < 0
|
||
):
|
||
raise ValueError("cost_eur must be a non-negative number or None")
|
||
|
||
object.__setattr__(self, "provider", str(self.provider).strip())
|
||
object.__setattr__(self, "model_id", str(self.model_id).strip())
|
||
object.__setattr__(self, "source", str(self.source).strip())
|
||
object.__setattr__(self, "cost_source", str(self.cost_source).strip())
|
||
object.__setattr__(
|
||
self,
|
||
"cost_usd",
|
||
None if self.cost_usd is None else float(self.cost_usd),
|
||
)
|
||
object.__setattr__(
|
||
self,
|
||
"cost_eur",
|
||
None if self.cost_eur is None else float(self.cost_eur),
|
||
)
|
||
object.__setattr__(self, "recorded_at", _normalise_datetime(self.recorded_at))
|
||
object.__setattr__(self, "tags", dict(self.tags or {}))
|
||
if self.fx_source is not None:
|
||
object.__setattr__(self, "fx_source", str(self.fx_source))
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
"""Convert to a JSON-serialisable dictionary."""
|
||
return {
|
||
"provider": self.provider,
|
||
"model_id": self.model_id,
|
||
"prompt_tokens": self.prompt_tokens,
|
||
"completion_tokens": self.completion_tokens,
|
||
"total_tokens": self.total_tokens,
|
||
"cost_usd": self.cost_usd,
|
||
"cost_eur": self.cost_eur,
|
||
"cost_source": self.cost_source,
|
||
"fx_source": self.fx_source,
|
||
"source": self.source,
|
||
"recorded_at": _serialise_datetime(self.recorded_at),
|
||
"tags": dict(self.tags),
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: dict[str, Any]) -> "UsageEvent":
|
||
"""Create an event from a JSON-decoded dictionary."""
|
||
return cls(
|
||
provider=data["provider"],
|
||
model_id=data["model_id"],
|
||
prompt_tokens=int(data["prompt_tokens"]),
|
||
completion_tokens=int(data["completion_tokens"]),
|
||
total_tokens=int(data["total_tokens"]),
|
||
cost_usd=data.get("cost_usd"),
|
||
cost_eur=data.get("cost_eur"),
|
||
cost_source=data.get("cost_source") or "unknown",
|
||
fx_source=data.get("fx_source"),
|
||
source=data.get("source") or "library",
|
||
recorded_at=data.get("recorded_at", _utc_now()),
|
||
tags=data.get("tags") or {},
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class UsageSummary:
|
||
"""Aggregated tokens and cost over a time window."""
|
||
|
||
start: datetime
|
||
end: datetime
|
||
event_count: int
|
||
prompt_tokens: int
|
||
completion_tokens: int
|
||
total_tokens: int
|
||
cost_usd: float | None
|
||
cost_eur: float | None
|
||
unknown_cost_events: int = 0
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"start": _serialise_datetime(self.start),
|
||
"end": _serialise_datetime(self.end),
|
||
"event_count": self.event_count,
|
||
"prompt_tokens": self.prompt_tokens,
|
||
"completion_tokens": self.completion_tokens,
|
||
"total_tokens": self.total_tokens,
|
||
"cost_usd": self.cost_usd,
|
||
"cost_eur": self.cost_eur,
|
||
"unknown_cost_events": self.unknown_cost_events,
|
||
}
|
||
|
||
|
||
class UsageLedger:
|
||
"""Append-only JSONL store for :class:`UsageEvent` records."""
|
||
|
||
def __init__(self, path: str | Path):
|
||
self._path = Path(path)
|
||
|
||
@property
|
||
def path(self) -> Path:
|
||
"""Ledger file path."""
|
||
return self._path
|
||
|
||
def append(self, event: UsageEvent) -> None:
|
||
"""Append one event as a locked JSONL record."""
|
||
line = json.dumps(event.to_dict(), sort_keys=True, separators=(",", ":"))
|
||
with _locked_file(self._path, "a") as handle:
|
||
handle.write(line + "\n")
|
||
handle.flush()
|
||
os.fsync(handle.fileno())
|
||
|
||
def read_all(self) -> list[UsageEvent]:
|
||
"""Return all parseable events, skipping malformed lines."""
|
||
events, _ = self._read_with_malformed_count()
|
||
return events
|
||
|
||
def iter_events(self) -> Iterator[UsageEvent]:
|
||
"""Yield parseable events in file order."""
|
||
yield from self.read_all()
|
||
|
||
def sum_range(
|
||
self,
|
||
start: datetime,
|
||
end: datetime,
|
||
) -> UsageSummary:
|
||
"""Aggregate events in the half-open interval ``[start, end)``."""
|
||
start_utc = _normalise_datetime(start)
|
||
end_utc = _normalise_datetime(end)
|
||
if end_utc < start_utc:
|
||
raise ValueError("end must be >= start")
|
||
|
||
prompt_tokens = 0
|
||
completion_tokens = 0
|
||
total_tokens = 0
|
||
cost_usd_total = 0.0
|
||
cost_eur_total = 0.0
|
||
has_usd = False
|
||
has_eur = False
|
||
unknown_cost = 0
|
||
count = 0
|
||
|
||
for event in self.read_all():
|
||
if event.recorded_at < start_utc or event.recorded_at >= end_utc:
|
||
continue
|
||
count += 1
|
||
prompt_tokens += event.prompt_tokens
|
||
completion_tokens += event.completion_tokens
|
||
total_tokens += event.total_tokens
|
||
if event.cost_usd is None:
|
||
unknown_cost += 1
|
||
else:
|
||
has_usd = True
|
||
cost_usd_total += event.cost_usd
|
||
if event.cost_eur is not None:
|
||
has_eur = True
|
||
cost_eur_total += event.cost_eur
|
||
|
||
return UsageSummary(
|
||
start=start_utc,
|
||
end=end_utc,
|
||
event_count=count,
|
||
prompt_tokens=prompt_tokens,
|
||
completion_tokens=completion_tokens,
|
||
total_tokens=total_tokens,
|
||
cost_usd=cost_usd_total if has_usd else (0.0 if count == 0 else None),
|
||
cost_eur=cost_eur_total if has_eur else (0.0 if count == 0 else None),
|
||
unknown_cost_events=unknown_cost,
|
||
)
|
||
|
||
def _read_with_malformed_count(self) -> tuple[list[UsageEvent], int]:
|
||
if not self._path.is_file():
|
||
return [], 0
|
||
|
||
events: list[UsageEvent] = []
|
||
malformed = 0
|
||
with _locked_file(self._path, "r") as handle:
|
||
for line in handle:
|
||
if not line.strip():
|
||
continue
|
||
try:
|
||
events.append(UsageEvent.from_dict(json.loads(line)))
|
||
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
|
||
malformed += 1
|
||
return events, malformed
|
||
|
||
|
||
def week_window(
|
||
which: str = "current",
|
||
*,
|
||
now: datetime | None = None,
|
||
tz: str | None = None,
|
||
) -> tuple[datetime, datetime]:
|
||
"""Return ``(start, end)`` for a Monday-based spend window.
|
||
|
||
* ``current`` — Monday 00:00 local → *now*
|
||
* ``last`` — previous Monday 00:00 → this Monday 00:00 (full Mon–Sun)
|
||
|
||
Endpoints are timezone-aware in the requested local zone (default
|
||
Europe/Berlin). Comparisons in the ledger normalise to UTC.
|
||
"""
|
||
which_normalised = str(which).strip().lower()
|
||
if which_normalised not in {"current", "last"}:
|
||
raise ValueError("which must be 'current' or 'last'")
|
||
|
||
zone_name = (tz or os.environ.get(ENV_TZ) or DEFAULT_TZ).strip() or DEFAULT_TZ
|
||
zone = ZoneInfo(zone_name)
|
||
reference = now or datetime.now(zone)
|
||
if reference.tzinfo is None:
|
||
reference = reference.replace(tzinfo=zone)
|
||
else:
|
||
reference = reference.astimezone(zone)
|
||
|
||
start_of_this_week = reference.replace(
|
||
hour=0, minute=0, second=0, microsecond=0
|
||
) - timedelta(days=reference.weekday())
|
||
|
||
if which_normalised == "current":
|
||
return start_of_this_week, reference
|
||
|
||
end = start_of_this_week
|
||
start = end - timedelta(days=7)
|
||
return start, end
|
||
|
||
|
||
def event_from_response(
|
||
response: LLMResponse,
|
||
*,
|
||
provider: str,
|
||
source: str,
|
||
registry: ModelRateRegistry | None = None,
|
||
fx: FxRate | float | None = None,
|
||
tags: dict[str, Any] | None = None,
|
||
recorded_at: datetime | None = None,
|
||
) -> UsageEvent:
|
||
"""Build a :class:`UsageEvent` from an :class:`LLMResponse` and cost model."""
|
||
usage = response.usage or {}
|
||
prompt_tokens = int(usage.get("prompt_tokens") or 0)
|
||
completion_tokens = int(usage.get("completion_tokens") or 0)
|
||
total_tokens = int(usage.get("total_tokens") or (prompt_tokens + completion_tokens))
|
||
model_id = response.model or "unknown"
|
||
estimate = estimate_cost(
|
||
model_id,
|
||
prompt_tokens,
|
||
completion_tokens,
|
||
registry=registry,
|
||
fx=fx,
|
||
)
|
||
return UsageEvent(
|
||
provider=provider,
|
||
model_id=model_id,
|
||
prompt_tokens=prompt_tokens,
|
||
completion_tokens=completion_tokens,
|
||
total_tokens=total_tokens,
|
||
cost_usd=estimate.cost_usd,
|
||
cost_eur=estimate.cost_eur,
|
||
cost_source=estimate.cost_source,
|
||
fx_source=estimate.fx_source,
|
||
source=source,
|
||
recorded_at=recorded_at or _utc_now(),
|
||
tags=tags or {},
|
||
)
|
||
|
||
|
||
def maybe_record_usage(
|
||
response: LLMResponse,
|
||
*,
|
||
provider: str,
|
||
source: str,
|
||
tags: dict[str, Any] | None = None,
|
||
env: dict[str, str] | None = None,
|
||
registry: ModelRateRegistry | None = None,
|
||
fx: FxRate | float | None = None,
|
||
) -> UsageEvent | None:
|
||
"""Record usage when ``LLM_CONNECT_USAGE_LEDGER`` is set; otherwise no-op.
|
||
|
||
Returns the recorded event, or ``None`` when recording is disabled.
|
||
"""
|
||
if _AUTO_RECORD_DISABLED.get():
|
||
return None
|
||
environ = env if env is not None else os.environ
|
||
path_value = environ.get(ENV_USAGE_LEDGER)
|
||
if not path_value or not str(path_value).strip():
|
||
return None
|
||
event = event_from_response(
|
||
response,
|
||
provider=provider,
|
||
source=source,
|
||
registry=registry,
|
||
fx=fx,
|
||
tags=tags,
|
||
)
|
||
UsageLedger(Path(str(path_value).strip()).expanduser()).append(event)
|
||
return event
|
||
|
||
|
||
def format_cost_footer(
|
||
estimate: CostEstimate,
|
||
*,
|
||
prompt_tokens: int,
|
||
completion_tokens: int,
|
||
total_tokens: int,
|
||
model_id: str,
|
||
) -> str:
|
||
"""Human-readable one-line cost summary for CLI footers."""
|
||
tokens_part = (
|
||
f"tokens: in={prompt_tokens} out={completion_tokens} total={total_tokens}"
|
||
)
|
||
if estimate.cost_usd is None:
|
||
usd_part = "usd=unknown"
|
||
else:
|
||
usd_part = f"usd={estimate.cost_usd:.6f}"
|
||
if estimate.cost_eur is None:
|
||
eur_part = "eur=unknown"
|
||
else:
|
||
eur_part = f"eur={estimate.cost_eur:.6f}"
|
||
return f"cost[{model_id}]: {tokens_part} · {eur_part} · {usd_part} · source={estimate.cost_source}"
|