claude-opus-4-7 の speed:fast が 7/24 で失効する — 能力ペア単位で守る preflight

from dataclasses import dataclass
from datetime import date
from typing import Optional
 
@dataclass(frozen=True)
class Capability:
    """モデルと速度指定の組み合わせ（=能力ペア）と、その失効期日。"""
    model: str
    speed: Optional[str]          # None は速度指定なし
    retires_on: Optional[date]    # None は失効予定なし
    successor: Optional["Capability"] = None  # 移行先
 
# 能力ペアのレジストリ。事実（期日・移行先）を1か所に集約する
OPUS_48_FAST = Capability("claude-opus-4-8", "fast", None)
 
CAPABILITIES = {
    ("claude-opus-4-7", "fast"): Capability(
        model="claude-opus-4-7",
        speed="fast",
        retires_on=date(2026, 7, 24),
        successor=OPUS_48_FAST,
    ),
    ("claude-opus-4-7", None): Capability("claude-opus-4-7", None, None),
    ("claude-opus-4-8", "fast"): OPUS_48_FAST,
}
 
def lookup(model: str, speed: Optional[str]) -> Optional[Capability]:
    return CAPABILITIES.get((model, speed))

from datetime import date
 
def check_deadline(model: str, speed: Optional[str],
                   today: Optional[date] = None,
                   warn_within_days: int = 14) -> None:
    """能力ペアの失効期日を確認する。期日超過は例外、接近は警告。"""
    today = today or date.today()
    cap = lookup(model, speed)
    if cap is None:
        # レジストリ外＝未知の組み合わせ。無人運用では明示的に止める
        raise RuntimeError(f"未登録の能力ペアです: model={model}, speed={speed}")
    if cap.retires_on is None:
        return  # 失効予定なし
    if today >= cap.retires_on:
        raise RuntimeError(
            f"{model} + speed={speed} は {cap.retires_on} に失効済みです。"
            f"移行先: {cap.successor.model} + speed={cap.successor.speed}"
        )
    remaining = (cap.retires_on - today).days
    if remaining <= warn_within_days:
        print(f"⚠️ {model} + speed={speed} は残り {remaining} 日で失効します "
              f"（{cap.retires_on}）。移行先: {cap.successor.model}")
 
# 使用例（2026-07-20 に実行したと仮定）
check_deadline("claude-opus-4-7", "fast", today=date(2026, 7, 20))
# → ⚠️ claude-opus-4-7 + speed=fast は残り 4 日で失効します（2026-07-24）。移行先: claude-opus-4-8

import anthropic
 
client = anthropic.Anthropic()
 
def probe_capability(model: str, speed: Optional[str]) -> bool:
    """本処理と同じ能力ペアで最小リクエストを投げ、生死を確かめる。"""
    kwargs = {
        "model": model,
        "max_tokens": 1,                       # 課金を最小化する
        "messages": [{"role": "user", "content": "ok"}],
    }
    if speed is not None:
        kwargs["speed"] = speed
    try:
        client.messages.create(**kwargs)
        return True
    except anthropic.BadRequestError:
        # 速度指定が無効になった場合はここに落ちる（400系）
        return False
    except anthropic.APIStatusError as e:
        # 一時的な 5xx 等は「不明」として扱い、呼び出し側で再試行させる
        raise RuntimeError(f"プローブ判定不能（一時障害の可能性）: {e}") from e
 
# max_tokens=1 のプローブは、本処理を1回走らせるより桁違いに安い

def resolve_capability(model: str, speed: Optional[str]) -> tuple[str, Optional[str]]:
    """使うべき能力ペアを解決する。失効していれば移行先を返す。"""
    cap = lookup(model, speed)
    if cap is None:
        raise RuntimeError(f"未登録の能力ペア: {model}/{speed}")
 
    # まず静的な期日で判定（プローブより速く・無料）
    if cap.retires_on and date.today() >= cap.retires_on:
        if cap.successor is None:
            raise RuntimeError(f"{model}/{speed} は失効済みだが移行先が未定義です")
        print(f"↪ 期日超過につき移行: {model}/{speed} → "
              f"{cap.successor.model}/{cap.successor.speed}")
        return cap.successor.model, cap.successor.speed
 
    # 期日前でも、実プローブが落ちたら前倒し失効とみなす
    if not probe_capability(cap.model, cap.speed):
        if cap.successor is None:
            raise RuntimeError(f"{model}/{speed} が失効。移行先が未定義です")
        print(f"↪ プローブ失敗につき移行: {model}/{speed} → "
              f"{cap.successor.model}/{cap.successor.speed}")
        return cap.successor.model, cap.successor.speed
 
    return cap.model, cap.speed
 
# パイプラインの入口で1回だけ解決する
MODEL, SPEED = resolve_capability("claude-opus-4-7", "fast")
# 以降の本処理は MODEL / SPEED を使う

import time
 
def measure(model: str, speed: Optional[str], prompt: str, n: int = 5) -> dict:
    """移行前後の比較用に、レイテンシと出力トークンを軽く実測する。"""
    latencies, out_tokens = [], []
    for _ in range(n):
        kwargs = {"model": model, "max_tokens": 256,
                  "messages": [{"role": "user", "content": prompt}]}
        if speed is not None:
            kwargs["speed"] = speed
        t0 = time.perf_counter()
        resp = client.messages.create(**kwargs)
        latencies.append(time.perf_counter() - t0)
        out_tokens.append(resp.usage.output_tokens)
    latencies.sort()
    return {
        "p50_latency_s": round(latencies[n // 2], 3),
        "avg_output_tokens": round(sum(out_tokens) / n, 1),
    }
 
# 移行前後で同じプロンプトを流し、閾値の再調整が必要か判断する