Opus 4.7 の fast mode 廃止（7/24）で無人ジョブを落とさない — speed パラメータの事前検査と移行

# safe_speed.py — 廃止済みの speed 指定を実行時に無害化するガード
import datetime as dt
import logging
 
logger = logging.getLogger("model_guard")
 
# Opus 4.7 fast mode の廃止日（UTC 基準で保守的に扱う）
FAST_MODE_SUNSET = dt.date(2026, 7, 24)
 
def resolve_call_options(model: str, speed: str | None,
                         today: dt.date | None = None) -> dict:
    """model と speed を安全な組み合わせに正規化して返す。
 
    - 廃止日以降に opus-4-7 + fast が来たら、fast を落として続行する
    - 判断はすべてログに残し、無言のフォールバックにしない
    """
    today = today or dt.date.today()
    opts = {"model": model}
 
    wants_fast = speed == "fast"
    is_legacy_fast = "opus-4-7" in model and wants_fast
 
    if is_legacy_fast and today >= FAST_MODE_SUNSET:
        # 期限切れの組み合わせ: fast を外して通常速度で続行
        logger.warning(
            "opus-4-7 fast mode は %s に廃止済み。speed 指定を落として続行します "
            "（呼び出し元の移行が未完了です）", FAST_MODE_SUNSET.isoformat())
        return opts  # speed を付けない
 
    if speed is not None:
        opts["speed"] = speed
    return opts
 
# 使い方
opts = resolve_call_options("claude-opus-4-7", "fast")
# opts == {"model": "claude-opus-4-7"}  # 廃止日以降は fast が外れる
response = client.messages.create(max_tokens=1024, messages=[...], **opts)

# guarded_client.py — messages.create を横断的に守るラッパー
class GuardedMessages:
    def __init__(self, inner):
        self._inner = inner
 
    def create(self, *, model, speed=None, **kwargs):
        opts = resolve_call_options(model, speed)  # 上のガードを再利用
        return self._inner.create(**opts, **kwargs)
 
class GuardedClient:
    """既存クライアントと同じ .messages.create インターフェースを保つ。"""
    def __init__(self, inner):
        self._inner = inner
        self.messages = GuardedMessages(inner.messages)
 
# 既存コードは client を差し替えるだけでよい
client = GuardedClient(Anthropic())

# measure.py — 移行前後を同一プロンプトで比較する簡易計測
import time
 
def measure(client, model, speed, prompt, runs=5):
    latencies, out_tokens = [], []
    for _ in range(runs):
        start = time.perf_counter()
        opts = {"model": model}
        if speed:
            opts["speed"] = speed
        r = client.messages.create(max_tokens=512,
                                    messages=[{"role": "user", "content": prompt}],
                                    **opts)
        latencies.append(time.perf_counter() - start)
        out_tokens.append(r.usage.output_tokens)
    latencies.sort()
    p50 = latencies[len(latencies) // 2]
    return {"p50_sec": round(p50, 2),
            "avg_out_tokens": sum(out_tokens) // len(out_tokens)}
 
# 移行前後を同じプロンプトで測って並べる
prompt = "次の変更履歴を3行で要約してください: ..."
# before = measure(old_client, "claude-opus-4-7", "fast", prompt)
# after  = measure(new_client, "claude-opus-4-8", "fast", prompt)