Coverage for merco/core/llm/error_ui.py: 95%

1"""LLM error UI helpers — classification, rendering, retry feedback. 

2 

3Responsibility: what errors look like. NOT when to show them (callers decide). 

4Zero side effects: no logging, no openai import (duck-typing only). 

5""" 

6from __future__ import annotations 

7 

8import re 

9from contextlib import asynccontextmanager 

10from dataclasses import dataclass 

11from typing import AsyncIterator 

12 

13 

14_SENSITIVE_KEYWORDS = ("api_key", "token", "secret", "authorization", "bearer") 

15 

16 

17@dataclass 

18class ErrorInfo: 

19 """Classified error metadata.""" 

20 label: str # Short label e.g. "认证失败", "服务端错误 (502)" 

21 hint: str # One-line user-facing hint 

22 exc: Exception # Original exception 

23 

24 

25def classify_error(exc: Exception) -> ErrorInfo: 

26 """Classify an exception into ErrorInfo using duck-typing. 

27 

28 Reads ``exc.status_code`` if present; falls back to scanning str(exc) and 

29 the exception class name. Does NOT import openai. 

30 """ 

31 status = getattr(exc, "status_code", None) 

32 body = str(exc).lower() 

33 name = type(exc).__name__ 

34 

35 # 401 / unauthorized / invalid/expired api key 

36 if ( 

37 status == 401 

38 or "unauthorized" in body 

39 or "authentication" in body 

40 or ("api key" in body and ("invalid" in body or "expired" in body)) 

41 ): 

42 return ErrorInfo(label="认证失败", hint="API Key 无效或已过期，请检查配置。", exc=exc) 

43 

44 # 403 / forbidden / permission 

45 if ( 

46 status == 403 

47 or "forbidden" in body 

48 or "permission" in body 

49 or "access denied" in body 

50 ): 

51 return ErrorInfo(label="权限不足", hint="账户无权限访问该资源。", exc=exc) 

52 

53 # 404 / model not found 

54 if ( 

55 status == 404 

56 or ("not found" in body and "model" in body) 

57 or "model not exist" in body 

58 ): 

59 return ErrorInfo(label="模型/接口不存在", hint="模型或接口不存在，检查模型名和 base_url。", exc=exc) 

60 

61 # 413 / context length 

62 if ( 

63 status == 413 

64 or "context length" in body 

65 or "maximum context" in body 

66 or "prompt too long" in body 

67 or "too long" in body 

68 ): 

69 return ErrorInfo(label="请求过长", hint="上下文超过模型上限，正在压缩后重试…", exc=exc) 

70 

71 # 429 / rate limit 

72 if ( 

73 status == 429 

74 or "rate limit" in body 

75 or "too many requests" in body 

76 ): 

77 return ErrorInfo(label="请求限流", hint="API 限流，稍后重试…", exc=exc) 

78 

79 # 5xx server errors 

80 if isinstance(status, int) and 500 <= status <= 599: 

81 return ErrorInfo(label=f"服务端错误 ({status})", hint="API 服务器异常，稍后重试…", exc=exc) 

82 

83 # 408 timeout 

84 if isinstance(status, int) and status == 408: 

85 return ErrorInfo(label="请求超时", hint="API 请求超时，稍后重试…", exc=exc) 

86 

87 # Timeout by class name or body 

88 if "timeout" in name.lower() or "timeout" in body or "timed out" in body: 

89 return ErrorInfo(label="请求超时", hint="API 响应超时，稍后重试…", exc=exc) 

90 

91 # Connection errors 

92 if "connection" in name.lower() or "connect" in body or "network" in body: 

93 return ErrorInfo(label="连接错误", hint="无法连接到 API 服务器，请检查网络或 base_url。", exc=exc) 

94 

95 # Other 4xx client errors 

96 if isinstance(status, int) and 400 <= status < 500: 

97 return ErrorInfo(label=f"请求错误 ({status})", hint="请求被服务端拒绝，稍后重试…", exc=exc) 

98 

99 # Fallback: class name or generic label 

100 label = name or "调用失败" 

101 return ErrorInfo(label=label, hint="API 调用失败，稍后重试…", exc=exc) 

102 

103 

104def sanitize_message(exc: Exception, max_len: int = 300) -> str: 

105 """Redact sensitive keywords (api_key/token/secret/authorization/bearer) 

106 and truncate to max_len characters (appending '…' if truncated).""" 

107 msg = str(exc) 

108 

109 for kw in _SENSITIVE_KEYWORDS: 

110 if re.search(r'(?i)\b' + re.escape(kw) + r'\b', msg): 

111 return "(包含敏感信息，已脱敏)" 

112 

113 if len(msg) > max_len: 

114 return msg[:max_len].rstrip() + "…" 

115 

116 return msg 

117 

118 

119def build_error_panel(info: ErrorInfo): 

120 """Return a Rich Panel for an error. Red border, title '⚠ API 错误'. 

121 

122 Layout: 

123 ❌ <label> (bold red) 

124 <hint> (red) 

125 <blank line> 

126 <sanitized detail> (dim) 

127 """ 

128 from rich.panel import Panel 

129 from rich.text import Text 

130 detail = sanitize_message(info.exc) 

131 text = Text.from_markup( 

132 f"[bold red]❌ {info.label}[/bold red]\n" 

133 f"[red]{info.hint}[/red]\n\n" 

134 f"[dim]{detail}[/dim]" 

135 ) 

136 return Panel( 

137 text, border_style="red", 

138 title="⚠ API 错误", title_align="left", padding=(0, 1), 

139 ) 

140 

141 

142def build_retry_line(info: ErrorInfo, attempt: int, max_attempts: int, 

143 actions: list[str]) -> str: 

144 """Return a yellow one-line retry status string. 

145 

146 Example: "↻ API 请求限流（第 1/3 次）— 等待 3s + 压缩上下文…" 

147 """ 

148 action_str = " + ".join(actions) if actions else "立即重试" 

149 return (f"[yellow]↻ API {info.label}（第 {attempt}/{max_attempts} 次）" 

150 f"— {action_str}…[/yellow]") 

151 

152 

153@asynccontextmanager 

154async def retry_spinner(label: str, seconds: float, console): 

155 """Async context manager showing a transient spinner during a wait. 

156 

157 Usage:: 

158 

159 async with retry_spinner("请求限流", 3.0, console): 

160 await asyncio.sleep(3.0) 

161 

162 The spinner displays e.g. "⠋ 等待 2.3s 冷却中…" in yellow, updates at 8fps, 

163 and is transient (disappears on exit). If ``seconds <= 1`` it's a no-op 

164 (the wait is too short to justify a spinner). 

165 """ 

166 import asyncio 

167 import itertools 

168 from rich.live import Live 

169 from rich.text import Text 

170 

171 if seconds <= 1: 

172 yield 

173 return 

174 

175 spinner = itertools.cycle("⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏") 

176 

177 async def _tick(live, total): 

178 waited = 0.0 

179 try: 

180 while waited < total: 

181 await asyncio.sleep(0.1) 

182 waited += 0.1 

183 remain = max(0.0, total - waited) 

184 live.update(Text.from_markup( 

185 f"[yellow]{next(spinner)} 等待 {remain:.1f}s 冷却中…[/yellow]" 

186 )) 

187 except asyncio.CancelledError: 

188 pass 

189 

190 with Live(Text.from_markup(f"[yellow]{next(spinner)} 等待 {seconds:.1f}s 冷却中…[/yellow]"), 

191 console=console, refresh_per_second=8, transient=True) as live: 

192 ticker = asyncio.create_task(_tick(live, seconds)) 

193 try: 

194 yield 

195 finally: 

196 ticker.cancel() 

197 try: 

198 await ticker 

199 except asyncio.CancelledError: 

200 pass 

201 

202 

203def error_message(info: ErrorInfo) -> str: 

204 """Return a Rich-markup string starting with "❌ " for the final error. 

205 

206 Used as the agent's return value on terminal failure; the REPL detects 

207 startswith("❌") and wraps it in a red Panel via Text.from_markup. 

208 """ 

209 detail = sanitize_message(info.exc) 

210 return (f"❌ [bold red]{info.label}[/bold red]：[red]{info.hint}[/red]\n" 

211 f"[dim]{detail}[/dim]")