Coverage for agentos/tools/feature_flag.py: 0%

1""" 

2FeatureFlag — runtime feature toggle system with percentage rollout. 

3 

4Supports: 

5 - Boolean flags 

6 - Percentage-based rollouts 

7 - Target rules (by user ID, group, environment) 

8 - Flag dependencies (flag A requires flag B enabled) 

9 - Overrides (force-on / force-off per context) 

10 - Thread-safe reads/writes 

11""" 

12 

13from __future__ import annotations 

14 

15import hashlib 

16import threading 

17from typing import Any, Callable, Dict, List, Optional, Set 

18 

19 

20class FeatureFlag: 

21 """Runtime feature toggle engine. 

22 

23 Usage: 

24 ff = FeatureFlag() 

25 

26 # Define flags 

27 ff.define("dark_mode", default=False) 

28 ff.define("new_checkout", default=False, rollout=10) # 10% users 

29 ff.define("beta_search", default=False, targets=["beta-users"]) 

30 ff.define("analytics_v2", default=True, depends_on=["new_checkout"]) 

31 

32 # Evaluate 

33 ff.is_enabled("dark_mode", context={"user_id": "user123"}) 

34 ff.is_enabled("new_checkout", context={"user_id": "user123"}) 

35 """ 

36 

37 def __init__(self): 

38 self._flags: Dict[str, _FlagDef] = {} 

39 self._lock = threading.RLock() 

40 

41 # ---------- Define ---------- 

42 

43 def define( 

44 self, 

45 name: str, 

46 default: bool = False, 

47 rollout: int = 0, 

48 targets: Optional[List[str]] = None, 

49 depends_on: Optional[List[str]] = None, 

50 ): 

51 """Register a feature flag. 

52 

53 Args: 

54 name: Flag name 

55 default: Default value when no rules match 

56 rollout: Percentage (0-100) of users who get the flag 

57 targets: User groups that get this flag 

58 depends_on: Other flags that must be enabled first 

59 """ 

60 if not (0 <= rollout <= 100): 

61 raise ValueError("rollout must be 0-100") 

62 

63 with self._lock: 

64 self._flags[name] = _FlagDef( 

65 name=name, 

66 default=default, 

67 rollout=rollout, 

68 targets=set(targets or []), 

69 depends_on=set(depends_on or []), 

70 overrides={}, 

71 ) 

72 

73 # ---------- Evaluate ---------- 

74 

75 def is_enabled(self, name: str, context: Optional[dict] = None) -> bool: 

76 """Check whether a feature flag is enabled for the given context. 

77 

78 Context may include: 

79 user_id: str 

80 groups: List[str] 

81 """ 

82 context = context or {} 

83 

84 with self._lock: 

85 if name not in self._flags: 

86 return False 

87 

88 flag = self._flags[name] 

89 user_id = context.get("user_id", "") 

90 groups = set(context.get("groups", [])) 

91 

92 # Check overrides 

93 override_key = user_id 

94 if override_key and override_key in flag.overrides: 

95 return flag.overrides[override_key] 

96 

97 # Check group targets 

98 if flag.targets and flag.targets & groups: 

99 return True 

100 

101 # Check percentage rollout 

102 if flag.rollout > 0 and user_id: 

103 if self._in_rollout(user_id, name, flag.rollout): 

104 return True 

105 

106 # Check dependencies 

107 if flag.depends_on: 

108 if not all(self.is_enabled(d, context) for d in flag.depends_on): 

109 return False 

110 

111 return flag.default 

112 

113 # ---------- Override ---------- 

114 

115 def set_override(self, name: str, user_id: str, value: bool): 

116 """Force a flag on/off for a specific user.""" 

117 with self._lock: 

118 if name not in self._flags: 

119 raise KeyError(f"Unknown flag: {name}") 

120 self._flags[name].overrides[user_id] = value 

121 

122 def clear_override(self, name: str, user_id: str): 

123 """Remove override for a user.""" 

124 with self._lock: 

125 if name in self._flags: 

126 self._flags[name].overrides.pop(user_id, None) 

127 

128 def clear_all_overrides(self, name: Optional[str] = None): 

129 """Clear all overrides, optionally for a specific flag.""" 

130 with self._lock: 

131 if name: 

132 if name in self._flags: 

133 self._flags[name].overrides.clear() 

134 else: 

135 for flag in self._flags.values(): 

136 flag.overrides.clear() 

137 

138 # ---------- Query ---------- 

139 

140 def list_flags(self) -> List[str]: 

141 with self._lock: 

142 return list(self._flags.keys()) 

143 

144 def get_definition(self, name: str) -> Optional[dict]: 

145 with self._lock: 

146 flag = self._flags.get(name) 

147 if not flag: 

148 return None 

149 return { 

150 "name": flag.name, 

151 "default": flag.default, 

152 "rollout": flag.rollout, 

153 "targets": list(flag.targets), 

154 "depends_on": list(flag.depends_on), 

155 } 

156 

157 def remove(self, name: str): 

158 with self._lock: 

159 self._flags.pop(name, None) 

160 

161 # ---------- Internal ---------- 

162 

163 @staticmethod 

164 def _in_rollout(user_id: str, flag_name: str, percentage: int) -> bool: 

165 """Deterministic percentage-based rollout. 

166 

167 Uses MD5 hash of (user_id + flag_name) to produce stable grouping. 

168 """ 

169 key = f"{user_id}:{flag_name}" 

170 h = hashlib.md5(key.encode()).hexdigest() 

171 bucket = int(h[:8], 16) % 100 

172 return bucket < percentage 

173 

174 

175class _FlagDef: 

176 __slots__ = ("name", "default", "rollout", "targets", "depends_on", "overrides") 

177 

178 def __init__(self, name, default, rollout, targets, depends_on, overrides): 

179 self.name = name 

180 self.default = default 

181 self.rollout = rollout 

182 self.targets: Set[str] = targets 

183 self.depends_on: Set[str] = depends_on 

184 self.overrides: Dict[str, bool] = overrides or {}