Coverage for agentos/system/approval.py: 43%

1""" 

2可视化授权引擎 — Agent 主动申请权限，用户可视化审批。 

3 

4与 permissions.py 的分工: 

5- permissions.py: 纯程序化权限检查（require/check） 

6- approval.py: HITL 授权流程（Agent 发起申请 → 用户审批 → 回调） 

7 

8流程: 

91. Agent 调用 request_approval(tier, resource, reason) 

102. 引擎生成 ApprovalTicket，挂起等待 

113. 通过 WebSocket 推送给桌面客户端 

124. 用户在 UI 点击"同意" / "拒绝" / "拒绝+记住" 

135. 回调触发，Agent 继续或拒绝 

14""" 

15 

16from __future__ import annotations 

17 

18import asyncio 

19import uuid 

20from collections.abc import Awaitable, Callable 

21from dataclasses import dataclass, field 

22from datetime import datetime 

23from enum import StrEnum 

24 

25from agentos.system.permissions import PermissionTier 

26 

27 

28class ApprovalStatus(StrEnum): 

29 PENDING = "pending" 

30 APPROVED = "approved" 

31 DENIED = "denied" 

32 DENIED_REMEMBER = "denied_remember" # 拒绝并记住，本次会话不再询问 

33 TIMEOUT = "timeout" 

34 

35 

36@dataclass 

37class ApprovalTicket: 

38 """授权申请票据。""" 

39 

40 ticket_id: str 

41 tier: PermissionTier 

42 resource: str # 申请访问的资源路径/命令 

43 reason: str # Agent 申请原因（AI 生成的中文说明） 

44 status: ApprovalStatus = ApprovalStatus.PENDING 

45 created_at: str = field(default_factory=lambda: datetime.now().isoformat()) 

46 resolved_at: str = "" 

47 session_id: str = "" 

48 # 回调 

49 _future: asyncio.Future | None = field(default=None, repr=False) 

50 

51 def __post_init__(self): 

52 if self._future is None: 

53 self._future = asyncio.Future() 

54 

55 async def wait(self, timeout: float = 60.0) -> ApprovalStatus: 

56 """等待审批结果。""" 

57 try: 

58 return await asyncio.wait_for(asyncio.shield(self._future), timeout=timeout) 

59 except TimeoutError: 

60 self.status = ApprovalStatus.TIMEOUT 

61 self.resolved_at = datetime.now().isoformat() 

62 return ApprovalStatus.TIMEOUT 

63 

64 def approve(self) -> None: 

65 """批准。""" 

66 self.status = ApprovalStatus.APPROVED 

67 self.resolved_at = datetime.now().isoformat() 

68 if self._future and not self._future.done(): 

69 self._future.set_result(ApprovalStatus.APPROVED) 

70 

71 def deny(self, remember: bool = False) -> None: 

72 """拒绝。""" 

73 self.status = ApprovalStatus.DENIED_REMEMBER if remember else ApprovalStatus.DENIED 

74 self.resolved_at = datetime.now().isoformat() 

75 if self._future and not self._future.done(): 

76 self._future.set_result(self.status) 

77 

78 

79class ApprovalEngine: 

80 """可视化授权引擎 — 管理审批流程。 

81 

82 用法: 

83 engine = ApprovalEngine(pm, session_id) 

84 engine.set_push_callback(send_to_ws) # 设置推送回调 

85 

86 # Agent 侧: 

87 approved = await engine.request( 

88 PermissionTier.SHELL_STANDARD, 

89 "rm -rf ./build/", 

90 "需要清理构建缓存以释放磁盘空间", 

91 ) 

92 if approved: 

93 # 执行操作 

94 """ 

95 

96 def __init__(self, perm_manager, session_id: str): 

97 self._pm = perm_manager 

98 self._sid = session_id 

99 self._push_callback: Callable[[dict], Awaitable[None]] | None = None 

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

101 self._denied_remember: set[str] = set() # 本次会话已记住拒绝的 (tier, pattern) 

102 

103 def set_push_callback(self, callback: Callable[[dict], Awaitable[None]]) -> None: 

104 """设置推送回调（发送到 WebSocket 客户端）。""" 

105 self._push_callback = callback 

106 

107 async def request( 

108 self, tier: PermissionTier, resource: str, reason: str, timeout: float = 60.0 

109 ) -> bool: 

110 """Agent 发起权限申请，返回是否获批。 

111 

112 如果已有 DENIED_REMEMBER 记录，直接返回 False。 

113 """ 

114 # 检查是否已记住拒绝 

115 deny_key = f"{tier.value}:{resource}" 

116 if deny_key in self._denied_remember: 

117 return False 

118 

119 ticket = ApprovalTicket( 

120 ticket_id=f"approval-{uuid.uuid4().hex[:12]}", 

121 tier=tier, 

122 resource=resource, 

123 reason=reason, 

124 session_id=self._sid, 

125 ) 

126 self._pending[ticket.ticket_id] = ticket 

127 

128 # 推送到客户端 

129 if self._push_callback: 

130 try: 

131 await self._push_callback( 

132 { 

133 "type": "approval_request", 

134 "data": { 

135 "ticket_id": ticket.ticket_id, 

136 "tier": ticket.tier.value, 

137 "tier_label": ticket.tier.label, 

138 "resource": ticket.resource, 

139 "reason": ticket.reason, 

140 "session_id": self._sid, 

141 "timeout": timeout, 

142 }, 

143 } 

144 ) 

145 except Exception: 

146 pass # 推送失败不阻塞 

147 

148 # 等待审批 

149 result = await ticket.wait(timeout=timeout) 

150 

151 # 处理记住拒绝 

152 if result == ApprovalStatus.DENIED_REMEMBER: 

153 self._denied_remember.add(deny_key) 

154 

155 # 如果审批通过，临时提升权限 

156 if result == ApprovalStatus.APPROVED: 

157 self._pm.escalate(self._sid, tier, resource) 

158 

159 # 清理 

160 self._pending.pop(ticket.ticket_id, None) 

161 

162 return result == ApprovalStatus.APPROVED 

163 

164 def resolve(self, ticket_id: str, approved: bool, remember: bool = False) -> bool: 

165 """手动审批票据（从 UI 调用）。""" 

166 ticket = self._pending.get(ticket_id) 

167 if not ticket or ticket.status != ApprovalStatus.PENDING: 

168 return False 

169 

170 if approved: 

171 ticket.approve() 

172 else: 

173 ticket.deny(remember=remember) 

174 return True 

175 

176 def get_pending_tickets(self) -> list[dict]: 

177 """获取所有待审批票据（供 UI 轮询）。""" 

178 return [ 

179 { 

180 "ticket_id": t.ticket_id, 

181 "tier": t.tier.value, 

182 "tier_label": t.tier.label, 

183 "resource": t.resource, 

184 "reason": t.reason, 

185 "created_at": t.created_at, 

186 "timeout_remaining": max( 

187 0, 60 - (datetime.now() - datetime.fromisoformat(t.created_at)).total_seconds() 

188 ), 

189 } 

190 for t in self._pending.values() 

191 if t.status == ApprovalStatus.PENDING 

192 ] 

193 

194 def clear_denied(self) -> None: 

195 """清除本次会话的记住拒绝记录。""" 

196 self._denied_remember.clear()