My Nightly Batch Was Quietly Running on a Bigger Model — Declaring Per-Task Model Ceilings and Enforcing Them

# entitlements.py
from __future__ import annotations
 
import json
import os
import time
from dataclasses import dataclass
 
import yaml
from anthropic import Anthropic
 
_CFG = yaml.safe_load(open("config/model_entitlements.yaml", encoding="utf-8"))
_LADDER: list[str] = _CFG["_ladder"]
_ALIASES: dict[str, str] = _CFG["_aliases"]
_TASKS: dict[str, dict] = _CFG["tasks"]
 
# USD per 1M tokens (input, output). Approximate, for reconciling against the bill.
_PRICE_USD = {
    "haiku-4-5": (1.00, 5.00),
    "sonnet-5": (2.00, 10.00),   # introductory pricing through 2026-08-31
    "opus-4-8": (5.00, 25.00),
}
 
 
class EntitlementError(RuntimeError):
    """A manifest violation. Never swallow this — let it reach the caller."""
 
 
@dataclass(frozen=True)
class Decision:
    task: str
    tier: str            # name on the ladder
    model: str           # concrete API model string
    max_output_tokens: int
    degraded: bool
 
 
def _rank(tier: str) -> int:
    try:
        return _LADDER.index(tier)
    except ValueError as exc:
        raise EntitlementError(f"unknown tier: {tier}") from exc
 
 
def resolve(task: str, want: str | None = None) -> Decision:
    """Decide which model a task is actually allowed to use.
 
    Omit `want` to take the ceiling as-is. If `want` exceeds the ceiling,
    on_ceiling_miss decides: `degrade` clamps down, `fail` raises.
    """
    spec = _TASKS.get(task)
    if spec is None:
        # Deny by default. An unregistered task spends zero tokens.
        raise EntitlementError(f"unregistered task: {task}")
 
    ceiling, floor = spec["ceiling"], spec["floor"]
    if _rank(floor) > _rank(ceiling):
        raise EntitlementError(f"{task}: floor is above ceiling")
 
    target = want or ceiling
    degraded = False
 
    if _rank(target) > _rank(ceiling):
        if spec.get("on_ceiling_miss", "fail") == "fail":
            raise EntitlementError(f"{task}: {target} exceeds ceiling {ceiling}")
        target, degraded = ceiling, True
 
    if _rank(target) < _rank(floor):
        # A floor breach is never repaired by degrading. Capability loss
        # must not pass silently.
        raise EntitlementError(f"{task}: {target} is below floor {floor}")
 
    return Decision(
        task=task,
        tier=target,
        model=_ALIASES[target],
        max_output_tokens=int(spec["max_output_tokens"]),
        degraded=degraded,
    )

_client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
_LEDGER = os.environ.get("ENTITLEMENT_LEDGER", "logs/entitlement.jsonl")
 
 
def _estimate_usd(tier: str, usage) -> float:
    pin, pout = _PRICE_USD[tier]
    return (usage.input_tokens * pin + usage.output_tokens * pout) / 1_000_000
 
 
def call(task: str, messages: list[dict], want: str | None = None, **kw):
    d = resolve(task, want)
    started = time.monotonic()
 
    resp = _client.messages.create(
        model=d.model,
        max_tokens=d.max_output_tokens,
        messages=messages,
        **kw,
    )
 
    record = {
        "ts": time.time(),
        "task": d.task,
        "declared_ceiling": _TASKS[d.task]["ceiling"],
        "resolved_tier": d.tier,
        # What the response claims to be. If this diverges from resolved,
        # the audit will catch it.
        "billed_model": resp.model,
        "degraded": d.degraded,
        "input_tokens": resp.usage.input_tokens,
        "output_tokens": resp.usage.output_tokens,
        "est_usd": round(_estimate_usd(d.tier, resp.usage), 6),
        "elapsed_ms": round((time.monotonic() - started) * 1000),
    }
    os.makedirs(os.path.dirname(_LEDGER), exist_ok=True)
    with open(_LEDGER, "a", encoding="utf-8") as fp:
        fp.write(json.dumps(record) + "\n")
 
    return resp

# audit_entitlements.py
import collections, datetime, json, sys
 
from entitlements import _TASKS, _ALIASES
 
WINDOW_H = 24
now = datetime.datetime.now().timestamp()
rows = [json.loads(l) for l in open(sys.argv[1], encoding="utf-8") if l.strip()]
rows = [r for r in rows if now - r["ts"] < WINDOW_H * 3600]
 
spend = collections.defaultdict(float)
violations: list[str] = []
 
for r in rows:
    spend[r["task"]] += r["est_usd"]
 
    expected = _ALIASES[r["resolved_tier"]]
    if not r["billed_model"].startswith(expected.split("-2025")[0]):
        violations.append(
            f"{r['task']}: resolved {expected} / billed {r['billed_model']}"
        )
 
for task, usd in sorted(spend.items(), key=lambda kv: -kv[1]):
    cap = _TASKS[task]["daily_usd"]
    pct = usd / cap * 100
    flag = "OVER" if usd > cap else ("WARN" if pct >= 80 else "ok")
    print(f"{flag:4}  {task:28}  ${usd:6.3f} / ${cap:5.2f}  ({pct:5.1f}%)")
 
degraded = sum(1 for r in rows if r["degraded"])
if degraded:
    print(f"\ndegraded calls: {degraded} of {len(rows)}")
 
for v in violations:
    print(f"VIOLATION  {v}")
 
sys.exit(1 if violations or any(spend[t] > _TASKS[t]["daily_usd"] for t in spend) else 0)