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 

17 

18 

19class FeatureFlag: 

20 """Runtime feature toggle engine. 

21 

22 Usage: 

23 ff = FeatureFlag() 

24 

25 # Define flags 

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

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

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

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

30 

31 # Evaluate 

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

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

34 """ 

35 

36 def __init__(self): 

37 self._flags: dict[str, _FlagDef] = {} 

38 self._lock = threading.RLock() 

39 

40 # ---------- Define ---------- 

41 

42 def define( 

43 self, 

44 name: str, 

45 default: bool = False, 

46 rollout: int = 0, 

47 targets: list[str] | None = None, 

48 depends_on: list[str] | None = None, 

49 ): 

50 """Register a feature flag. 

51 

52 Args: 

53 name: Flag name 

54 default: Default value when no rules match 

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

56 targets: User groups that get this flag 

57 depends_on: Other flags that must be enabled first 

58 """ 

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

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

61 

62 with self._lock: 

63 self._flags[name] = _FlagDef( 

64 name=name, 

65 default=default, 

66 rollout=rollout, 

67 targets=set(targets or []), 

68 depends_on=set(depends_on or []), 

69 overrides={}, 

70 ) 

71 

72 # ---------- Evaluate ---------- 

73 

74 def is_enabled(self, name: str, context: dict | None = None) -> bool: 

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

76 

77 Context may include: 

78 user_id: str 

79 groups: List[str] 

80 """ 

81 context = context or {} 

82 

83 with self._lock: 

84 if name not in self._flags: 

85 return False 

86 

87 flag = self._flags[name] 

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

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

90 

91 # Check overrides 

92 override_key = user_id 

93 if override_key and override_key in flag.overrides: 

94 return flag.overrides[override_key] 

95 

96 # Check group targets 

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

98 return True 

99 

100 # Check percentage rollout 

101 if flag.rollout > 0 and user_id: 

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

103 return True 

104 

105 # Check dependencies 

106 if flag.depends_on: 

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

108 return False 

109 

110 return flag.default 

111 

112 # ---------- Override ---------- 

113 

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

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

116 with self._lock: 

117 if name not in self._flags: 

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

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

120 

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

122 """Remove override for a user.""" 

123 with self._lock: 

124 if name in self._flags: 

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

126 

127 def clear_all_overrides(self, name: str | None = None): 

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

129 with self._lock: 

130 if name: 

131 if name in self._flags: 

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

133 else: 

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

135 flag.overrides.clear() 

136 

137 # ---------- Query ---------- 

138 

139 def list_flags(self) -> list[str]: 

140 with self._lock: 

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

142 

143 def get_definition(self, name: str) -> dict | None: 

144 with self._lock: 

145 flag = self._flags.get(name) 

146 if not flag: 

147 return None 

148 return { 

149 "name": flag.name, 

150 "default": flag.default, 

151 "rollout": flag.rollout, 

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

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

154 } 

155 

156 def remove(self, name: str): 

157 with self._lock: 

158 self._flags.pop(name, None) 

159 

160 # ---------- Internal ---------- 

161 

162 @staticmethod 

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

164 """Deterministic percentage-based rollout. 

165 

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

167 """ 

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

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

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

171 return bucket < percentage 

172 

173 

174class _FlagDef: 

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

176 

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

178 self.name = name 

179 self.default = default 

180 self.rollout = rollout 

181 self.targets: set[str] = targets 

182 self.depends_on: set[str] = depends_on 

183 self.overrides: dict[str, bool] = overrides or {}