Coverage for agentos/models/error.py: 60%

1"""AgentOS Error Models — typed HTTP exceptions for the API layer. 

2 

3Provides a hierarchy of FastAPI-compatible HTTP exceptions with: 

4- Machine-readable error codes 

5- RFC 9457 problem detail fields 

6- Structured validation error accumulation 

7- Built-in logging integration 

8""" 

9 

10from __future__ import annotations 

11 

12import logging 

13from enum import Enum 

14from typing import Any, Dict, List, Optional 

15 

16from fastapi import HTTPException 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21class ErrorCode(str, Enum): 

22 """Machine-readable error codes for API responses.""" 

23 

24 # General 

25 INTERNAL_ERROR = "INTERNAL_ERROR" 

26 SERVICE_UNAVAILABLE = "SERVICE_UNAVAILABLE" 

27 NOT_IMPLEMENTED = "NOT_IMPLEMENTED" 

28 

29 # Validation 

30 VALIDATION_ERROR = "VALIDATION_ERROR" 

31 INVALID_INPUT = "INVALID_INPUT" 

32 MISSING_REQUIRED = "MISSING_REQUIRED" 

33 TYPE_MISMATCH = "TYPE_MISMATCH" 

34 VALUE_OUT_OF_RANGE = "VALUE_OUT_OF_RANGE" 

35 

36 # Auth 

37 UNAUTHENTICATED = "UNAUTHENTICATED" 

38 UNAUTHORIZED = "UNAUTHORIZED" 

39 TOKEN_EXPIRED = "TOKEN_EXPIRED" 

40 TOKEN_INVALID = "TOKEN_INVALID" 

41 INSUFFICIENT_SCOPE = "INSUFFICIENT_SCOPE" 

42 

43 # Resource 

44 NOT_FOUND = "NOT_FOUND" 

45 CONFLICT = "CONFLICT" 

46 ALREADY_EXISTS = "ALREADY_EXISTS" 

47 GONE = "GONE" 

48 

49 # Rate limiting 

50 RATE_LIMITED = "RATE_LIMITED" 

51 QUOTA_EXCEEDED = "QUOTA_EXCEEDED" 

52 

53 # Agent-specific 

54 AGENT_NOT_FOUND = "AGENT_NOT_FOUND" 

55 AGENT_RUN_FAILED = "AGENT_RUN_FAILED" 

56 AGENT_TIMEOUT = "AGENT_TIMEOUT" 

57 TOOL_NOT_FOUND = "TOOL_NOT_FOUND" 

58 TOOL_EXECUTION_FAILED = "TOOL_EXECUTION_FAILED" 

59 

60 # Model / LLM 

61 MODEL_NOT_AVAILABLE = "MODEL_NOT_AVAILABLE" 

62 MODEL_TIMEOUT = "MODEL_TIMEOUT" 

63 CONTEXT_LENGTH_EXCEEDED = "CONTEXT_LENGTH_EXCEEDED" 

64 CONTENT_FILTER = "CONTENT_FILTER" 

65 

66 

67# ============================================================================ 

68# Base error 

69# ============================================================================ 

70 

71class AgentOSError(HTTPException): 

72 """Base exception for all AgentOS API errors. 

73 

74 Extends FastAPI's HTTPException with structured error codes 

75 and optional field-level details. 

76 """ 

77 

78 def __init__( 

79 self, 

80 status_code: int, 

81 code: ErrorCode, 

82 detail: str = "", 

83 field: Optional[str] = None, 

84 headers: Optional[Dict[str, str]] = None, 

85 ): 

86 super().__init__(status_code=status_code, detail=detail, headers=headers) 

87 self.code = code 

88 self.field = field 

89 self._log() 

90 

91 def _log(self): 

92 """Log the error at appropriate level.""" 

93 log_msg = f"[{self.code.value}] {self.detail}" 

94 if self.field: 

95 log_msg += f" (field: {self.field})" 

96 if self.status_code >= 500: 

97 logger.error(log_msg) 

98 else: 

99 logger.warning(log_msg) 

100 

101 def to_api_error(self) -> dict: 

102 """Convert to APIErrorDetail-compatible dict.""" 

103 return { 

104 "type": f"https://errors.agentos.dev/{self.code.value.lower()}", 

105 "title": self._title(), 

106 "status": self.status_code, 

107 "detail": self.detail, 

108 "code": self.code.value, 

109 "field": self.field, 

110 } 

111 

112 def _title(self) -> str: 

113 titles = { 

114 400: "Bad Request", 

115 401: "Unauthorized", 

116 403: "Forbidden", 

117 404: "Not Found", 

118 409: "Conflict", 

119 429: "Too Many Requests", 

120 500: "Internal Server Error", 

121 503: "Service Unavailable", 

122 } 

123 return titles.get(self.status_code, "Error") 

124 

125 

126# ============================================================================ 

127# Concrete error types 

128# ============================================================================ 

129 

130class ValidationError(AgentOSError): 

131 """422 — Input validation failed. 

132 

133 Supports accumulating multiple field errors via add_error(). 

134 """ 

135 

136 def __init__( 

137 self, 

138 detail: str = "Validation failed", 

139 field: Optional[str] = None, 

140 errors: Optional[List[Dict[str, str]]] = None, 

141 ): 

142 super().__init__( 

143 status_code=422, 

144 code=ErrorCode.VALIDATION_ERROR, 

145 detail=detail, 

146 field=field, 

147 ) 

148 self.errors: List[Dict[str, str]] = errors or [] 

149 if field and detail: 

150 self.errors.append({"field": field, "message": detail}) 

151 

152 def add_error(self, field: str, message: str) -> None: 

153 """Accumulate an additional field error.""" 

154 self.errors.append({"field": field, "message": message}) 

155 if not self.detail: 

156 self.detail = f"Validation error on '{field}': {message}" 

157 

158 @property 

159 def has_errors(self) -> bool: 

160 return len(self.errors) > 0 

161 

162 def to_dict(self) -> dict: 

163 return { 

164 **self.to_api_error(), 

165 "errors": self.errors, 

166 } 

167 

168 

169class NotFoundError(AgentOSError): 

170 """404 — Resource not found.""" 

171 

172 def __init__(self, resource_type: str = "Resource", identifier: str = ""): 

173 detail = f"{resource_type} not found" 

174 if identifier: 

175 detail = f"{resource_type} '{identifier}' not found" 

176 super().__init__( 

177 status_code=404, 

178 code=ErrorCode.NOT_FOUND, 

179 detail=detail, 

180 ) 

181 

182 

183class AuthenticationError(AgentOSError): 

184 """401 — Missing or invalid credentials.""" 

185 

186 def __init__(self, detail: str = "Authentication required"): 

187 super().__init__( 

188 status_code=401, 

189 code=ErrorCode.UNAUTHENTICATED, 

190 detail=detail, 

191 ) 

192 

193 

194class AuthorizationError(AgentOSError): 

195 """403 — Insufficient permissions.""" 

196 

197 def __init__(self, detail: str = "Insufficient permissions"): 

198 super().__init__( 

199 status_code=403, 

200 code=ErrorCode.UNAUTHORIZED, 

201 detail=detail, 

202 ) 

203 

204 

205class RateLimitError(AgentOSError): 

206 """429 — Too many requests. 

207 

208 Includes retry-after header when available. 

209 """ 

210 

211 def __init__( 

212 self, 

213 detail: str = "Rate limit exceeded", 

214 retry_after: Optional[int] = None, 

215 ): 

216 headers = {} 

217 if retry_after is not None: 

218 headers["Retry-After"] = str(retry_after) 

219 super().__init__( 

220 status_code=429, 

221 code=ErrorCode.RATE_LIMITED, 

222 detail=detail, 

223 headers=headers, 

224 ) 

225 self.retry_after = retry_after 

226 

227 

228class InternalError(AgentOSError): 

229 """500 — Unexpected internal error. User-safe, no stack traces exposed.""" 

230 

231 def __init__(self, detail: str = "An unexpected error occurred"): 

232 super().__init__( 

233 status_code=500, 

234 code=ErrorCode.INTERNAL_ERROR, 

235 detail=detail, 

236 ) 

237 

238 

239class ServiceUnavailableError(AgentOSError): 

240 """503 — Service temporarily unavailable (e.g., during maintenance).""" 

241 

242 def __init__( 

243 self, 

244 detail: str = "Service temporarily unavailable", 

245 retry_after: Optional[int] = None, 

246 ): 

247 headers = {} 

248 if retry_after is not None: 

249 headers["Retry-After"] = str(retry_after) 

250 super().__init__( 

251 status_code=503, 

252 code=ErrorCode.SERVICE_UNAVAILABLE, 

253 detail=detail, 

254 headers=headers, 

255 )