1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Append-only JSONL cost ledger with a run-level spend cap. Never stores prompts or credentials."""3from __future__ import annotations45import json6import threading7import time8from pathlib import Path910from ..config import RESULTS, max_spend_usd, model_info111213class SpendCapExceeded(RuntimeError):14 pass151617class CostLedger:18 def __init__(self, path: Path | None = None, cap_usd: float | None = None, run_id: str = "adhoc"):19 self.path = path or (RESULTS / "ledger" / f"{run_id}.jsonl")20 self.path.parent.mkdir(parents=True, exist_ok=True)21 self.cap = cap_usd if cap_usd is not None else max_spend_usd()22 self.run_id = run_id23 self._lock = threading.Lock()24 self.total_usd = 0.025 self.calls = 026 if self.path.exists():27 for line in self.path.read_text().splitlines():28 try:29 self.total_usd += float(json.loads(line).get("cost_usd", 0.0))30 self.calls += 131 except Exception:32 pass3334 @staticmethod35 def estimate_cost(model: str, input_tokens: int, output_tokens: int, cached_tokens: int = 0) -> float:36 try:37 m = model_info(model)38 except KeyError:39 return 0.040 uncached = max(input_tokens - cached_tokens, 0)41 return (42 uncached * m["input_per_m"] + cached_tokens * m.get("cached_input_per_m", m["input_per_m"]) + output_tokens * m["output_per_m"]43 ) / 1e64445 def record(self, **rec) -> None:46 with self._lock:47 rec.setdefault("ts", time.time())48 rec["run_id"] = self.run_id49 self.total_usd += float(rec.get("cost_usd", 0.0))50 self.calls += 151 with open(self.path, "a", encoding="utf-8") as fh:52 fh.write(json.dumps(rec, ensure_ascii=False) + "\n")53 if self.total_usd > self.cap:54 raise SpendCapExceeded(f"run {self.run_id}: estimated spend {self.total_usd:.2f} USD exceeds cap {self.cap:.2f}")55