Claude Codeで429のRetry-Afterを守るHTTPクライアントを書く — バックオフをテストで固める

Retry-After: 120
Retry-After: Fri, 31 Dec 1999 23:59:59 GMT

## 外部APIクライアントの規約
- 再試行は httpx + tenacity。自前の while ループと time.sleep は書かない
- Retry-After（秒数・HTTP日付の両形式）があれば最優先で使う
- 1回の待ちには上限（既定 30 秒）を付け、Retry-After が上限を超えたら待たず例外にする
- ヘッダーなしの待ちは指数バックオフ + フルジッタ。試行回数にも上限を付ける
- sleep は引数で差し替え可能にし、テストは実時間を待たない
- 完了条件: モック転送で 429 連続・5xx・タイムアウトを再現する pytest が全件通ること

retry_client.py に、httpx.Client を包む RetryingClient を書いて。
CLAUDE.md の規約に従い、test_retry_client.py も同時に作って、pytest が通るまで直して。

from __future__ import annotations
 
import time
from datetime import timezone
from email.utils import parsedate_to_datetime
 
import httpx
from tenacity import (
    Retrying,
    retry_if_exception,
    stop_after_attempt,
    wait_random_exponential,
)
 
RETRYABLE_STATUS = {429, 500, 502, 503, 504}
 
 
class RetryableStatus(Exception):
    def __init__(self, response, retry_after):
        super().__init__(f"HTTP {response.status_code}")
        self.response = response
        self.retry_after = retry_after
 
 
class RetryAfterTooLong(Exception):
    """Retry-After が上限を超えるので待たずに諦める。"""
 
 
def parse_retry_after(value, now=None):
    if not value:
        return None
    value = value.strip()
    if value.isdigit():
        return float(value)
    try:
        when = parsedate_to_datetime(value)
    except (TypeError, ValueError):
        return None
    if when.tzinfo is None:
        when = when.replace(tzinfo=timezone.utc)
    now = time.time() if now is None else now
    return max(0.0, when.timestamp() - now)
 
 
class RetryingClient:
    def __init__(self, client, *, max_attempts=5, base=0.5,
                 max_wait=30.0, sleep=time.sleep):
        self.client, self.max_attempts = client, max_attempts
        self.max_wait, self.sleep = max_wait, sleep
        self._backoff = wait_random_exponential(
            multiplier=base, max=max_wait)
 
    def _wait(self, state):
        exc = state.outcome.exception()
        if isinstance(exc, RetryableStatus) and exc.retry_after is not None:
            return exc.retry_after
        return self._backoff(state)
 
    def _retryable(self, exc):
        if isinstance(exc, RetryableStatus):
            return exc.retry_after is None or exc.retry_after <= self.max_wait
        return isinstance(exc, httpx.TransportError)
 
    def request(self, method, url, **kw):
        def once():
            resp = self.client.request(method, url, **kw)
            if resp.status_code in RETRYABLE_STATUS:
                ra = parse_retry_after(resp.headers.get("Retry-After"))
                if ra is not None and ra > self.max_wait:
                    raise RetryAfterTooLong(
                        f"Retry-After {ra:.0f}s > {self.max_wait}s")
                raise RetryableStatus(resp, ra)
            return resp
 
        return Retrying(
            retry=retry_if_exception(self._retryable),
            wait=self._wait,
            stop=stop_after_attempt(self.max_attempts),
            sleep=self.sleep,
            reraise=True,
        )(once)

def seq(*responses):
    it = iter(responses)
    calls = []
    def handler(req):
        calls.append(req)
        r = next(it)
        if isinstance(r, Exception):
            raise r
        return r
    return handler, calls

def test_retry_after_seconds_is_used_as_is():
    h, calls = seq(httpx.Response(429, headers={"Retry-After": "3"}),
                   httpx.Response(200))
    c, sleeps = make(h)
    assert c.request("GET", "http://x/").status_code == 200
    assert sleeps == [3.0] and len(calls) == 2
 
 
def test_retry_after_over_cap_gives_up_without_sleeping():
    h, calls = seq(httpx.Response(429, headers={"Retry-After": "3600"}))
    c, sleeps = make(h, max_wait=30)
    with pytest.raises(RetryAfterTooLong):
        c.request("GET", "http://x/")
    assert sleeps == [] and len(calls) == 1
 
 
def test_timeout_is_retried_then_succeeds():
    h, calls = seq(httpx.ReadTimeout("slow"), httpx.Response(200))
    c, sleeps = make(h)
    assert c.request("GET", "http://x/").status_code == 200
    assert len(sleeps) == 1

........                                                                 [100%]
8 passed in 0.10s

def test_envelope_is_exact_when_jitter_takes_the_upper_bound(monkeypatch):
    import tenacity.wait as tw
    monkeypatch.setattr(tw.random, "uniform", lambda lo, hi: hi)
    h, _calls = seq(*[httpx.Response(503)] * 5)
    c, sleeps = make(h, max_attempts=5, base=1.0, max_wait=4.0)
    with pytest.raises(RetryableStatus):
        c.request("GET", "http://x/")
    assert sleeps == [1.0, 2.0, 4.0, 4.0]

import logging
from tenacity import before_sleep_log
 
logger = logging.getLogger("retry")
 
Retrying(
    ...,
    before_sleep=before_sleep_log(logger, logging.WARNING),
)

import httpx
from retry_client import RetryAfterTooLong, RetryableStatus, RetryingClient
 
http = httpx.Client(timeout=httpx.Timeout(10.0, connect=5.0))
client = RetryingClient(http, max_attempts=5, max_wait=30.0)
 
try:
    resp = client.request("GET", "https://api.example.com/items")
except RetryAfterTooLong:
    ...  # 上限を超える待ちを要求された。ジョブを後回しにする
except RetryableStatus as e:
    ...  # 試行を使い切った。e.response に最後の応答が残っている

{
  "hooks": {
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": ".claude/hooks/pytest-gate.sh"
          }
        ]
      }
    ]
  }
}

#!/usr/bin/env bash
input=$(cat)
# 続行中の再発火では二重に止めない
if [ "$(echo "$input" | jq -r '.stop_hook_active')" = "true" ]; then
  exit 0
fi
out=$(python -m pytest -q 2>&1) || { echo "$out" >&2; exit 2; }