Retry-After 要等六秒，剩余预算也只有六秒时怎么办

一个读取任务的总预算是 10 秒。它已经用了 4 秒，刚收到 429，响应带着 Retry-After: 6。如果还要求给下一次尝试留出 1 秒准入余量，程序现在就应该停止。把等待裁成 5 秒会提前重试；原样等完 6 秒，任务又已经到截止点。这个响应提供了等待信息，却没有创造新的任务预算。

本文实现一个可以独立测试的读取策略：只对固定、已准入的 GET 处理 429 和 503；先解析等待值，再决定是否还容得下下一次尝试。所有取消、拒绝和预算不足都走终止路径。连接异常与其他未知异常原样向上传播，不会被兜底重试掩盖。

代码运行在真实 HTTPX 与 asyncio 控制流上，HTTP 响应由 MockTransport 提供，时间由虚拟钟推进，没有真实网络或真实 sleep。它演示的是下一次请求的准入预算，没有实现生产环境的任务总超时。后文的 0.25 秒、1 秒和 10 秒都是测试参数。

本例的入口只能读取 https://fixture.test/read。每次请求显式设置 follow_redirects=False，403、500、302 都由 raise_for_status() 抛出 HTTPStatusError。只有 429 或 503 且携带可解析的单个 Retry-After，才进入重试决策。服务端拒绝访问、调用方取消、输入不合法，都不会引发另一轮请求。

GET 是本文主动选择的范围。不能把“某方法具有幂等语义”直接等同于“当前任务允许自动再发”。测试中的 POST 只作为纯决策函数的字符串输入，预期为 method_not_allowed；测试从未构造或发送 POST。

HTTPX transport 文档说明，transport 的连接重试处理 ConnectError 和 ConnectTimeout，503 这类状态需要另外设计策略。HTTPX 0.28.1 源码中 retries 默认值是 0，httpcore 1.0.9 的连接循环也限定了所捕获的连接错误。本次额外执行了两个 MockTransport 探针：直接调用一次 get，handler 分别返回 429 或 503，调用数都为 1。源码结论与该探针的覆盖范围需要分开看，探针没有执行真实 transport 的连接重试。

本教程可以与逐跳检查 URL、方法和凭据的 redirect 读取器互补，不过两段实现尚未组合。这里遇到 302 就终止，没有测“重定向途中再重试”或“重试后再次重定向”的跨策略状态。

RFC 9110 的 Retry-After 定义允许 HTTP-date 或非负十进制整数秒数。秒数从收到响应后算起。这里先用 headers.get_list('retry-after') 保留重复字段的可见性，再要求恰好一个值。不能先把多个字段拼成字符串，然后挑出其中一个数字。

秒数分支只接受 ASCII 的 0 到 9。前导零和两侧空格、制表符可以接受，负号、正号、小数、全角数字和逗号列表都会终止。解析失败不会降级成 0，否则一个污染字段就可能变成紧密循环。

日期分支接受 HTTP-date 的三种形式：IMF-fixdate、旧式 RFC850 和 asctime。收到响应时，以 UTC 日期相减得到相对等待值，过去日期取 0。之后的调度使用任务的单调时间概念；本例用只前进的 VirtualClock 演示，未模拟真实时钟校准或服务器时钟偏差。

RFC850 的两位年必须相对接收时刻解释。规范处理的是看起来超过未来 50 年的时间戳，不能直接套用固定的“00 到 49 属于某世纪”分界。实现先构造相邻世纪候选，再比较接收时刻加 50 年的阈值，必要时回退 100 年。附加测试取 1970、2026、2090、2100 四个接收年份，各检查 -49、-1、0、1、49 年共 20 个样本；这证明了所列跨世纪样本，尚未证明精确 50 年阈值、闰日和所有日期组合。

实现还采用了几条应用限制：字段最多 128 字符，等待最多 2**31-1 秒，日期与星期必须一致，大小写及 GMT 写法严格匹配。它们不应被写成 RFC 为所有客户端规定的长度或等待上限。秒数 60 归一化到下一 UTC 秒，没有闰秒数据库，也没有真实闰秒时钟验证。

纯函数 decide() 接收方法、状态、已发次数、总次数上限、剩余时间和解析结果。它不读取时钟，不发请求，也不等待。等待值采用 max(server_delay, minimum_gap)，本例 minimum_gap 为 0.25 秒。服务器给出 0 或过去日期，只说明服务器没有要求继续等；应用仍然可以设置自己的最小间隔。

只有 wait + reserve <= remaining 时才准许等待，reserve 默认是 1 秒。剩余 10 秒且要等 9 秒时，等号边界可以准入；剩余 6 秒却要等 6 秒时，立即以 insufficient_budget 停止。余量是一项政策门槛，不能保证下一次请求能在 1 秒内完成。

max_attempts=3 表示初始请求加最多两次重试。计数在启动请求前增加，失败或取消不会退还额度。持续收到 429 且 Retry-After 为 0 的样本，发送 3 次、等待 2 次，每次 0.25 秒；到上限后不再等待。把参数名写成 max_retries 却仍按 3 次总发送实现，会给调用方留下歧义。

deadline 属于整次任务，不能在每轮改成“现在加 10 秒”。测试分别模拟任务此前已用 4 秒、当前响应消耗 4 秒；二者收到 6 秒等待要求后都停止。将已用预算放在循环外，才能避免每轮重新获得完整额度。

次数参数只接受内置整数：max_attempts 必须大于零，纯决策函数的 attempts 必须在 0 到 max_attempts 之间。布尔值、浮点数（包括 3.0、NaN 和无穷大）、字符串都不作为次数。入口先验证再发出初次请求，不能等到响应回来后才发现配置无效。文末给出 2026-10-20 的缺陷复现与回归测试。

执行器使用由可信应用代码注入的 clock、sleeper 和 client/transport，这些依赖不能交给模型或外部响应任意提供。clock 必须可靠地单调前进；sleeper 必须实际等待至少所要求的时长，并正确传播取消；client/transport 不得私自增加请求次数、吞掉异常或取消。以下行为以这些依赖契约成立为前提。

执行器在每次发送前检查取消信号、截止点和余量，拿到响应后立即再检查一次。因此即使返回的是 200，只要虚拟钟已经到 deadline，也报告 deadline，不把迟到结果当作任务成功。

决定重试后先关闭当前响应，再调用注入的异步 sleeper。睡醒重新读取时钟，进入下一轮时再检查余量。oversleep 样本要求等待 2 秒，却让虚拟钟推进到第 10 秒；结果只有初次发送，没有第二次 GET。不能沿用睡前算出的 remaining。

睡后检查只复核截止点与余量，没有独立强制执行最短等待。例如注入的 sleeper 收到 2 秒却直接返回且不推进虚拟钟，代码会在同一时刻再次发送。保留 server_delay 的决策逻辑并不能防御这种违约依赖；本文测试采用按指定规则推进时间的可信 sleeper。

取消具有优先级。已置位的 Event 会阻止初次发送；响应返回同时置位时，不再消费响应为成功；等待期间或 transport 尚未返回时调用 task.cancel()，CancelledError 会传播。asyncio 的取消语义要求协程在适当位置收到取消，并允许清理代码执行。本例没有捕获取消后继续重试，但不能保证任意注入依赖都遵守这一点。如果 transport 吞掉 task.cancel() 引发的 CancelledError 后返回 200，且取消 Event 未置位、预算仍有效，read_get 仍可能返回 200。现有取消样本中的 MockTransport handler 会传播取消，测试结论限于这条控制流。

Event 只是门控信号。生产代码如果注入普通 asyncio.sleep，单独 Event.set() 不会立即打断那次 sleep，需要另行设计可取消的等待或取消整个 task。本文没有交付生产 sleeper，也没有以虚拟测试证明真实 socket 已停止。

HTTPStatusError、ConnectError、ReadTimeout、未知 RuntimeError 与 sleeper 故障均结束循环。测试对人工注入的连接、读取和未知异常检查对象身份，避免把原异常包成看似可重试的错误。只有 Retry-After 解析错误被转成可识别的 RetryStopped，并保留 InvalidRetryAfter 原因链。同一响应同时遇到无效字段和次数上限时，集成代码先解析字段，因此诊断为 invalid_retry_after；两种条件都不会允许下一次发送。

下面四个文件可以完整复制到同一个空目录：retry_budget.py、expected.json、run_tests.py 和 independent_checks.py。run_tests.py 内含 29 个解析夹具及响应场景；expected.json 是运行前手写的 32 个客户端预期，不从 observed.json 反向生成。附加脚本用独立日历差值和预算不等式核对，不调用被测实现来产生预期。

本次使用已有的 Python 3.12.14、HTTPX 0.28.1、httpcore 1.0.9，没有安装软件。复跑环境需预先具备这些依赖；若缺失，先按自己的环境管理流程准备，不能把缺包失败当作策略测试结果。运行时不要使用 python -O，因为这些教学测试使用 assert。

python --version
python -c "import httpx,httpcore; print(httpx.__version__,httpcore.__version__)"
python run_tests.py
python independent_checks.py

主脚本在 MockTransport 外还阻断 socket.connect、connect_ex 和 getaddrinfo，以防误走真实网络。附加脚本同样只使用显式 MockTransport，没有真实 sleep。客户端关闭由 async with 负责，主脚本还检查所有已返回响应都已关闭。

本次实际输出如下。63 是场景数的合计，不是源代码中 assert 语句执行次数。

PASS: 63 assertions/cases; 32 client cases, 29 parser cases, 2 decision cases
{"python": "3.12.14", "httpx": "0.28.1", "httpcore": "1.0.9"}
PASS: 20 century cases, 1 pending-transport cancellation, 48 budget grid combinations

第一次研究运行曾发现测试分支把 http_date 名称误认成 http_ 状态码并解析失败。修正夹具分支后才得到这些通过结果。这个修正属于测试程序，不涉及真实网络行为。完整复跑会写出 observed.json 与 independent-results.json，便于逐项核对；不要用它们覆盖预期文件。

保存为 retry_budget.py。

"""Research fixture. No production hard-deadline or network-security guarantee."""
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import asyncio
import math
import re
import httpx

UTC = timezone.utc
MONTHS = {m: n for n, m in enumerate('Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec'.split(), 1)}
DAYS = 'Mon Tue Wed Thu Fri Sat Sun'.split()
LONG_DAYS = 'Monday Tuesday Wednesday Thursday Friday Saturday Sunday'.split()
MONTH = '(' + '|'.join(MONTHS) + ')'
DAY = '(' + '|'.join(DAYS) + ')'
LONG_DAY = '(' + '|'.join(LONG_DAYS) + ')'
IMF = re.compile(DAY + r', ([0-9]{2}) ' + MONTH + r' ([0-9]{4}) ([0-9]{2}):([0-9]{2}):([0-9]{2}) GMT')
RFC850 = re.compile(LONG_DAY + r', ([0-9]{2})-' + MONTH + r'-([0-9]{2}) ([0-9]{2}):([0-9]{2}):([0-9]{2}) GMT')
ASCTIME = re.compile(DAY + ' ' + MONTH + r' ([0-9]{2}| [0-9]) ([0-9]{2}):([0-9]{2}):([0-9]{2}) ([0-9]{4})')

class InvalidRetryAfter(ValueError):
    pass

class RetryStopped(RuntimeError):
    def __init__(self, reason, attempts, status=None):
        self.reason, self.attempts, self.status = reason, attempts, status
        super().__init__(f'{reason}: attempts={attempts}, status={status}')

@dataclass(frozen=True)
class Parsed:
    delay: float
    form: str

def parse_retry_after(values: list[str], received_wall: datetime) -> Parsed:
    """Strict bounded input; malformed, duplicate, missing fields fail closed.

    Accept three HTTP-date formats. Leap second uses the next UTC second.
    More than 128 characters or 2**31-1 seconds is an application limit, not an RFC limit.
    Past dates impose zero server wait; decision policy supplies a minimum gap.
    """
    if len(values) != 1:
        raise InvalidRetryAfter('missing_or_duplicate')
    raw = values[0].strip(' \t')
    if not raw or len(raw) > 128:
        raise InvalidRetryAfter('empty_or_oversized')
    if received_wall.tzinfo is None or received_wall.utcoffset() != timedelta(0):
        raise ValueError('received_wall must be UTC aware')
    if re.fullmatch(r'[0-9]+', raw):
        seconds = int(raw)
        if seconds > 2**31 - 1:
            raise InvalidRetryAfter('delay_limit')
        return Parsed(float(seconds), 'seconds')
    form = ''
    m = IMF.fullmatch(raw)
    if m:
        weekday, day, month, year, hour, minute, second = m.groups()
        form = 'imf'
    else:
        m = RFC850.fullmatch(raw)
        if m:
            weekday, day, month, yy, hour, minute, second = m.groups()
            # Interpret relative to the receipt year, not Python's 1969 cutoff.
            year = received_wall.year // 100 * 100 + int(yy)
            if year < received_wall.year - 50:
                year += 100
            form = 'rfc850'
        else:
            m = ASCTIME.fullmatch(raw)
            if not m:
                raise InvalidRetryAfter('syntax')
            weekday, month, day, hour, minute, second, year = m.groups()
            form = 'asctime'
    try:
        sec = int(second)
        if sec > 60:
            raise ValueError('invalid second')
        date = datetime(int(year), MONTHS[month], int(day), int(hour), int(minute), min(sec, 59), tzinfo=UTC)
        if form == 'rfc850':
            try:
                threshold = received_wall.replace(year=received_wall.year + 50)
            except ValueError:  # Feb 29 in a non-leap threshold year
                threshold = received_wall.replace(year=received_wall.year + 50, day=28)
            if date > threshold:
                date = date.replace(year=date.year - 100)
        day_names = LONG_DAYS if form == 'rfc850' else DAYS
        if day_names[date.weekday()] != weekday:
            raise ValueError('weekday mismatch')
        date += timedelta(seconds=sec == 60)
    except (ValueError, OverflowError) as exc:
        raise InvalidRetryAfter('invalid_date') from exc
    delay = max(0.0, (date - received_wall).total_seconds())
    if delay > 2**31 - 1:
        raise InvalidRetryAfter('delay_limit')
    return Parsed(delay, form)

@dataclass(frozen=True)
class Decision:
    retry: bool
    reason: str
    delay: float = 0.0

def decide(*, method, status, attempts, max_attempts, remaining, parsed, minimum_gap=0.25, reserve=1.0):
    """Attempts includes the original send; reserve is policy, not performance."""
    if (type(max_attempts) is not int or type(attempts) is not int
            or max_attempts < 1 or attempts < 0 or attempts > max_attempts):
        raise ValueError('invalid attempt counters')
    if not all(math.isfinite(x) for x in (remaining, minimum_gap, reserve)) or minimum_gap <= 0 or reserve <= 0:
        raise ValueError('invalid time policy')
    if method != 'GET':
        return Decision(False, 'method_not_allowed')
    if status not in (429, 503):
        return Decision(False, 'status_not_allowed')
    if attempts >= max_attempts:
        return Decision(False, 'attempt_limit')
    if parsed is None:
        return Decision(False, 'invalid_retry_after')
    if not math.isfinite(parsed.delay) or parsed.delay < 0:
        raise ValueError('invalid parsed delay')
    delay = max(parsed.delay, minimum_gap)
    if delay + reserve > remaining:
        return Decision(False, 'insufficient_budget')
    return Decision(True, 'retry', delay)

@dataclass
class VirtualClock:
    now: float = 0.0
    origin: datetime = datetime(2026, 10, 7, 12, 0, 0, tzinfo=UTC)
    def wall(self):
        return self.origin + timedelta(seconds=self.now)
    def advance(self, seconds):
        if not math.isfinite(seconds) or seconds < 0:
            raise ValueError('invalid clock advance')
        self.now += seconds

async def read_get(client, *, clock, sleep, cancelled, deadline, max_attempts=3, reserve=1.0, trace=None):
    """Fixed approved .test URL; injected async sleep must be cancellation-aware.

    Caller owns the client. No catches around network/sleep/cancellation errors.
    Virtual admission/accounting only. HTTPX timeout is NOT a total deadline.
    """
    if type(max_attempts) is not int or max_attempts < 1:
        raise ValueError('max_attempts must be a positive integer')
    if not math.isfinite(deadline) or not math.isfinite(reserve) or reserve <= 0:
        raise ValueError('invalid policy')
    trace = trace if trace is not None else []
    attempts = 0
    def check():
        if cancelled.is_set():
            raise asyncio.CancelledError
        if clock.now >= deadline:
            raise RetryStopped('deadline', attempts)
    while True:
        check()
        if deadline - clock.now < reserve:
            raise RetryStopped('insufficient_budget', attempts)
        if attempts >= max_attempts:
            raise RetryStopped('attempt_limit', attempts)
        attempts += 1
        trace.append({'send': attempts, 't': clock.now})
        response = await client.get('https://fixture.test/read', follow_redirects=False, timeout=deadline-clock.now)
        # Responses are fully buffered here; no streaming/body-size guarantees.
        try:
            check()  # Late results do not become success or fresh retries.
            if response.status_code not in (429, 503):
                response.raise_for_status()
                return response.status_code
            try:
                parsed = parse_retry_after(response.headers.get_list('retry-after'), clock.wall())
            except InvalidRetryAfter as exc:
                # Explicit terminal error with cause; never turn malformed into 0.
                raise RetryStopped('invalid_retry_after', attempts, response.status_code) from exc
            plan = decide(method='GET', status=response.status_code, attempts=attempts,
                          max_attempts=max_attempts, remaining=deadline-clock.now,
                          parsed=parsed, reserve=reserve)
            if not plan.retry:
                raise RetryStopped(plan.reason, attempts, response.status_code)
            trace.append({'wait': plan.delay, 't': clock.now})
        finally:
            await response.aclose()
        await sleep(plan.delay)
        # Recheck actual clock and cancellation after scheduler/sleep overshoot.
        check()

保存为 expected.json。

{
  "default_429": {"status":429,"attempts":1},
  "default_503": {"status":503,"attempts":1},
  "seconds_429": {"outcome":"ok","attempts":2,"waits":[2.0],"clock":2.0},
  "seconds_503": {"outcome":"ok","attempts":2,"waits":[3.0],"clock":3.0},
  "http_date": {"outcome":"ok","attempts":2,"waits":[5.0],"clock":5.0},
  "past_date": {"outcome":"ok","attempts":2,"waits":[0.25],"clock":0.25},
  "zero": {"outcome":"ok","attempts":2,"waits":[0.25],"clock":0.25},
  "missing": {"outcome":"invalid_retry_after","attempts":1,"waits":[],"clock":0.0},
  "invalid": {"outcome":"invalid_retry_after","attempts":1,"waits":[],"clock":0.0},
  "duplicate": {"outcome":"invalid_retry_after","attempts":1,"waits":[],"clock":0.0},
  "over_deadline": {"outcome":"insufficient_budget","attempts":1,"waits":[],"clock":0.0},
  "equal_deadline": {"outcome":"insufficient_budget","attempts":1,"waits":[],"clock":0.0},
  "reserve_boundary": {"outcome":"ok","attempts":2,"waits":[9.0],"clock":9.0},
  "already_used": {"outcome":"insufficient_budget","attempts":1,"waits":[],"clock":4.0},
  "response_cost": {"outcome":"insufficient_budget","attempts":1,"waits":[],"clock":4.0},
  "oversleep": {"outcome":"deadline","attempts":1,"waits":[2.0],"clock":10.0},
  "attempt_cap": {"outcome":"attempt_limit","attempts":3,"waits":[0.25,0.25],"clock":0.5},
  "one_attempt": {"outcome":"attempt_limit","attempts":1,"waits":[],"clock":0.0},
  "expired_initial": {"outcome":"deadline","attempts":0,"waits":[],"clock":10.0},
  "tiny_initial": {"outcome":"insufficient_budget","attempts":0,"waits":[],"clock":9.5},
  "late_success": {"outcome":"deadline","attempts":1,"waits":[],"clock":10.0},
  "http_403": {"outcome":"HTTPStatusError","attempts":1,"waits":[],"clock":0.0},
  "http_500": {"outcome":"HTTPStatusError","attempts":1,"waits":[],"clock":0.0},
  "redirect": {"outcome":"HTTPStatusError","attempts":1,"waits":[],"clock":0.0},
  "connect_error": {"outcome":"ConnectError","attempts":1,"waits":[],"clock":0.0},
  "read_timeout": {"outcome":"ReadTimeout","attempts":1,"waits":[],"clock":0.0},
  "unknown_error": {"outcome":"RuntimeError","attempts":1,"waits":[],"clock":0.0},
  "sleep_error": {"outcome":"RuntimeError","attempts":1,"waits":[2.0],"clock":0.0},
  "cancel_before": {"outcome":"CancelledError","attempts":0,"waits":[],"clock":0.0},
  "cancel_sleep": {"outcome":"CancelledError","attempts":1,"waits":[2.0],"clock":1.0},
  "cancel_response": {"outcome":"CancelledError","attempts":1,"waits":[],"clock":0.0},
  "task_cancel_sleep": {"outcome":"CancelledError","attempts":1,"waits":[2.0],"clock":0.0}
}

保存为 run_tests.py。

"""Assertions against hand-authored expected.json; no real network or sleeps."""
import asyncio
from datetime import datetime, timezone
import inspect
import json
from pathlib import Path
import platform
import socket
import httpx
import httpcore
from retry_budget import *
ROOT=Path(__file__).parent
# Defense in depth: any accidental socket path aborts, rather than reaching a host.
def deny_network(*args, **kwargs):
    raise AssertionError('network forbidden in this laboratory')
socket.socket.connect = deny_network
socket.socket.connect_ex = deny_network
socket.getaddrinfo = deny_network

EXPECTED=json.loads((ROOT/'expected.json').read_text())
observed={}
PARSER_CASES=[
    ('seconds', ['2'], 2.0), ('ows',[' \t2\t '],2.0), ('zero',['0'],0.0),
    ('leading_zero',['0002'],2.0), ('negative',['-1'],None), ('plus',['+1'],None),
    ('decimal',['0.5'],None), ('unicode_digit',['２'],None), ('empty',[''],None),
    ('missing',[],None), ('duplicate',['1','2'],None), ('merged',['1, 2'],None),
    ('oversized',['9'*129],None), ('huge_rejected',['9'*128],None), ('delay_limit',['2147483648'],None), ('max_delay',['2147483647'],2147483647.0),
    ('imf',['Wed, 07 Oct 2026 12:00:05 GMT'],5.0),
    ('rfc850',['Wednesday, 07-Oct-26 12:00:05 GMT'],5.0),
    ('asctime',['Wed Oct  7 12:00:05 2026'],5.0),
    ('past',['Wed, 07 Oct 2026 11:59:59 GMT'],0.0),
    ('bad_day',['Wed, 32 Oct 2026 12:00:05 GMT'],None),
    ('bad_weekday',['Thu, 07 Oct 2026 12:00:05 GMT'],None),
    ('non_gmt',['Wed, 07 Oct 2026 12:00:05 UTC'],None),
    ('trailing',['Wed, 07 Oct 2026 12:00:05 GMT extra'],None),
    ('missing_zone',['Wed, 07 Oct 2026 12:00:05'],None),
    ('date_delay_limit',['Fri, 31 Dec 9999 23:59:59 GMT'],None),
    ('rfc850_future',['Monday, 07-Oct-75 12:00:05 GMT'],1546300805.0),
    ('rfc850_past',['Friday, 07-Oct-77 12:00:05 GMT'],0.0),
    ('leap_second',['Wed, 07 Oct 2026 11:59:60 GMT'],0.0)
]

def parser_tests():
    results=[]
    wall=datetime(2026,10,7,12,0,0,tzinfo=timezone.utc)
    for name, values, expected in PARSER_CASES:
        try:actual=parse_retry_after(values,wall).delay
        except InvalidRetryAfter:actual=None
        assert actual == expected,(name,expected,actual)
        results.append({'name':name,'expected':expected,'observed':actual})
    # Pure decision test only: no POST is constructed or sent.
    d=decide(method='POST',status=503,attempts=1,max_attempts=3,remaining=10,parsed=Parsed(1,'seconds'))
    assert (d.retry,d.reason)==(False,'method_not_allowed')
    d=decide(method='GET',status=500,attempts=1,max_attempts=3,remaining=10,parsed=Parsed(1,'seconds'))
    assert (d.retry,d.reason)==(False,'status_not_allowed')
    return results

async def run_case(name):
    clock=VirtualClock(now={'already_used':4,'expired_initial':10,'tiny_initial':9.5}.get(name,0.0))
    cancelled=asyncio.Event(); entered=asyncio.Event(); never=asyncio.Event()
    calls=[]; waits=[]; responses=[]; trace=[]
    original_error=None
    if name=='cancel_before': cancelled.set()
    async def handler(request):
        nonlocal original_error
        assert request.method=='GET' and str(request.url)=='https://fixture.test/read'
        assert 'authorization' not in request.headers and 'cookie' not in request.headers
        calls.append(str(request.url))
        if name in ('connect_error','read_timeout','unknown_error'):
            original_error={'connect_error':httpx.ConnectError('fixture'), 'read_timeout':httpx.ReadTimeout('fixture'), 'unknown_error':RuntimeError('fixture')}[name]
            raise original_error
        if name=='response_cost':clock.advance(4)
        if name=='late_success':clock.advance(10)
        if name=='cancel_response':cancelled.set()
        status=429; headers=[('Retry-After','2')]
        if name=='seconds_503':status=503;headers=[('Retry-After','3')]
        if name=='http_date':headers=[('Retry-After','Wed, 07 Oct 2026 12:00:05 GMT')]
        if name=='past_date':headers=[('Retry-After','Wed, 07 Oct 2026 11:59:59 GMT')]
        if name in ('zero','attempt_cap'):headers=[('Retry-After','0')]
        if name=='missing':headers=[]
        if name=='invalid':headers=[('Retry-After','oops')]
        if name=='duplicate':headers=[('Retry-After','1'),('Retry-After','2')]
        if name=='over_deadline':headers=[('Retry-After','11')]
        if name=='equal_deadline':headers=[('Retry-After','10')]
        if name=='reserve_boundary':headers=[('Retry-After','9')]
        if name in ('already_used','response_cost'):headers=[('Retry-After','6')]
        if name in ('http_403', 'http_500'):status=int(name[5:])
        if name=='redirect':status=302;headers=[('Location','https://fixture.test/other')]
        if name=='late_success' or (len(calls)>1 and name!='attempt_cap'):status=200;headers=[]
        r=httpx.Response(status,headers=headers,content=b'fixture');responses.append(r);return r
    async def sleep(delay):
        nonlocal original_error
        waits.append(delay)
        if name=='sleep_error':
            original_error=RuntimeError('sleep fixture');raise original_error
        if name=='task_cancel_sleep':
            entered.set();await never.wait()
        elif name=='cancel_sleep':
            clock.advance(1);cancelled.set();raise asyncio.CancelledError
        else:clock.advance(10 if name=='oversleep' else delay)
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler),trust_env=False) as client:
        coroutine=read_get(client,clock=clock,sleep=sleep,cancelled=cancelled,deadline=10,
                           max_attempts=1 if name=='one_attempt' else 3,trace=trace)
        try:
            if name=='task_cancel_sleep':
                task=asyncio.create_task(coroutine)
                await entered.wait();task.cancel();await task
            else:await coroutine
            outcome='ok'
        except RetryStopped as exc:
            outcome=exc.reason
            if outcome=='invalid_retry_after':assert isinstance(exc.__cause__,InvalidRetryAfter)
        except (asyncio.CancelledError,httpx.HTTPStatusError,httpx.ConnectError,httpx.ReadTimeout,RuntimeError) as exc:
            outcome=type(exc).__name__
            if original_error is not None:assert exc is original_error,'exception swallowed or replaced'
    assert all(r.is_closed for r in responses),'response leak'
    actual={'outcome':outcome,'attempts':len(calls),'waits':waits,'clock':clock.now}
    assert actual==EXPECTED[name],(name,EXPECTED[name],actual)
    observed[name]=actual

async def main():
    parsers=parser_tests()
    for status in (429,503):
        calls=[]
        def handler(request):calls.append(request);return httpx.Response(status,headers={'Retry-After':'1'})
        async with httpx.AsyncClient(transport=httpx.MockTransport(handler),trust_env=False) as client:
            r=await client.get('https://fixture.test/read')
            actual={'status':r.status_code,'attempts':len(calls)}
            await r.aclose()
        assert actual==EXPECTED[f'default_{status}'];observed[f'default_{status}']=actual
    for name in EXPECTED:
        if not name.startswith('default_'):await run_case(name)
    output={'environment':{'python':platform.python_version(),'httpx':httpx.__version__,'httpcore':httpcore.__version__},
            'scope':'Virtual policy time only; all HTTP uses MockTransport; sockets denied; no real sleep',
            'cases':observed,'parser_cases':parsers,'pure_decision_cases':['POST rejected without sending','500 not retryable'],
            'passed':len(observed)+len(parsers)+2}
    (ROOT/'observed.json').write_text(json.dumps(output,ensure_ascii=False,indent=2)+'\n')
    print(f'PASS: {output["passed"]} assertions/cases; {len(observed)} client cases, {len(parsers)} parser cases, 2 decision cases')
    print(json.dumps(output['environment']))

if __name__=='__main__':asyncio.run(main())

保存为 independent_checks.py。

"""Supplemental checks; expected values do not call the implementation to derive them."""
from datetime import datetime, timezone
import asyncio
import json
from pathlib import Path
import httpx
from retry_budget import parse_retry_after, VirtualClock, read_get, decide, Parsed, RetryStopped

async def main():
    results=[]
    for year in (1970,2026,2090,2100):
        now=datetime(year,1,1,tzinfo=timezone.utc)
        for delta in (-49,-1,0,1,49):
            target=now.replace(year=year+delta)
            value=target.strftime('%A, %d-%b-%y %H:%M:%S GMT')
            expected=max(0,(target-now).total_seconds())
            observed=parse_retry_after([value],now).delay
            assert observed==expected,(year,delta,value,observed)
            results.append({'kind':'rfc850_sliding_century','year':year,'delta':delta,'pass':True})
    # Task cancellation during a pending transport operation, no real socket.
    entered=asyncio.Event(); pending=asyncio.Event(); cancelled=asyncio.Event()
    calls=[];waits=[]
    async def handler(request):
        assert request.method=='GET' and str(request.url)=='https://fixture.test/read'
        calls.append(str(request.url));entered.set();await pending.wait()
        raise AssertionError('cancelled handler continued')
    async def sleep(delay):waits.append(delay)
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler),trust_env=False) as client:
        task=asyncio.create_task(read_get(client,clock=VirtualClock(),sleep=sleep,cancelled=cancelled,deadline=10))
        await entered.wait();task.cancel()
        try:await task
        except asyncio.CancelledError:pass
        else:raise AssertionError('cancellation swallowed')
    assert len(calls)==1 and waits==[] and task.cancelled()
    results.append({'kind':'cancel_pending_transport','calls':1,'waits':[],'pass':True})
    # Server lower bound is never shortened: property matrix independent of parser.
    checked=0
    for delay in (0,0.1,0.25,1,2,9,10,11):
        for remaining in (0,0.25,1,2,5,10):
            d=decide(method='GET',status=503,attempts=1,max_attempts=3,remaining=remaining,parsed=Parsed(delay,'test'))
            allowed=max(delay,0.25)+1<=remaining
            assert d.retry==allowed
            if allowed:assert d.delay>=delay and d.delay+1<=remaining
            checked+=1
    results.append({'kind':'server_delay_budget_grid','checked':checked,'pass':True})
    path=Path(__file__).with_name('independent-results.json')
    path.write_text(json.dumps(results,indent=2)+'\n')
    print('PASS: 20 century cases, 1 pending-transport cancellation, 48 budget grid combinations')
asyncio.run(main())

这组测试支持所列输入的解析行为、下一次 GET 的准入判断、合成预算记账和 asyncio 取消路径。保证以可信的单调钟、至少等待请求时长且传播取消的 sleeper，以及不私自重试、不吞异常或取消的 client/transport 为前提；实现没有防御任意注入依赖。它没有测真实 DNS、TLS、代理、HTTP/2、服务端处理、网络吞吐、真实网络取消、并发共享预算或跨进程恢复。所有时间推进都是夹具行为，不能换算成网络性能或超时精度成绩。

即便请求调用里传入 timeout=deadline-clock.now，也不能声称提供总 deadline。HTTPX timeout分别涉及 connect、read、write、pool 等网络阶段，MockTransport 不模拟这些阶段。真正的任务总时限需要外层异步时限、取消传播和资源清理的组合，并另外验证底层执行何时停止。

本文没有流式读取、body 大小限制、解压限制、凭据重放控制或 URL/DNS 安全边界。完整响应在返回前已缓冲。生产读取器需要在自己的授权与网络安全边界内引入这项策略，再验证 redirects、重试、取消与预算共用时的行为。重试准入不会授予新 URL 或新动作的权限。

没有 Retry-After 的 429 在这里终止。若业务另有退避协议，应独立写明输入、最大等待和授权条件，不能借这个例子的通过结果默认启用指数退避、jitter 或供应商专用限流头。首先保持拒绝和取消能够可靠结束，再讨论扩大重试范围。

正文包含三张原创说明图，均表达程序和合成时间关系，不是产品截图。HTML 做了本地结构和引用检查；没有进行浏览器 GUI 排版验证。资料与执行版本核对日期为 2026-10-07。

原文已经说明：睡醒后的检查只复核截止点与余量，无法防御立即返回的违约 sleeper。本节给这一已知边界加一个可选适配器。第一次响应是 429，带 Retry-After: 2；第二次响应是 200。故意注入一个只推进虚拟钟 1 秒就返回的异步等待器，原入口的发送时刻为 [0, 1]；如果等待器完全不推进时钟，则为 [0, 0]。两种情况下都能收到 200，但都没有达到两秒等待下限。

这是自定义异步等待器、适配层和时钟配对的合成故障注入，不是 CPython 的 sleep 缺陷复现。Python 3.12 文档明确说明，标准 time.sleep 自 3.5 起会至少等待请求时长，即使被信号打断也会重新计算并继续，除非信号处理器抛出异常。此处没有调用真实 time.sleep，也没有测任何平台的 asyncio.sleep 提前返回。

RFC 9110 §10.2.3定义 Retry-After 的等待信息；本例把它作为下一次请求的等待下限。增补不改变原解析器、GET 地址、可重试状态、次数上限或预算策略。原入口继续保留；需要这层防御时，调用方明确选择 guarded_read。

适配器在响应关闭后记录 start，仅一次计算 target = start + delay。每次等待返回都从同一时间域重新读钟：达到 target 才交回请求循环；尚未达到但有进展，就只等待 target - now。提前返回不重置 target，也不会重新获得完整任务预算。关闭响应消耗的时间没有抵扣本轮等待，这是保守选择，可能多等或更早耗尽预算，却不会提前重试。

本例最多调用 sleeper 三次，属于演示策略。如果三次小步推进仍未到达 target，报 early_wake_limit；完全无进展则第一次就报 no_clock_progress。时钟倒退、NaN、无穷大或无法表示的等待目标也终止。所有这些停止都不会发出第二次请求。取消优先检查，等待器异常继续向上传播，不转为零秒等待。

CheckedClock只验证观测值有限且非递减，不能证明时钟说的是真实时间，也不能发现采样间先倒退再追上的变化。time.monotonic 的契约适合比较经过时间；本测试仍使用 VirtualClock。时钟、deadline 和 sleeper 必须由可信代码提供并使用同一时间域。HTTP-date 仍由原入口解释；本轮 delay 固定后，墙钟跳变不会改写 target。

到第 10 秒返回时，适配器报告 wait_deadline；第 9.5 秒返回尚未到 deadline，但原入口仍会因不足一秒余量报 insufficient_budget。因此等待完成只是再次准入的必要条件。一个永远不返回的 sleeper 仍会挂住；这层门控没有创建硬总超时，也不能约束 transport 内部的重试或取消吞噬。asyncio 的取消传播与清理仍需依赖正确配合。

guarded_wait.py

"""Local fixture: sampled clock checks and a bounded not-before sleep adapter."""
import asyncio
import math
from retry_budget import RetryStopped, read_get

class ClockFault(RuntimeError):
    pass

class CheckedClock:
    def __init__(self, inner):
        self.inner = inner
        self.previous = None
    @property
    def now(self):
        value = self.inner.now
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
            raise ClockFault('non_finite_clock')
        if self.previous is not None and value < self.previous:
            raise ClockFault('backward_clock')
        self.previous = value
        return value
    def wall(self):
        return self.inner.wall()

def make_wait(clock, sleeper, cancelled, deadline, max_wakes=3):
    if type(max_wakes) is not int or max_wakes < 1:
        raise ValueError('max_wakes must be a positive integer')
    async def wait(delay):
        if cancelled.is_set():
            raise asyncio.CancelledError
        start = clock.now
        target = start + delay
        if not math.isfinite(target) or delay <= 0 or target <= start:
            raise ClockFault('unrepresentable_wait')
        for _ in range(max_wakes):
            if cancelled.is_set():
                raise asyncio.CancelledError
            before = clock.now
            if before >= deadline:
                raise ClockFault('wait_deadline')
            if before >= target:
                return
            await sleeper(target - before)
            if cancelled.is_set():
                raise asyncio.CancelledError
            after = clock.now
            if after >= deadline:
                raise ClockFault('wait_deadline')
            if after >= target:
                return
            if after == before:
                raise ClockFault('no_clock_progress')
        raise ClockFault('early_wake_limit')
    return wait

async def guarded_read(client, *, clock, sleep, cancelled, deadline, max_wakes=3, **kwargs):
    checked = CheckedClock(clock)
    wait = make_wait(checked, sleep, cancelled, deadline, max_wakes)
    return await read_get(client, clock=checked, sleep=wait, cancelled=cancelled,
                          deadline=deadline, **kwargs)

所有测试固定读取 https://fixture.test/read，使用 HTTPX MockTransport 与合成时间，没有真实 HTTP、模型调用或生产请求。通常从 0 秒开始，deadline 为 10 秒，Retry-After 为 2，最多三次尝试并保留一秒准入余量。只有 zero_header 把头改为 0，initial_nan 把初始读数改为 NaN。下列夹具源码给出每个等待器的完整输入规则；预期 JSON 是预先写定的独立判据，运行不会根据观察结果重写它。

正常等待与一次提前返回后的补等都必须在 [0, 2] 发送，waits 分别为 [2] 和 [2, 1]；分三次到达时 waits 为 [2, 1.5, 1]。无进展、三次小步不足、倒退、非有限读数、取消和异常都只保留第一次发送；初始 NaN 连第一次都不发送。零秒头沿用 0.25 秒应用间隔；墙钟前跳 30 天仍补等至单调时间 2 秒。精确状态和全部列表如下，不只检查成功码。

wake_expected.json

{
 "normal": {"outcome":"ok","sends":[0,2],"waits":[2]},
 "early_recovery": {"outcome":"ok","sends":[0,2],"waits":[2,1]},
 "three_wakes": {"outcome":"ok","sends":[0,2],"waits":[2,1.5,1]},
 "no_progress": {"outcome":"no_clock_progress","sends":[0],"waits":[2]},
 "wake_cap": {"outcome":"early_wake_limit","sends":[0],"waits":[2,1.75,1.5]},
 "backward": {"outcome":"backward_clock","sends":[0],"waits":[2]},
 "nan": {"outcome":"non_finite_clock","sends":[0],"waits":[2]},
 "infinity": {"outcome":"non_finite_clock","sends":[0],"waits":[2]},
 "initial_nan": {"outcome":"non_finite_clock","sends":[],"waits":[]},
 "oversleep": {"outcome":"wait_deadline","sends":[0],"waits":[2]},
 "cancel": {"outcome":"cancelled","sends":[0],"waits":[2]},
 "sleeper_error": {"outcome":"sleeper_error","sends":[0],"waits":[2]},
 "zero_header": {"outcome":"ok","sends":[0,0.25],"waits":[0.25]},
 "wall_jump": {"outcome":"ok","sends":[0,2],"waits":[2,1]},
 "reserve_after_overshoot": {"outcome":"insufficient_budget","sends":[0],"waits":[2]}
}

两个负对照让未加门控的原入口运行相同夹具，必须准确观察到 [0, 1] 与 [0, 0]，并确认它们不满足新判据。故障场景结束后另建健康夹具，确认可得到 [0, 2]；这是测试明确发起的新调用，不是失败任务暗中重启。响应关闭、固定 GET、无凭据和 socket 禁用也有检查。

test_guarded_wait.py

"""Synthetic time only. No sockets, real sleeps, models or production requests."""
import asyncio
from datetime import timedelta
import json
from pathlib import Path
import platform
import socket
import httpx
from retry_budget import VirtualClock, RetryStopped, read_get
from guarded_wait import guarded_read, ClockFault
ROOT = Path(__file__).parent
EXPECTED = json.loads((ROOT / 'wake_expected.json').read_text())

def denied(*args, **kwargs):
    raise AssertionError('network forbidden')
socket.socket.connect = denied
socket.socket.connect_ex = denied
socket.getaddrinfo = denied

async def scenario(name, reader=guarded_read):
    clock = VirtualClock()
    if name == 'initial_nan':
        clock.now = float('nan')
    cancel = asyncio.Event()
    sends, waits, responses = ([], [], [])

    async def handler(request):
        if not (request.method == 'GET' and str(request.url) == 'https://fixture.test/read'):
            raise AssertionError('check failed')
        if not ('authorization' not in request.headers and 'cookie' not in request.headers):
            raise AssertionError('check failed')
        sends.append(clock.now)
        response = httpx.Response(429 if len(sends) == 1 else 200, headers={'Retry-After': '0' if name == 'zero_header' else '2'}, content=b'fixture')
        responses.append(response)
        return response

    async def sleep(delay):
        if not responses[-1].is_closed:
            raise AssertionError('response must close before waiting')
        waits.append(delay)
        if name == 'cancel':
            cancel.set()
            return
        if name == 'sleeper_error':
            raise RuntimeError('sleeper_error')
        if name == 'no_progress':
            return
        if name == 'backward':
            clock.now = -1
            return
        if name == 'nan':
            clock.now = float('nan')
            return
        if name == 'infinity':
            clock.now = float('inf')
            return
        if name == 'oversleep':
            clock.advance(10)
            return
        if name == 'reserve_after_overshoot':
            clock.advance(9.5)
            return
        if name == 'wake_cap':
            clock.advance(0.25)
            return
        if name == 'three_wakes':
            clock.advance(0.5 if len(waits) < 3 else delay)
            return
        if name in ('early_recovery', 'wall_jump') and len(waits) == 1:
            if name == 'wall_jump':
                clock.origin += timedelta(days=30)
            clock.advance(1)
            return
        clock.advance(delay)
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as client:
        try:
            await reader(client, clock=clock, sleep=sleep, cancelled=cancel, deadline=10)
            outcome = 'ok'
        except asyncio.CancelledError:
            outcome = 'cancelled'
        except RetryStopped as exc:
            outcome = exc.reason
        except (ClockFault, RuntimeError) as exc:
            outcome = str(exc)
    if not all((r.is_closed for r in responses)):
        raise AssertionError('check failed')
    return {'outcome': outcome, 'sends': sends, 'waits': waits}

async def main():
    observed = {}
    for name, expected in EXPECTED.items():
        actual = await scenario(name)
        if not actual == expected:
            raise AssertionError((name, actual, expected))
        observed[name] = actual
    bad_early = await scenario('early_recovery', read_get)
    if not bad_early == {'outcome': 'ok', 'sends': [0, 1], 'waits': [2]}:
        raise AssertionError('check failed')
    if not bad_early != EXPECTED['early_recovery']:
        raise AssertionError('check failed')
    bad_stalled = await scenario('no_progress', read_get)
    if not bad_stalled == {'outcome': 'ok', 'sends': [0, 0], 'waits': [2]}:
        raise AssertionError('check failed')
    if not bad_stalled != EXPECTED['no_progress']:
        raise AssertionError('check failed')
    recovery = await scenario('normal')
    if not recovery == EXPECTED['normal']:
        raise AssertionError('check failed')
    result = {'environment': {'python': platform.python_version(), 'httpx': httpx.__version__}, 'cases': observed, 'negative_controls': {'unguarded_early': bad_early, 'unguarded_stalled': bad_stalled}, 'fresh_healthy_fixture': recovery, 'passed_cases': 15, 'negative_controls_detected': 2, 'scope': 'Local HTTPX MockTransport and synthetic time only; no production timeout claim'}
    (ROOT / 'wake_observed.json').write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n')
    print('PASS: 15 fixed-oracle scenarios; 2 negative controls detected; 1 fresh recovery fixture')
    print(json.dumps(result['environment']))
if __name__ == '__main__':
    asyncio.run(main())

把前文与文末共九个文件放在同一目录：retry_budget.py、expected.json、run_tests.py、independent_checks.py、guarded_wait.py、wake_expected.json、test_guarded_wait.py、verify_all.py、文末的 test_attempt_policy.py。依赖为 Python 3.12、HTTPX 0.28.1 和 httpcore 1.0.9；此次复跑使用已有环境，没有安装软件。执行 python verify_all.py，再执行 python -O verify_all.py。

原有测试仍包含 assert，因此不要直接把原测试脚本用 -O 跑后当成验收。此入口显式以 compile(..., optimize=0) 执行四套可信本地测试，并先运行一个必定失败的 assert 探针，验证断言没有被优化掉。新增等待测试自身使用显式检查。入口不接受来自网络、命令行或模型输出的代码或文件名。

verify_all.py

"""Run the unchanged baseline with assertions enabled, even under python -O.

Only execute the trusted files shipped alongside this script. The harness does
not accept filenames or code from HTTP responses, arguments, or model output.
"""
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))

def checked_exec(source, filename, namespace):
    exec(compile(source, filename, 'exec', optimize=0), namespace)

def main():
    # This deliberately false assertion must still execute under python -O.
    try:
        checked_exec('assert False, "assertion probe"', '<probe>', {})
    except AssertionError:
        pass
    else:
        raise RuntimeError('assertions disabled in test harness')
    for name in ('run_tests.py', 'independent_checks.py', 'test_guarded_wait.py', 'test_attempt_policy.py'):
        path = ROOT / name
        checked_exec(path.read_text(encoding='utf-8'), str(path),
                     {'__name__': '__main__', '__file__': str(path)})
    print('PASS: assertion probe and all four suites (explicit optimize=0)')

if __name__ == '__main__':
    main()

2026-10-14 本地复跑环境为 Python 3.12.14 / HTTPX 0.28.1 / httpcore 1.0.9。普通模式和通过上述入口的 -O 模式均通过：原主测试 63 项（32 个客户端场景、29 个解析场景、2 个决策场景），另有 20 个跨世纪样本、1 个 pending transport 取消样本、48 个预算组合；新增 15 个固定预期场景、2 个错误对照和1 个全新健康恢复夹具也通过。完整源码、HTML 内嵌源码提取件、离线 ZIP 解压件分别复跑；故意把正常样本的预期发送时刻改成 [0, 1]，两种启动模式都失败。

官方依据核对到 CPython v3.12.14 的 time 文档源码与 HTTPX 0.28.1 的 MockTransport 源码。MockTransport 调用 handler 产生合成响应，不是网络取消、DNS、TLS、吞吐或跨平台计时精度实验。合成的两秒不能换算成速度成绩，断言通过也不构成生产总时限保证。

遇到 no_clock_progress，先检查等待器与时钟是否配对；遇到 backward_clock 或 non_finite_clock，先修复时钟来源；反复提前返回则检查适配层契约。不要靠删除门控、把等待改为零或盲目扩大次数来消除报错。恢复依赖后，再由调用方明确发起新调用验证。图 4 是原创控制流示意图，不是实际产品截图。

原实现只比较次数的大小，没有先验证类型。2026-10-20 在固定 429、Retry-After: 0、初始时间 0、deadline 为 2 秒、余量 1 秒的夹具中，max_attempts=float('nan') 或正无穷大会发送 5 次 GET，最后因 insufficient_budget 停止，而不是在输入阶段拒绝。max_attempts=1.5 则先发送两次，收到第二个响应后才由 decide 抛出 ValueError。这个问题不是无限循环复现，但已经说明无效次数配置可能先产生请求。

上文 retry_budget.py 已修正两处：read_get 在初次发送前要求 type(max_attempts) is int 且大于零；decide 同时验证 max_attempts 和 attempts 的整数类型及范围。guarded_read 继续调用同一个 read_get，不另写一套次数政策。Python 的布尔类型文档说明 bool 是 int 的子类，因此这里只用 isinstance(value, int) 不足以排除 True。只接受内置 int 是本例明确选择的应用契约，不会把字符串、浮点值或自定义数字类型自动转换成次数。

下面测试对两个读取入口分别输入 11 种无效上限，要求 ValueError、零发送、零等待、零虚拟时间推进及空 trace；另有 26 个纯决策计数用例和 6 个合法上限用例。合法上限 1、2、3 的持续 429 夹具，必须分别恰好发送 1、2、3 次，并只等待 0、1、2 次。预期值来自这些明确规则，不由被测函数生成。测试不使用 assert 作为通过判据。

保存为 test_attempt_policy.py，与前文八个文件同目录，再运行更新后的 python verify_all.py 与 python -O verify_all.py。

test_attempt_policy.py

"""Fixed attempt-counter oracles; MockTransport and virtual time only."""
import asyncio
import json
from pathlib import Path
import socket
import httpx
from retry_budget import Parsed, RetryStopped, VirtualClock, decide, read_get
from guarded_wait import guarded_read

BAD_CAPS = [
    ("nan", float("nan")), ("infinity", float("inf")),
    ("negative_infinity", float("-inf")), ("fraction", 1.5),
    ("integral_float", 3.0), ("true", True), ("false", False),
    ("zero", 0), ("negative", -1), ("string", "3"), ("none", None),
]
BAD_ATTEMPTS = [
    ("nan", float("nan")), ("infinity", float("inf")),
    ("negative_infinity", float("-inf")), ("fraction", 0.5),
    ("integral_float", 1.0), ("true", True), ("false", False),
    ("negative", -1), ("over_cap", 4), ("string", "1"), ("none", None),
]

def check(condition, detail):
    if not condition:
        raise AssertionError(detail)

def deny_network(*args, **kwargs):
    raise AssertionError("network forbidden in this laboratory")

socket.socket.connect = deny_network
socket.socket.connect_ex = deny_network
socket.getaddrinfo = deny_network

def decision(attempts=1, cap=3):
    return decide(method="GET", status=429, attempts=attempts,
                  max_attempts=cap, remaining=10, parsed=Parsed(0, "seconds"))

def pure_checks():
    count = 0
    for field, cases in (("cap", BAD_CAPS), ("attempts", BAD_ATTEMPTS)):
        for name, value in cases:
            try:
                decision(**{field: value})
            except ValueError as exc:
                check(str(exc) == "invalid attempt counters", (field, name, exc))
            else:
                raise AssertionError(("invalid counter accepted", field, name))
            count += 1
    for attempts in (0, 1, 2, 3):
        plan = decision(attempts=attempts)
        expected = (True, "retry", 0.25) if attempts < 3 else (False, "attempt_limit", 0.0)
        check((plan.retry, plan.reason, plan.delay) == expected, attempts)
        count += 1
    return count

async def trial(reader, cap):
    clock = VirtualClock()
    sends, waits, responses, trace = [], [], [], []
    async def handler(request):
        check(request.method == "GET" and str(request.url) == "https://fixture.test/read",
              "unexpected request")
        check("authorization" not in request.headers and "cookie" not in request.headers,
              "unexpected credentials")
        sends.append(clock.now)
        response = httpx.Response(429, headers={"Retry-After": "0"}, content=b"fixture")
        responses.append(response)
        return response
    async def sleep(delay):
        check(responses[-1].is_closed, "response must close before waiting")
        waits.append(delay)
        clock.advance(delay)
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as client:
        try:
            await reader(client, clock=clock, sleep=sleep, cancelled=asyncio.Event(),
                         deadline=10, max_attempts=cap, trace=trace)
        except ValueError as exc:
            outcome = ("ValueError", str(exc))
        except RetryStopped as exc:
            outcome = (exc.reason, exc.attempts, exc.status)
        else:
            raise AssertionError("429 fixture unexpectedly succeeded")
    check(all(response.is_closed for response in responses), "response leak")
    return outcome, sends, waits, clock.now, trace

async def main():
    pure = pure_checks()
    invalid = valid = 0
    for reader in (read_get, guarded_read):
        for name, cap in BAD_CAPS:
            actual = await trial(reader, cap)
            expected = (("ValueError", "max_attempts must be a positive integer"),
                        [], [], 0.0, [])
            check(actual == expected, (reader.__name__, name, actual, expected))
            invalid += 1
        for cap in (1, 2, 3):
            outcome, sends, waits, now, trace = await trial(reader, cap)
            check(outcome == ("attempt_limit", cap, 429), (reader.__name__, cap, outcome))
            check(sends == [i * 0.25 for i in range(cap)], (cap, sends))
            check(waits == [0.25] * (cap - 1), (cap, waits))
            check(now == (cap - 1) * 0.25, (cap, now))
            check([item["send"] for item in trace if "send" in item] == list(range(1, cap + 1)),
                  (cap, trace))
            valid += 1
    result = {"pure_cases": pure, "invalid_cap_no_send_cases": invalid,
              "valid_cap_cases": valid, "scope": "MockTransport and virtual time only"}
    Path(__file__).with_name("attempt-policy-results.json").write_text(
        json.dumps(result, indent=2) + "\n", encoding="utf-8")
    print("PASS: 26 pure counter cases; 22 invalid-cap no-send cases; 6 valid-cap cases")

if __name__ == "__main__":
    asyncio.run(main())

本轮使用已有的 Python 3.12.14 / HTTPX 0.28.1 / httpcore 1.0.9，没有安装依赖。新增 54 个用例在普通与 -O 模式均通过，原三套测试也全部通过；将新增测试放到修正前的源码旁，两种模式都会检出 NaN 上限被接受。本文内嵌文件重新提取后也完成相同复跑。所有请求仍由 MockTransport 处理并禁用 socket，时钟为合成时间，没有真实网络、浏览器 GUI、生产总超时或跨平台计时验证，也没有重新验收原有示意图。