Coverage for agentos/hitl/approver.py: 0%

1""" 

2Human-in-the-Loop approval engine — request construction, risk assessment, 

3policy evaluation, and decision processing. 

4""" 

5 

6from collections.abc import Callable 

7from dataclasses import dataclass, field 

8from enum import StrEnum 

9from typing import Any 

10 

11 

12class ApprovalStatus(StrEnum): 

13 """Status of an approval request.""" 

14 

15 PENDING = "pending" 

16 APPROVED = "approved" 

17 REJECTED = "rejected" 

18 MODIFIED = "modified" 

19 TIMED_OUT = "timed_out" 

20 SKIPPED = "skipped" 

21 

22 

23class RiskLevel(StrEnum): 

24 """Risk classification for approval decisions.""" 

25 

26 LOW = "low" 

27 MEDIUM = "medium" 

28 HIGH = "high" 

29 CRITICAL = "critical" 

30 

31 

32@dataclass 

33class ApprovalRequest: 

34 """A structured request for human approval.""" 

35 

36 request_id: str 

37 action: str 

38 description: str 

39 risk_level: RiskLevel = RiskLevel.MEDIUM 

40 tool_name: str = "" 

41 tool_args: dict[str, Any] = field(default_factory=dict) 

42 estimated_cost_usd: float = 0.0 

43 data_affected: list[str] = field(default_factory=list) 

44 context: dict[str, Any] = field(default_factory=dict) 

45 metadata: dict[str, Any] = field(default_factory=dict) 

46 

47 

48@dataclass 

49class ApprovalDecision: 

50 """Human decision on an approval request.""" 

51 

52 request_id: str 

53 status: ApprovalStatus 

54 reason: str = "" 

55 modified_args: dict[str, Any] | None = None 

56 metadata: dict[str, Any] = field(default_factory=dict) 

57 

58 @property 

59 def is_approved(self) -> bool: 

60 return self.status in (ApprovalStatus.APPROVED, ApprovalStatus.MODIFIED) 

61 

62 @property 

63 def is_rejected(self) -> bool: 

64 return self.status == ApprovalStatus.REJECTED 

65 

66 

67@dataclass 

68class ApprovalPolicy: 

69 """Configures which actions require human approval.""" 

70 

71 require_approval_for_risk: set[RiskLevel] = field( 

72 default_factory=lambda: {RiskLevel.HIGH, RiskLevel.CRITICAL} 

73 ) 

74 auto_approve_domains: set[str] = field(default_factory=set) 

75 block_domains: set[str] = field(default_factory=set) 

76 max_auto_approve_cost_usd: float = 0.01 

77 require_approval_for_new_tools: bool = True 

78 timeout_seconds: int = 120 

79 max_pending_requests: int = 10 

80 cache_approval_seconds: int = 300 

81 

82 

83ApprovalCallback = Callable[[ApprovalRequest], ApprovalDecision] 

84 

85 

86class HumanInTheLoop: 

87 """Manages the human approval workflow for tool calls and mutations. 

88 

89 Supports synchronous callbacks (CLI prompt, webhook, etc.) and 

90 configurable auto-approval rules based on risk and domain. 

91 """ 

92 

93 def __init__( 

94 self, 

95 policy: ApprovalPolicy | None = None, 

96 callback: ApprovalCallback | None = None, 

97 ): 

98 self.policy = policy or ApprovalPolicy() 

99 self.callback = callback 

100 self._pending: dict[str, ApprovalRequest] = {} 

101 self._decisions: dict[str, ApprovalDecision] = {} 

102 self._history: list[tuple[ApprovalRequest, ApprovalDecision]] = [] 

103 self._approval_cache: dict[str, tuple[float, ApprovalDecision]] = {} 

104 

105 def request_approval( 

106 self, 

107 action: str, 

108 description: str = "", 

109 risk_level: RiskLevel = RiskLevel.MEDIUM, 

110 tool_name: str = "", 

111 tool_args: dict[str, Any] | None = None, 

112 estimated_cost_usd: float = 0.0, 

113 data_affected: list[str] | None = None, 

114 ) -> ApprovalRequest: 

115 """Create an approval request and submit it for decision.""" 

116 import time 

117 import uuid 

118 

119 request_id = uuid.uuid4().hex[:12] 

120 req = ApprovalRequest( 

121 request_id=request_id, 

122 action=action, 

123 description=description, 

124 risk_level=risk_level, 

125 tool_name=tool_name, 

126 tool_args=tool_args or {}, 

127 estimated_cost_usd=estimated_cost_usd, 

128 data_affected=data_affected or [], 

129 ) 

130 

131 # Check cache 

132 cache_key = f"{tool_name}:{action}" 

133 if cache_key in self._approval_cache: 

134 ts, decision = self._approval_cache[cache_key] 

135 if time.time() - ts < self.policy.cache_approval_seconds: 

136 self._decisions[request_id] = decision 

137 self._history.append((req, decision)) 

138 return req 

139 

140 # Evaluate auto-approval policy 

141 decision = self._evaluate_policy(req) 

142 if decision is not None: 

143 self._decisions[request_id] = decision 

144 self._history.append((req, decision)) 

145 return req 

146 

147 # Needs human input 

148 if len(self._pending) >= self.policy.max_pending_requests: 

149 decision = ApprovalDecision( 

150 request_id=request_id, 

151 status=ApprovalStatus.REJECTED, 

152 reason="Max pending requests exceeded.", 

153 ) 

154 self._decisions[request_id] = decision 

155 self._history.append((req, decision)) 

156 return req 

157 

158 self._pending[request_id] = req 

159 return req 

160 

161 def decide(self, request_id: str, decision: ApprovalDecision) -> None: 

162 """Record a human decision and remove from pending.""" 

163 self._decisions[request_id] = decision 

164 if request_id in self._pending: 

165 req = self._pending.pop(request_id) 

166 self._history.append((req, decision)) 

167 # Cache if approved 

168 if decision.is_approved: 

169 import time 

170 

171 cache_key = f"{req.tool_name}:{req.action}" 

172 self._approval_cache[cache_key] = (time.time(), decision) 

173 

174 def get_decision(self, request_id: str) -> ApprovalDecision | None: 

175 return self._decisions.get(request_id) 

176 

177 def get_pending(self) -> list[ApprovalRequest]: 

178 return list(self._pending.values()) 

179 

180 def get_history(self) -> list[tuple[ApprovalRequest, ApprovalDecision]]: 

181 return self._history.copy() 

182 

183 def clear_cache(self) -> None: 

184 self._approval_cache.clear() 

185 

186 def _evaluate_policy(self, req: ApprovalRequest) -> ApprovalDecision | None: 

187 """Determine if the request can be auto-decided without human input.""" 

188 

189 # Blocked domains always rejected 

190 domain = req.tool_name.split(".")[0] if req.tool_name else "" 

191 if domain and domain in self.policy.block_domains: 

192 return ApprovalDecision( 

193 request_id=req.request_id, 

194 status=ApprovalStatus.REJECTED, 

195 reason=f"Domain '{domain}' is blocked by policy.", 

196 ) 

197 

198 # Auto-approve domains + low risk 

199 if domain and domain in self.policy.auto_approve_domains: 

200 if req.estimated_cost_usd <= self.policy.max_auto_approve_cost_usd: 

201 return ApprovalDecision( 

202 request_id=req.request_id, 

203 status=ApprovalStatus.APPROVED, 

204 reason=f"Auto-approved: domain '{domain}' is trusted.", 

205 ) 

206 

207 # Risk level check 

208 if req.risk_level not in self.policy.require_approval_for_risk: 

209 return ApprovalDecision( 

210 request_id=req.request_id, 

211 status=ApprovalStatus.SKIPPED, 

212 reason=f"Risk level '{req.risk_level.value}' does not require approval.", 

213 ) 

214 

215 return None # Needs human input 

216 

217 def request_and_decide( 

218 self, 

219 action: str, 

220 description: str = "", 

221 risk_level: RiskLevel = RiskLevel.MEDIUM, 

222 tool_name: str = "", 

223 tool_args: dict[str, Any] | None = None, 

224 ) -> tuple[ApprovalRequest, ApprovalDecision]: 

225 """Create request, attempt auto-decision, invoke callback if needed.""" 

226 req = self.request_approval( 

227 action=action, 

228 description=description, 

229 risk_level=risk_level, 

230 tool_name=tool_name, 

231 tool_args=tool_args, 

232 ) 

233 decision = self.get_decision(req.request_id) 

234 if decision is not None: 

235 return req, decision 

236 

237 if self.callback: 

238 decision = self.callback(req) 

239 self.decide(req.request_id, decision) 

240 else: 

241 decision = ApprovalDecision( 

242 request_id=req.request_id, 

243 status=ApprovalStatus.TIMED_OUT, 

244 reason="No human callback configured.", 

245 ) 

246 

247 return req, decision