Coverage for little_loops / fsm / policy_rules.py: 0%

1"""Shared rule grammar for policy-based decision-table routing. 

2 

3This module implements parse / serialize / evaluate for the policy-router 

4fragment library (lib/policy-router.yaml). The grammar is intentionally 

5minimal and conjunctive-only in v1: rows of AND-joined predicates, first 

6match wins, optional catch-all wildcard. 

7 

8This module is the single source of truth for the grammar so that 

9lib/policy-router.yaml's ``policy_table_dispatch`` fragment and 

10ENH-2233's ``edit-routes`` lens both import the same parse / serialize logic. 

11 

12Public API: 

13 parse_rules(text) -> list[Rule] 

14 serialize_rules(rules) -> str 

15 evaluate_rules(rules, scores) -> str | None 

16""" 

17 

18from __future__ import annotations 

19 

20import re 

21from dataclasses import dataclass, field 

22 

23# --------------------------------------------------------------------------- 

24# Constants 

25# --------------------------------------------------------------------------- 

26 

27_ORDERED_OPS: frozenset[str] = frozenset({">=", "<=", "<", ">"}) 

28_ALL_OPS: frozenset[str] = frozenset({">=", "<=", "==", "!=", "<", ">"}) 

29 

30# Predicate format: <dim>:<op><value> (e.g. "confidence:>=85", "flag:==true") 

31# The dim may contain word chars, spaces, and hyphens. 

32_PRED_PATTERN = re.compile( 

33 r"^(?P<dim>[\w][\w\s\-]*?)\s*:\s*(?P<op>>=|<=|==|!=|<|>)\s*(?P<value>\S.*?)$" 

34) 

35 

36# Valid state-name characters for the -> target 

37_TARGET_PATTERN = re.compile(r"^[\w][\w\-]*$") 

38 

39 

40# --------------------------------------------------------------------------- 

41# Data types 

42# --------------------------------------------------------------------------- 

43 

44 

45@dataclass 

46class Predicate: 

47 """Single comparison in a rule LHS.""" 

48 

49 dim: str 

50 op: str # one of: >=, <=, ==, !=, <, > 

51 value: str # raw string from rule text; numeric for ordered ops 

52 

53 

54@dataclass 

55class Rule: 

56 """One row of the decision table. 

57 

58 A catch-all rule (``is_catchall=True``) has an empty ``predicates`` list 

59 and matches unconditionally when reached. 

60 """ 

61 

62 predicates: list[Predicate] = field(default_factory=list) 

63 target: str = "" 

64 is_catchall: bool = False 

65 

66 

67# --------------------------------------------------------------------------- 

68# Parsing 

69# --------------------------------------------------------------------------- 

70 

71 

72def _parse_predicate(text: str) -> Predicate: 

73 """Parse ``<dim>:<op><value>`` into a :class:`Predicate`. 

74 

75 Raises :class:`ValueError` if the text does not match the expected format 

76 or if an ordered operator is given a non-numeric value. 

77 """ 

78 m = _PRED_PATTERN.match(text.strip()) 

79 if not m: 

80 raise ValueError( 

81 f"Invalid predicate {text!r} — expected '<dim>:<op><value>' " 

82 f"where op is one of {sorted(_ALL_OPS)}" 

83 ) 

84 dim = m.group("dim").strip() 

85 op = m.group("op") 

86 value = m.group("value").strip() 

87 if op in _ORDERED_OPS: 

88 try: 

89 float(value) 

90 except ValueError as err: 

91 raise ValueError( 

92 f"Ordered operator {op!r} requires a numeric value; " 

93 f"got {value!r} in predicate {text!r}" 

94 ) from err 

95 return Predicate(dim=dim, op=op, value=value) 

96 

97 

98def parse_rules(text: str) -> list[Rule]: 

99 """Parse a newline-separated policy rule table into a list of :class:`Rule`. 

100 

101 Syntax:: 

102 

103 # comment lines (skipped) 

104 <blank lines> (skipped) 

105 <dim>:<op><value> -> <state> 

106 <dim>:<op><value> & <dim2>:<op2><value2> -> <state> 

107 * -> <state> 

108 

109 Operators: ``>=``, ``<=``, ``==``, ``!=``, ``<``, ``>`` 

110 

111 Ordered operators (``>=``, ``<=``, ``<``, ``>``) require numeric values; 

112 a non-numeric value for an ordered operator raises :class:`ValueError` at 

113 parse time. The ``==`` and ``!=`` operators accept any string value. 

114 

115 The wildcard catch-all ``* -> <state>`` matches unconditionally and should 

116 appear as the last rule. 

117 

118 Args: 

119 text: Raw rule table text. 

120 

121 Returns: 

122 Ordered list of :class:`Rule` objects (source order preserved). 

123 

124 Raises: 

125 ValueError: On malformed rules (missing ``->``, empty target, invalid 

126 predicate syntax, or non-numeric value with an ordered operator). 

127 """ 

128 rules: list[Rule] = [] 

129 for lineno, raw in enumerate(text.splitlines(), 1): 

130 line = raw.strip() 

131 if not line or line.startswith("#"): 

132 continue 

133 if "->" not in line: 

134 raise ValueError(f"Line {lineno}: rule is missing '->': {line!r}") 

135 lhs, _, rhs = line.partition("->") 

136 lhs = lhs.strip() 

137 target = rhs.strip() 

138 if not target: 

139 raise ValueError(f"Line {lineno}: empty target state in: {line!r}") 

140 if not _TARGET_PATTERN.match(target): 

141 raise ValueError( 

142 f"Line {lineno}: invalid target state name {target!r} " 

143 "(only word chars and hyphens allowed)" 

144 ) 

145 if lhs == "*": 

146 rules.append(Rule(predicates=[], target=target, is_catchall=True)) 

147 else: 

148 parts = [p.strip() for p in lhs.split("&") if p.strip()] 

149 if not parts: 

150 raise ValueError(f"Line {lineno}: empty LHS in: {line!r}") 

151 preds = [_parse_predicate(p) for p in parts] 

152 rules.append(Rule(predicates=preds, target=target, is_catchall=False)) 

153 return rules 

154 

155 

156# --------------------------------------------------------------------------- 

157# Serialization 

158# --------------------------------------------------------------------------- 

159 

160 

161def serialize_rules(rules: list[Rule]) -> str: 

162 """Serialize a rule list back to text (lossless inverse of :func:`parse_rules`). 

163 

164 The round-trip ``parse_rules(serialize_rules(rules))`` produces a :class:`Rule` 

165 list equal in structure to the original (stable). 

166 

167 Args: 

168 rules: List of :class:`Rule` objects. 

169 

170 Returns: 

171 Newline-separated rule table text. 

172 """ 

173 lines: list[str] = [] 

174 for rule in rules: 

175 if rule.is_catchall: 

176 lines.append(f"* -> {rule.target}") 

177 else: 

178 pred_strs = [f"{p.dim}:{p.op}{p.value}" for p in rule.predicates] 

179 lines.append(f"{' & '.join(pred_strs)} -> {rule.target}") 

180 return "\n".join(lines) 

181 

182 

183# --------------------------------------------------------------------------- 

184# Evaluation 

185# --------------------------------------------------------------------------- 

186 

187 

188def _eval_predicate(pred: Predicate, scores: dict[str, object]) -> bool: 

189 """Evaluate a single predicate against the scores dict.""" 

190 raw = scores.get(pred.dim) 

191 

192 if raw is None: 

193 # Missing dimension: treat != as matching (absent != any value), 

194 # all other ops as non-matching. 

195 return pred.op == "!=" 

196 

197 op = pred.op 

198 

199 if op in _ORDERED_OPS: 

200 # Both sides coerced to float (guaranteed numeric by parse_rules). 

201 try: 

202 lhs = float(str(raw)) 

203 rhs = float(pred.value) 

204 except ValueError: 

205 return False 

206 if op == ">": 

207 return lhs > rhs 

208 if op == "<": 

209 return lhs < rhs 

210 if op == ">=": 

211 return lhs >= rhs 

212 # op == "<=" 

213 return lhs <= rhs 

214 

215 # == or != : try numeric coercion first, fall back to string comparison. 

216 try: 

217 lhs_f = float(str(raw)) 

218 rhs_f = float(pred.value) 

219 if op == "==": 

220 return lhs_f == rhs_f 

221 # op == "!=" 

222 return lhs_f != rhs_f 

223 except ValueError: 

224 lhs_s = str(raw) 

225 rhs_s = pred.value 

226 if op == "==": 

227 return lhs_s == rhs_s 

228 # op == "!=" 

229 return lhs_s != rhs_s 

230 

231 

232def evaluate_rules(rules: list[Rule], scores: dict[str, object]) -> str | None: 

233 """Return the target state of the first matching rule, or ``None`` if none match. 

234 

235 Evaluation order follows the list order (priority: first match wins). 

236 A catch-all rule always matches and should be the last rule in the list. 

237 

238 Numeric coercion: ordered operators (``>=``, ``<=``, ``<``, ``>``) compare 

239 both sides as ``float``. For example, ``"9" < "10"`` evaluates to ``True`` 

240 (numeric), not ``False`` (lexical). 

241 

242 Args: 

243 rules: Ordered list of :class:`Rule` objects from :func:`parse_rules`. 

244 scores: Dict mapping dimension names to score values (numeric or string). 

245 

246 Returns: 

247 The target state name of the first matching rule, or ``None``. 

248 """ 

249 for rule in rules: 

250 if rule.is_catchall: 

251 return rule.target 

252 if all(_eval_predicate(p, scores) for p in rule.predicates): 

253 return rule.target 

254 return None