Coverage for src/keel/guard.py: 100%

1"""Deterministic blocker ruleset — the pure core behind ``keel guard``. 

2 

3Blocker promotion is what unlocks the night-window bypass at s10 (``keel 

4merge --hotfix``). Before this module, that promotion was pure agent judgment: 

5an agent could declare any issue a blocker and merge at 3am. This module makes 

6the decision a **deterministic, configurable function** of the issue's facts — 

7its title and labels — so a claimed blocker can be verified against the rule it 

8allegedly matched. 

9 

10The matching is pure (no network/subprocess/clock/random and no I/O): given an 

11issue title, its labels, and a resolved set of :class:`Rule` objects, it returns 

12the ids of the rules that fired. The CLI gathers the live issue facts and reads 

13the configured rules; this module only decides. 

14 

15Rules are resolved from ``policy_pack.blocker_rules`` when present, falling back 

16to built-in defaults (back-compatible: an absent config yields the defaults). 

17Each rule is one of two kinds: 

18 

19* ``label`` — fires when one of the rule's labels is present on the issue 

20 (case-insensitive exact match). 

21* ``title-regex`` — fires when the rule's regex matches the issue title. 

22 

23The built-in defaults cover the heuristics named in the audit (GAP-11): 

24word-boundary ``\bhotfix\b`` / ``\bsecurity\b`` / ``\bblocker\b`` title regexes 

25and a configurable blocker label. 

26""" 

27 

28from __future__ import annotations 

29 

30import re 

31from dataclasses import dataclass, field 

32from typing import Any 

33 

34from . import config as cfg 

35 

36GUARD_SCHEMA_VERSION = "keel.guard.v1" 

37 

38#: Built-in defaults, used when ``policy_pack.blocker_rules`` is absent/empty. 

39DEFAULT_RULES: tuple[dict[str, Any], ...] = ( 

40 {"id": "blocker-label", "kind": "label", "labels": ["blocker"]}, 

41 {"id": "hotfix-label", "kind": "label", "labels": ["hotfix"]}, 

42 {"id": "security-label", "kind": "label", "labels": ["security"]}, 

43 { 

44 "id": "blocker-title-regex", 

45 "kind": "title-regex", 

46 "pattern": r"\b(?:hotfix|security|blocker)\b", 

47 }, 

48) 

49 

50 

51class GuardError(ValueError): 

52 """Raised when a configured blocker rule is malformed.""" 

53 

54 

55@dataclass(frozen=True) 

56class Rule: 

57 """A single resolved, immutable blocker rule.""" 

58 

59 id: str 

60 kind: str # "label" | "title-regex" 

61 labels: tuple[str, ...] = () 

62 pattern: str | None = None 

63 _frozenset_labels: frozenset[str] = field( 

64 init=False, repr=False, compare=False, hash=False, default=frozenset() 

65 ) 

66 

67 def __post_init__(self): 

68 if self.kind == "label": 

69 object.__setattr__( 

70 self, 

71 "_frozenset_labels", 

72 frozenset(want.strip().casefold() for want in self.labels), 

73 ) 

74 

75 def matches(self, title: str, labels: tuple[str, ...]) -> bool: 

76 """True if this rule fires for the given issue facts (pure).""" 

77 if self.kind == "label": 

78 present = {label.strip().casefold() for label in labels} 

79 return not self._frozenset_labels.isdisjoint(present) 

80 # title-regex — ``pattern`` is guaranteed non-empty by :func:`resolve_rules`. 

81 return re.search(self.pattern or "", title, re.IGNORECASE) is not None 

82 

83 

84@dataclass(frozen=True) 

85class GuardResult: 

86 """The structured outcome of evaluating an issue against the ruleset.""" 

87 

88 title: str 

89 labels: tuple[str, ...] 

90 matched: tuple[str, ...] 

91 rule_ids: tuple[str, ...] 

92 

93 @property 

94 def is_blocker(self) -> bool: 

95 """True when at least one rule fired.""" 

96 return bool(self.matched) 

97 

98 def as_dict(self) -> dict[str, Any]: 

99 return { 

100 "schema_version": GUARD_SCHEMA_VERSION, 

101 "title": self.title, 

102 "labels": list(self.labels), 

103 "is_blocker": self.is_blocker, 

104 "matched": list(self.matched), 

105 "rule_ids": list(self.rule_ids), 

106 } 

107 

108 

109def resolve_rules(config: cfg.ProjectConfig | None) -> tuple[Rule, ...]: 

110 """Resolve the active blocker rules from config, falling back to defaults. 

111 

112 Reads ``policy_pack.blocker_rules`` (a list of rule dicts). When absent or 

113 not a list, the built-in :data:`DEFAULT_RULES` are used — keeping projects 

114 without any blocker config fully back-compatible. Raises :class:`GuardError` 

115 on a malformed configured rule (fail-closed: a typo must not silently widen 

116 or narrow the bypass surface). 

117 """ 

118 raw_rules: Any = None 

119 if config is not None and isinstance(config.policy_pack, dict): 

120 raw_rules = config.policy_pack.get("blocker_rules") 

121 if not isinstance(raw_rules, list) or not raw_rules: 

122 return _build_rules(DEFAULT_RULES, source="defaults") 

123 return _build_rules(raw_rules, source="policy_pack.blocker_rules") 

124 

125 

126def _build_rules(raw_rules: Any, *, source: str) -> tuple[Rule, ...]: 

127 rules: list[Rule] = [] 

128 seen: set[str] = set() 

129 for index, raw in enumerate(raw_rules): 

130 where = f"{source}[{index}]" 

131 if not isinstance(raw, dict): 

132 raise GuardError(f"{where}: expected an object") 

133 rule_id = raw.get("id") 

134 if not isinstance(rule_id, str) or not rule_id.strip(): 

135 raise GuardError(f"{where}: missing non-empty 'id'") 

136 rule_id = rule_id.strip() 

137 if rule_id in seen: 

138 raise GuardError(f"{where}: duplicate rule id {rule_id!r}") 

139 seen.add(rule_id) 

140 kind = raw.get("kind") 

141 if kind == "label": 

142 labels = raw.get("labels") 

143 if not isinstance(labels, list) or not labels: 

144 raise GuardError(f"{where}: label rule needs a non-empty 'labels' list") 

145 clean = tuple(str(label) for label in labels) 

146 rules.append(Rule(id=rule_id, kind="label", labels=clean)) 

147 elif kind == "title-regex": 

148 pattern = raw.get("pattern") 

149 if not isinstance(pattern, str) or not pattern: 

150 raise GuardError(f"{where}: title-regex rule needs a non-empty 'pattern'") 

151 try: 

152 re.compile(pattern) 

153 except re.error as exc: 

154 raise GuardError(f"{where}: invalid regex {pattern!r}: {exc}") from exc 

155 rules.append(Rule(id=rule_id, kind="title-regex", pattern=pattern)) 

156 else: 

157 raise GuardError(f"{where}: unknown rule kind {kind!r}") 

158 return tuple(rules) 

159 

160 

161def evaluate( 

162 title: str, labels: tuple[str, ...] | list[str], *, rules: tuple[Rule, ...] 

163) -> GuardResult: 

164 """Evaluate the issue facts against ``rules`` (pure). 

165 

166 Returns a :class:`GuardResult` carrying the ids of every rule that fired 

167 (in rule order) plus the full set of rule ids considered. Rule ids are 

168 unique by construction (:func:`resolve_rules` rejects duplicates), so each 

169 fired rule appears at most once without an explicit dedup step. 

170 """ 

171 norm_labels = tuple(labels) 

172 matched = tuple(rule.id for rule in rules if rule.matches(title, norm_labels)) 

173 return GuardResult( 

174 title=title, 

175 labels=norm_labels, 

176 matched=matched, 

177 rule_ids=tuple(rule.id for rule in rules), 

178 ) 

179 

180 

181def evaluate_config( 

182 title: str, labels: tuple[str, ...] | list[str], *, config: cfg.ProjectConfig | None 

183) -> GuardResult: 

184 """Convenience: resolve rules from ``config`` then :func:`evaluate`.""" 

185 return evaluate(title, labels, rules=resolve_rules(config))