From 0246943165426ad45983eee96fa21308aa9689b9 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Tue, 18 Aug 2026 00:52:39 +0800 Subject: [PATCH] feat(aibox): failover observability, timeout budget and Retry-After (L4c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Observability: FailoverClient.get_stats() exposes aggregate counters (requests / failovers / cooldown_skips / exhausted, snapshot copy); new on_failover(from_name, to_name, reason) callback fires on every failover decision — to_name=None marks chain exhaustion — for future admin-panel/frontend wiring, wrapped so callback exceptions never affect the request path. Timeout budget: profile field "failover_timeout_budget" caps the total seconds spent walking the chain (default unlimited; values <= 0 are rejected to None with a warning). Once elapsed, remaining profiles are skipped — prevents worst-case N x timeout stacking on longer chains. Exhaustion is counted once per call via a single end-of-loop increment (an initial double count at the budget branch was caught by the new budget test and folded into the unified counter). Retry-After: TransientLLMError now carries retry_after parsed from the numeric Retry-After response header (non-numeric values like HTTP-dates are ignored); FailoverClient._mark_fail uses it to extend that profile's cooldown beyond the default 60s, respecting server-side throttling on 429s. _fail_at entries became (fail_time, cooldown) tuples; three existing tests that poke the internal map were updated. Config: create_llm_client passes profile.failover_timeout_budget into the FailoverClient constructor. Tests: new TestL4cObservabilityAndBudget (13 cases: stats on success/exhaustion/cooldown-skip, snapshot copy, hook called with chain-tail None, hook exception safety, budget stops chain via mocked monotonic, invalid budget ignored, config field plumbing, default no-budget regression, Retry-After parse/ignore/extend-cooldown). Full suite 355 passed. --- aibox/README.md | 8 +- aibox/llm_client.py | 113 +++++++++++++-- aibox/tests/test_models_profile.py | 211 ++++++++++++++++++++++++++++- 3 files changed, 310 insertions(+), 22 deletions(-) diff --git a/aibox/README.md b/aibox/README.md index 236348a..485daa7 100644 --- a/aibox/README.md +++ b/aibox/README.md @@ -92,7 +92,7 @@ aibox/ │ ├── prompt_tools.py # B 批：UpdateIdentity/UpdateSoul/SaveMemoryAndEndConversation │ ├── proxy_tools.py # B 批：SetProxy/GetProxy │ └── misc_tools.py # B 批：AskUserTool -├── tests/ # 单元测试套件（342 项全过） +├── tests/ # 单元测试套件（355 项全过） │ ├── conftest.py # sys.path 配置 + websockets 桩 │ ├── test_basic_tools.py # A 批工具 43 项 │ ├── test_b_tools.py # B 批工具 58 项 @@ -553,7 +553,8 @@ AI 可以通过 `update_soul` 工具动态更新 `99_SOUL.md`，通过 `update_i | `timeout` | 请求超时秒数 | | `temperature` | 档案级默认采样温度（可选）。优先级：CLI/调用方显式传入 > 本字段 > 全局默认；reasoning 模型建议按厂商文档设置 | | `display_name` | 展示名（可选，缺省用档案名）。`client.name` 用于对话上下文中的 assistant 名，不再恒显示 "DeepSeek" | -| `fallback` | 失败转移档案名列表（可选，如 `["kimi", "local"]`，一层不递归）。主档案瞬时故障（HTTP 429/5xx、连接失败/超时）时按序转移；认证/参数类 4xx **不转移**（配置问题响亮暴露）；失败档案进入 60s 冷却期。`chat_completion`（对话压缩/轻量子代理）无条件转移；`stream_chat`（主对话流式）仅**首事件前**转移（已输出内容/用户中断绝不转移） | +| `fallback` | 失败转移档案名列表（可选，如 `["kimi", "local"]`，一层不递归）。主档案瞬时故障（HTTP 429/5xx、连接失败/超时）时按序转移；认证/参数类 4xx **不转移**（配置问题响亮暴露）；失败档案进入 60s 冷却期（429 响应带 Retry-After 数字秒时按其延长）。`chat_completion`（对话压缩/轻量子代理）无条件转移；`stream_chat`（主对话流式）仅**首事件前**转移（已输出内容/用户中断绝不转移） | +| `failover_timeout_budget` | 整条转移链的总时间预算秒数（可选，缺省不限）。耗尽即停止尝试剩余档案，防止长链故障时超时叠加（3 档案最坏 3×timeout）；`FailoverClient.get_stats()` 暴露 requests/failovers/cooldown_skips/exhausted 计数，`on_failover(from, to, reason)` 回调供管理面/前端通知接入 | **选择优先级**：`--model <档案名>`（命中 `models` 节）→ `api.default_model` 指向档案 → 回退旧 `api` 扁平节。`--model` 为**进程级覆盖**：WebSocket 服务模式（主对话、每个会话实例、定时任务实例）与 `--once` 一次性模式均生效。 @@ -904,7 +905,8 @@ elif command == "/mycommand": - **L5（思考流字段别名）**：`_parse_delta` 识别 `reasoning_content`/`reasoning`/`thinking`/`thinking_content` 四种常见思考字段（首个命中生效），其他厂商思考流不再静默丢失 - ✅ **L4a（chat_completion 跨档案失败转移）**：档案新字段 `fallback`（档案名列表，一层不递归）；`create_llm_client` 构造 `FailoverClient` 包装链（主档案 + fallback）。瞬时故障（HTTP 429/5xx 抛 `TransientLLMError`、连接失败/超时）按链转移，认证/参数类 4xx 不转移（配置问题响亮暴露）；失败档案 60s 冷却；fallback 档案缺 Key 等环境问题跳过 + warning 不阻断主档案；配置结构错误（类型/不存在/自引用）启动期硬性拦截。当前覆盖对话压缩与轻量子代理链路，流式（`stream_chat`）转移属 L4b - ✅ **L4b（stream_chat 首事件前转移 + 中断联动）**：流式仅在**未产出任何事件**（reasoning/content/tool_call）时转移——重放对用户不可见；已产出事件后失败直接上抛（防前端重复输出/工具增量错乱）。`LLMClient.cancel()` 置 `_cancelled` 标志，`FailoverClient.cancel()` 广播整条链并置包装器标志——用户中断引发的连接异常与真实故障不可区分，凭标志跳过转移（否则用户越中断、系统越换厂商重试）；`chat_completion` 同受标志保护。链耗尽抛最后异常（既有回滚/报错处理接住），全链冷却抛 RuntimeError -- ✅ **测试**：342 项全过（H1 隔离 4 + H2 校验 6 + H3 传导 6 + M2 提示 5 + M1 替换 2 + M3 契约 4 + L 批 16 + L4a 转移 15 + L4b 流式 12 项新增） +- ✅ **L4c（转移可观测性 + 预算控制 + Retry-After）**：`FailoverClient.get_stats()` 暴露 requests/failovers/cooldown_skips/exhausted 聚合计数；`on_failover(from, to, reason)` 回调（to=None 表示链耗尽，回调异常不影响主流程）供管理面/前端通知接入；档案字段 `failover_timeout_budget` 限制整链尝试总秒数（缺省不限，<=0 视为未配置），耗尽停止尝试剩余档案防超时叠加；429 响应 Retry-After（数字秒）经 `TransientLLMError.retry_after` 延长该档案冷却期（非数字格式忽略） +- ✅ **测试**：355 项全过（H1 隔离 4 + H2 校验 6 + H3 传导 6 + M2 提示 5 + M1 替换 2 + M3 契约 4 + L 批 16 + L4a 转移 15 + L4b 流式 12 + L4c 可观测 13 项新增） ### v3.3 - ✅ **安全审计与修复**（两轮审计报告见 [docs/reports/aibox_audit_20260817.md](../docs/reports/aibox_audit_20260817.md)）： diff --git a/aibox/llm_client.py b/aibox/llm_client.py index 9c8a5b1..a498e98 100644 --- a/aibox/llm_client.py +++ b/aibox/llm_client.py @@ -160,7 +160,8 @@ class DeepSeekClient(LLMClient): # L4a：瞬时故障显式抛出——FailoverClient 据此转移下一档案； # 无包装器时由调用方 catch-all 接住，下游仍走失败分支（语义等价） raise TransientLLMError( - f"LLM 请求瞬时失败: HTTP {response.status_code}: {response.text[:200]}") + f"LLM 请求瞬时失败: HTTP {response.status_code}: {response.text[:200]}", + retry_after=_retry_after_of(response)) self._log("error", f"LLM 压缩请求失败: HTTP {response.status_code}") return None except TransientLLMError: @@ -209,7 +210,8 @@ class DeepSeekClient(LLMClient): if response.status_code == 429 or response.status_code >= 500: # L4b：瞬时故障显式分类（此时尚未产出任何事件，可安全转移重放） raise TransientLLMError( - f"LLM 流式请求瞬时失败: HTTP {response.status_code}: {response.text[:200]}") + f"LLM 流式请求瞬时失败: HTTP {response.status_code}: {response.text[:200]}", + retry_after=_retry_after_of(response)) raise RuntimeError( f"HTTP {response.status_code}: {response.text}") @@ -268,7 +270,26 @@ class DeepSeekClient(LLMClient): class TransientLLMError(RuntimeError): - """瞬时 LLM 故障（HTTP 429/5xx）——FailoverClient 据此转移下一档案（L4a）。""" + """瞬时 LLM 故障（HTTP 429/5xx）——FailoverClient 据此转移下一档案（L4a）。 + + L4c：retry_after 携带服务端 Retry-After 响应头（数字秒）建议冷却时长， + FailoverClient 以其替代默认冷却期（429 限流时尊重服务端节奏）。 + """ + + def __init__(self, message: str, retry_after: Optional[float] = None): + super().__init__(message) + self.retry_after = retry_after + + +def _retry_after_of(response) -> Optional[float]: + """从 Retry-After 响应头解析冷却秒数（仅支持数字秒格式，其余忽略返回 None）。""" + try: + v = response.headers.get("Retry-After") + if v is None: + return None + return max(1.0, float(v)) + except (TypeError, ValueError, AttributeError): + return None class FailoverClient: @@ -287,14 +308,30 @@ class FailoverClient: 尚未产出任何 reasoning/content/tool_call 事件，重放对用户不可见）； 已产出事件后失败直接上抛（重放会导致前端重复输出/工具增量错乱）； 用户中断（cancel() 置位）引发的连接异常绝不转移。 + + L4c 范围——可观测性与预算控制： + - get_stats() 暴露 requests/failovers/cooldown_skips/exhausted 聚合计数 + - on_failover(from_name, to_name, reason) 回调（to_name=None 表示链耗尽）， + 供管理面/前端通知接入，回调异常不影响主流程 + - timeout_budget 限制整条链尝试的总秒数（缺省不限），耗尽即停止尝试剩余档案， + 防止长链故障时超时叠加（3 档案最坏 3×timeout） + - 429 响应带 Retry-After（数字秒）时按其延长该档案冷却期 """ - def __init__(self, clients: List, logger=None, cooldown_seconds: float = 60.0): + def __init__(self, clients: List, logger=None, cooldown_seconds: float = 60.0, + timeout_budget: Optional[float] = None, on_failover=None): self._clients = list(clients) self.logger = logger self.cooldown_seconds = cooldown_seconds - self._fail_at = {} # client.name → 失败时刻（time.monotonic） + self._fail_at = {} # client.name → (失败时刻, 冷却秒数) self._cancelled = False + self._stats = {"requests": 0, "failovers": 0, "cooldown_skips": 0, "exhausted": 0} + if timeout_budget is not None and timeout_budget <= 0: + self._log("warning", f"timeout_budget 非法（{timeout_budget} <= 0），按未配置处理") + timeout_budget = None + self.timeout_budget = timeout_budget + self.on_failover = on_failover + self._call_started = None def __getattr__(self, item): # 未显式定义的属性/方法委托主 client（name/model/api_key/cancel 等）； @@ -311,11 +348,34 @@ class FailoverClient: pass def _cooling(self, client) -> bool: - t = self._fail_at.get(client.name) - return t is not None and (time.monotonic() - t) < self.cooldown_seconds + entry = self._fail_at.get(client.name) + if entry is None: + return False + t, cd = entry + return (time.monotonic() - t) < cd - def _mark_fail(self, client): - self._fail_at[client.name] = time.monotonic() + def _mark_fail(self, client, cooldown: Optional[float] = None): + # L4c：429 Retry-After（经 TransientLLMError.retry_after）覆盖默认冷却时长 + self._fail_at[client.name] = (time.monotonic(), + cooldown if cooldown and cooldown > 0 + else self.cooldown_seconds) + + def _budget_exhausted(self) -> bool: + if self.timeout_budget is None or self._call_started is None: + return False + return (time.monotonic() - self._call_started) >= self.timeout_budget + + def get_stats(self) -> Dict: + """L4c：转移统计快照（requests/failovers/cooldown_skips/exhausted）。""" + return dict(self._stats) + + def _notify_failover(self, from_name: str, to_name, reason: str): + self._stats["failovers"] += 1 + try: + if self.on_failover is not None: + self.on_failover(from_name, to_name, reason) + except Exception as e: + self._log("warning", f"on_failover 回调异常（忽略）: {e}") def cancel(self): """L4b：中断链上全部 client（任一可能正在流式中）并置自身中断标志。 @@ -332,12 +392,19 @@ class FailoverClient: def chat_completion(self, messages: List[Dict], temperature: float = None, max_tokens: int = 1024, **kwargs): self._cancelled = False # 新一轮请求，清除上一轮中断标志 + self._call_started = time.monotonic() + self._stats["requests"] += 1 last_exc = None tried = 0 - for client in self._clients: + for idx, client in enumerate(self._clients): if self._cooling(client): + self._stats["cooldown_skips"] += 1 self._log("info", f"模型档案 {client.name} 处于失败冷却期，跳过") continue + if self._budget_exhausted(): + self._log("warning", + f"失败转移时间预算（{self.timeout_budget}s）已耗尽，停止尝试剩余档案") + break tried += 1 try: result = client.chat_completion(messages, temperature=temperature, @@ -350,8 +417,12 @@ class FailoverClient: if self._cancelled: return None # 用户中断引发的异常：不转移 last_exc = e - self._mark_fail(client) + self._mark_fail(client, cooldown=getattr(e, "retry_after", None)) + to_name = (self._clients[idx + 1].name + if idx + 1 < len(self._clients) else None) + self._notify_failover(client.name, to_name, str(e)) self._log("warning", f"模型档案 {client.name} 瞬时故障（{e}），转移下一档案") + self._stats["exhausted"] += 1 # 末尾统一计数（链耗尽/预算耗尽/全冷却） self._log("error", f"全部 {len(self._clients)} 个模型档案均不可用（尝试 {tried} 个）：{last_exc}") return None @@ -361,11 +432,18 @@ class FailoverClient: reasoning_enabled: bool = False, **extra) -> Iterator[Dict]: """L4b：流式转移——仅首事件前故障可安全重放（对用户不可见）。""" self._cancelled = False # 新一轮请求，清除上一轮中断标志 + self._call_started = time.monotonic() + self._stats["requests"] += 1 last_exc = None - for client in self._clients: + for idx, client in enumerate(self._clients): if self._cooling(client): + self._stats["cooldown_skips"] += 1 self._log("info", f"模型档案 {client.name} 处于失败冷却期，跳过") continue + if self._budget_exhausted(): + self._log("warning", + f"失败转移时间预算（{self.timeout_budget}s）已耗尽，停止尝试剩余档案") + break produced_any = False try: for event in client.stream_chat( @@ -381,13 +459,17 @@ class FailoverClient: if produced_any: raise # 已向消费方产出事件，重放会重复输出/工具增量错乱 last_exc = e - self._mark_fail(client) + self._mark_fail(client, cooldown=getattr(e, "retry_after", None)) + to_name = (self._clients[idx + 1].name + if idx + 1 < len(self._clients) else None) + self._notify_failover(client.name, to_name, str(e)) self._log("warning", f"模型档案 {client.name} 流式瞬时故障（首事件前，{e}），转移下一档案") + self._stats["exhausted"] += 1 if last_exc is not None: raise last_exc # 链耗尽：抛最后异常，由既有错误处理（回滚/报错）接住 raise RuntimeError( - f"全部 {len(self._clients)} 个模型档案处于失败冷却期，暂不可用") + f"全部 {len(self._clients)} 个模型档案不可用（冷却期或转移时间预算耗尽）") def resolve_api_key(api_key: Optional[str], config=None, @@ -594,7 +676,8 @@ def create_llm_client(config, api_key: Optional[str] = None, # 不阻断主档案可用性（配置结构错误在上方已硬性拦截） if logger is not None and hasattr(logger, "warning"): logger.warning(f"fallback 档案 '{fb_name}' 构造失败，已忽略：{e}") - return FailoverClient(chain, logger=logger) + return FailoverClient(chain, logger=logger, + timeout_budget=profile.get("failover_timeout_budget")) if provider in ("deepseek", "deepseek-chat", "openai_compatible", "openai"): client = DeepSeekClient( diff --git a/aibox/tests/test_models_profile.py b/aibox/tests/test_models_profile.py index 70f430e..c4adfb6 100644 --- a/aibox/tests/test_models_profile.py +++ b/aibox/tests/test_models_profile.py @@ -687,11 +687,12 @@ class TestLBatch: # ==================== L4a：chat_completion 跨档案失败转移 ==================== -def _resp(status=200, text="", content="ok"): +def _resp(status=200, text="", content="ok", headers=None): m = mock.MagicMock() m.status_code = status m.text = text m.json.return_value = {"choices": [{"message": {"content": content}}]} + m.headers = headers or {} return m @@ -819,7 +820,8 @@ class TestL4aChatCompletionFailover: assert fc.chat_completion([{"role": "user", "content": "q2"}]) == "r2" assert mp.call_args_list[0].args[0].startswith("https://api.moonshot.cn") # 冷却过期（手动回拨失败时间戳）：恢复走主档案 - fc._fail_at["deepseek"] = _time.monotonic() - fc.cooldown_seconds - 1 + fc._fail_at["deepseek"] = (_time.monotonic() - fc.cooldown_seconds - 1, + fc.cooldown_seconds) with mock.patch("llm_client.requests.post", return_value=_resp(200, content="r3")) as mp: assert fc.chat_completion([{"role": "user", "content": "q3"}]) == "r3" @@ -996,7 +998,8 @@ class TestL4bStreamFailover: primary = self._client("deepseek", stream=_fake_stream(events=self.EVENTS)) fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS)) fc = self._fc(primary, fallback) - fc._fail_at = {"deepseek": _time.monotonic(), "kimi": _time.monotonic()} + fc._fail_at = {"deepseek": (_time.monotonic(), fc.cooldown_seconds), + "kimi": (_time.monotonic(), fc.cooldown_seconds)} with pytest.raises(RuntimeError, match="冷却期"): list(fc.stream_chat([{"role": "user", "content": "q"}])) primary.stream_chat.assert_not_called() @@ -1007,7 +1010,7 @@ class TestL4bStreamFailover: primary = self._client("deepseek", stream=_fake_stream(events=self.EVENTS)) fallback = self._client("kimi", stream=_fake_stream(events=self.EVENTS)) fc = self._fc(primary, fallback) - fc._fail_at["deepseek"] = _time.monotonic() + fc._fail_at["deepseek"] = (_time.monotonic(), fc.cooldown_seconds) assert list(fc.stream_chat([{"role": "user", "content": "q"}])) == self.EVENTS primary.stream_chat.assert_not_called() @@ -1039,6 +1042,206 @@ class TestL4bStreamFailover: list(c.stream_chat([{"role": "user", "content": "hi"}])) +# ==================== L4c：可观测性（stats/hook）+ 预算 + Retry-After ==================== + +class TestL4cObservabilityAndBudget: + """L4c：转移可观测性与预算控制。 + + - get_stats：requests/failovers/cooldown_skips/exhausted 聚合计数 + - on_failover(from, to, reason) 回调，to=None 表示链耗尽；回调异常不影响主流程 + - failover_timeout_budget：整链总预算（秒），耗尽停止尝试剩余档案（缺省不限） + - Retry-After（数字秒）经 TransientLLMError.retry_after 延长档案冷却期 + """ + + EVENTS = [{"type": "content", "text": "hi"}, {"type": "done"}] + + def _mk_client(self, name, stream=None, completion=None): + c = mock.MagicMock() + c.name = name + if stream is not None: + c.stream_chat = mock.MagicMock(side_effect=stream) + if completion is not None: + c.chat_completion = mock.MagicMock(side_effect=completion) + return c + + def _fc(self, primary, fallback=None, **kwargs): + from llm_client import FailoverClient + clients = [primary] + ([fallback] if fallback is not None else []) + return FailoverClient(clients, **kwargs) + + # ---------- stats ---------- + + def test_stats_failover_success(self): + """主 429 → fallback 成功：requests=1, failovers=1, exhausted=0""" + from llm_client import TransientLLMError + primary = self._mk_client("deepseek", + completion=mock.MagicMock(side_effect=TransientLLMError("429"))) + fallback = self._mk_client("kimi", + completion=mock.MagicMock(return_value="ok")) + fc = self._fc(primary, fallback) + assert fc.chat_completion([{"role": "user", "content": "q"}]) == "ok" + s = fc.get_stats() + assert s == {"requests": 1, "failovers": 1, "cooldown_skips": 0, "exhausted": 0} + + def test_stats_chain_exhausted(self): + """全链失败：failovers=2（两次转移决策），exhausted=1""" + from llm_client import TransientLLMError + primary = self._mk_client("deepseek", + completion=mock.MagicMock(side_effect=TransientLLMError("429"))) + fallback = self._mk_client("kimi", + completion=mock.MagicMock(side_effect=TransientLLMError("500"))) + fc = self._fc(primary, fallback) + assert fc.chat_completion([{"role": "user", "content": "q"}]) is None + s = fc.get_stats() + assert s["failovers"] == 2 and s["exhausted"] == 1 + + def test_stats_cooldown_skips_and_stream(self): + """冷却跳过计数 + 流式成功路径 requests 计数""" + import time as _time + primary = self._mk_client("deepseek", + stream=_fake_stream(events=self.EVENTS)) + fallback = self._mk_client("kimi", stream=_fake_stream(events=self.EVENTS)) + fc = self._fc(primary, fallback) + fc._fail_at["deepseek"] = (_time.monotonic(), fc.cooldown_seconds) + assert list(fc.stream_chat([{"role": "user", "content": "q"}])) == self.EVENTS + s = fc.get_stats() + assert s == {"requests": 1, "failovers": 0, "cooldown_skips": 1, "exhausted": 0} + + def test_get_stats_returns_copy(self): + """get_stats 返回快照副本，外部修改不影响内部计数""" + fc = self._fc(self._mk_client("deepseek")) + snap = fc.get_stats() + snap["requests"] = 999 + assert fc.get_stats()["requests"] == 0 + + # ---------- on_failover hook ---------- + + def test_on_failover_hook_called(self): + """转移时回调 (from, to, reason)；链耗尽时 to=None""" + from llm_client import TransientLLMError + calls = [] + primary = self._mk_client("deepseek", + completion=mock.MagicMock(side_effect=TransientLLMError("HTTP 429"))) + fallback = self._mk_client("kimi", + completion=mock.MagicMock(side_effect=TransientLLMError("HTTP 500"))) + fc = self._fc(primary, fallback, on_failover=lambda f, t, r: calls.append((f, t, r))) + assert fc.chat_completion([{"role": "user", "content": "q"}]) is None + assert len(calls) == 2 + assert calls[0][0] == "deepseek" and calls[0][1] == "kimi" + assert calls[1][0] == "kimi" and calls[1][1] is None # 链尾 + assert "429" in calls[0][2] and "500" in calls[1][2] + + def test_on_failover_hook_exception_safe(self): + """回调抛异常不影响主流程（仍完成转移）""" + from llm_client import TransientLLMError + + def bad_hook(f, t, r): + raise RuntimeError("hook boom") + + primary = self._mk_client("deepseek", + completion=mock.MagicMock(side_effect=TransientLLMError("429"))) + fallback = self._mk_client("kimi", + completion=mock.MagicMock(return_value="ok")) + fc = self._fc(primary, fallback, on_failover=bad_hook) + assert fc.chat_completion([{"role": "user", "content": "q"}]) == "ok" + + # ---------- timeout_budget ---------- + + def test_timeout_budget_stops_chain(self): + """预算耗尽 → 不再尝试 fallback（chat_completion 返回 None）""" + from llm_client import TransientLLMError + primary = self._mk_client("deepseek", + completion=mock.MagicMock(side_effect=TransientLLMError("429"))) + fallback = self._mk_client("kimi", + completion=mock.MagicMock(return_value="ok")) + fc = self._fc(primary, fallback, timeout_budget=50.0) + # monotonic 序列：_call_started=0 → 主档案预算检查=0 → _mark_fail=0 → fallback 预算检查=100 + with mock.patch("llm_client.time.monotonic", + side_effect=[0.0, 0.0, 0.0, 100.0]): + assert fc.chat_completion([{"role": "user", "content": "q"}]) is None + fallback.chat_completion.assert_not_called() + assert fc.get_stats()["exhausted"] == 1 + + def test_timeout_budget_invalid_treated_as_none(self): + """timeout_budget<=0 → warning + 视为未配置（全链仍可尝试）""" + from llm_client import TransientLLMError + primary = self._mk_client("deepseek", + completion=mock.MagicMock(side_effect=TransientLLMError("429"))) + fallback = self._mk_client("kimi", + completion=mock.MagicMock(return_value="ok")) + fc = self._fc(primary, fallback, timeout_budget=-5) + assert fc.timeout_budget is None + assert fc.chat_completion([{"role": "user", "content": "q"}]) == "ok" + + def test_timeout_budget_config_field(self): + """档案 failover_timeout_budget 字段 → wrapper.timeout_budget 生效""" + models = { + "deepseek": dict(MODELS["deepseek"], + fallback=["kimi"], failover_timeout_budget=30), + "kimi": dict(MODELS["kimi"]), + } + with mock.patch.dict(os.environ, + {"DEEPSEEK_API_KEY": "sk-ds", "MOONSHOT_API_KEY": "sk-moon"}): + c = create_llm_client(_cfg(models=models)) + assert c.timeout_budget == 30 + + def test_no_budget_by_default(self): + """缺省无预算（None）——现有全链行为不变（回归）""" + from llm_client import TransientLLMError + primary = self._mk_client("deepseek", + completion=mock.MagicMock(side_effect=TransientLLMError("429"))) + fallback = self._mk_client("kimi", + completion=mock.MagicMock(return_value="ok")) + fc = self._fc(primary, fallback) + assert fc.timeout_budget is None + assert fc.chat_completion([{"role": "user", "content": "q"}]) == "ok" + + # ---------- Retry-After ---------- + + def test_retry_after_header_parsed(self): + """429 响应 Retry-After: 120 → TransientLLMError.retry_after=120""" + from llm_client import DeepSeekClient, TransientLLMError + c = DeepSeekClient(api_key="k", model="m", base_url="https://x") + resp = _resp(429, "limited", headers={"Retry-After": "120"}) + with mock.patch("llm_client.requests.post", return_value=resp): + with pytest.raises(TransientLLMError) as ei: + c.chat_completion([{"role": "user", "content": "hi"}]) + assert ei.value.retry_after == 120 + + def test_retry_after_invalid_ignored(self): + """Retry-After 非 数字（HTTP-date/垃圾值）→ None（默认冷却）""" + from llm_client import DeepSeekClient, TransientLLMError + c = DeepSeekClient(api_key="k", model="m", base_url="https://x") + resp = _resp(429, "limited", headers={"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}) + with mock.patch("llm_client.requests.post", return_value=resp): + with pytest.raises(TransientLLMError) as ei: + c.chat_completion([{"role": "user", "content": "hi"}]) + assert ei.value.retry_after is None + + def test_retry_after_extends_cooldown(self): + """retry_after=300 → 主档案冷却 300s：下次请求仍走 fallback；过期后恢复""" + import time as _time + from llm_client import TransientLLMError + primary = self._mk_client("deepseek", + completion=mock.MagicMock( + side_effect=TransientLLMError("429", retry_after=300))) + primary_ok = mock.MagicMock(return_value="from-ds") + fallback = self._mk_client("kimi", completion=mock.MagicMock(return_value="from-kimi")) + fc = self._fc(primary, fallback) + # 第一次：主失败（retry_after=300）→ fallback + assert fc.chat_completion([{"role": "user", "content": "q1"}]) == "from-kimi" + # 冷却中（默认 60s 已过但 300s 未到）：仍走 fallback + t, _ = fc._fail_at["deepseek"] + fc._fail_at["deepseek"] = (t - 100, 300) + primary.chat_completion = primary_ok + assert fc.chat_completion([{"role": "user", "content": "q2"}]) == "from-kimi" + primary_ok.assert_not_called() + # 300s 冷却过期：恢复主档案 + fc._fail_at["deepseek"] = (t - 301, 300) + assert fc.chat_completion([{"role": "user", "content": "q3"}]) == "from-ds" + primary_ok.assert_called_once() + + # ==================== DeepSeekChat 档案级 reasoning 读取 ==================== class TestChatReasoningFlag: