夜間バッチが勝手に上位モデルで走っていた — タスクごとにモデル上限を宣言して機械で守る

# 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"]
 
# 1M トークンあたりの単価（入力, 出力）。請求と突き合わせるための概算値
_PRICE_USD = {
    "haiku-4-5": (1.00, 5.00),
    "sonnet-5": (2.00, 10.00),   # 2026-08-31 までの導入価格
    "opus-4-8": (5.00, 25.00),
}
 
 
class EntitlementError(RuntimeError):
    """マニフェスト違反。握りつぶさず、必ず呼び出し元まで上げる。"""
 
 
@dataclass(frozen=True)
class Decision:
    task: str
    tier: str            # ladder 上の名前
    model: str           # 実際の API モデル文字列
    max_output_tokens: int
    degraded: bool
 
 
def _rank(tier: str) -> int:
    try:
        return _LADDER.index(tier)
    except ValueError as exc:
        raise EntitlementError(f"未知のモデル階梯: {tier}") from exc
 
 
def resolve(task: str, want: str | None = None) -> Decision:
    """タスク名と希望モデルから、実際に使ってよいモデルを決める。
 
    want を渡さなければ ceiling をそのまま使う。want が ceiling を超えるときは
    on_ceiling_miss に従い、degrade なら ceiling まで引き下げ、fail なら例外。
    """
    spec = _TASKS.get(task)
    if spec is None:
        # 既定拒否。マニフェストに無いタスクは 1 トークンも使わせない
        raise EntitlementError(f"未登録のタスク: {task}")
 
    ceiling, floor = spec["ceiling"], spec["floor"]
    if _rank(floor) > _rank(ceiling):
        raise EntitlementError(f"{task}: floor が 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} は上限 {ceiling} を超えています"
            )
        target, degraded = ceiling, True
 
    if _rank(target) < _rank(floor):
        # 下限割れは degrade では救わない。能力不足は静かに通してはいけない
        raise EntitlementError(f"{task}: {target} は下限 {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,
        # 応答が名乗ったモデル。ここが resolved と食い違ったら監査で拾う
        "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, ensure_ascii=False) + "\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']}: 解決 {expected} / 課金 {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"\n引き下げ発生: {degraded} 件 / {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)