claude-opus-4-7 Fast Mode Retires on 7/24 — Guard It at the Capability-Pair Level

from dataclasses import dataclass
from datetime import date
from typing import Optional
 
@dataclass(frozen=True)
class Capability:
    """A model + speed combination (a capability pair) and its retirement date."""
    model: str
    speed: Optional[str]          # None means no speed option
    retires_on: Optional[date]    # None means no scheduled retirement
    successor: Optional["Capability"] = None  # migration target
 
# Registry of capability pairs. Facts (dates, targets) live in one place.
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:
    """Check a pair's retirement date. Past due raises; approaching warns."""
    today = today or date.today()
    cap = lookup(model, speed)
    if cap is None:
        # Not in the registry = unknown combination. Stop explicitly when unattended.
        raise RuntimeError(f"Unregistered capability pair: model={model}, speed={speed}")
    if cap.retires_on is None:
        return  # no scheduled retirement
    if today >= cap.retires_on:
        raise RuntimeError(
            f"{model} + speed={speed} retired on {cap.retires_on}. "
            f"Migrate to: {cap.successor.model} + speed={cap.successor.speed}"
        )
    remaining = (cap.retires_on - today).days
    if remaining <= warn_within_days:
        print(f"⚠️ {model} + speed={speed} retires in {remaining} days "
              f"({cap.retires_on}). Target: {cap.successor.model}")
 
# Example (assume it runs on 2026-07-20)
check_deadline("claude-opus-4-7", "fast", today=date(2026, 7, 20))
# → ⚠️ claude-opus-4-7 + speed=fast retires in 4 days (2026-07-24). Target: claude-opus-4-8

import anthropic
 
client = anthropic.Anthropic()
 
def probe_capability(model: str, speed: Optional[str]) -> bool:
    """Send a minimal request with the same pair as the main work to test it."""
    kwargs = {
        "model": model,
        "max_tokens": 1,                       # minimize the cost
        "messages": [{"role": "user", "content": "ok"}],
    }
    if speed is not None:
        kwargs["speed"] = speed
    try:
        client.messages.create(**kwargs)
        return True
    except anthropic.BadRequestError:
        # An invalid speed option lands here (4xx)
        return False
    except anthropic.APIStatusError as e:
        # Treat transient 5xx as "unknown" and let the caller retry
        raise RuntimeError(f"Probe inconclusive (possible transient failure): {e}") from e
 
# A max_tokens=1 probe is orders of magnitude cheaper than one real run

def resolve_capability(model: str, speed: Optional[str]) -> tuple[str, Optional[str]]:
    """Resolve the pair to use. Return the successor if it has retired."""
    cap = lookup(model, speed)
    if cap is None:
        raise RuntimeError(f"Unregistered capability pair: {model}/{speed}")
 
    # Judge by the static deadline first (faster and free versus a probe)
    if cap.retires_on and date.today() >= cap.retires_on:
        if cap.successor is None:
            raise RuntimeError(f"{model}/{speed} retired but no successor defined")
        print(f"↪ Past deadline, migrating: {model}/{speed} → "
              f"{cap.successor.model}/{cap.successor.speed}")
        return cap.successor.model, cap.successor.speed
 
    # Even before the deadline, a failing probe means an early retirement
    if not probe_capability(cap.model, cap.speed):
        if cap.successor is None:
            raise RuntimeError(f"{model}/{speed} retired; no successor defined")
        print(f"↪ Probe failed, migrating: {model}/{speed} → "
              f"{cap.successor.model}/{cap.successor.speed}")
        return cap.successor.model, cap.successor.speed
 
    return cap.model, cap.speed
 
# Resolve once at the pipeline entrance
MODEL, SPEED = resolve_capability("claude-opus-4-7", "fast")
# The rest of the work uses MODEL / SPEED

import time
 
def measure(model: str, speed: Optional[str], prompt: str, n: int = 5) -> dict:
    """A light measurement of latency and output tokens for a before/after compare."""
    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),
    }
 
# Run the same prompt before and after to decide whether to retune thresholds