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 {"id": "blocker-title-regex", "kind": "title-regex", 

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

45) 

46 

47 

48class GuardError(ValueError): 

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

50 

51 

52@dataclass(frozen=True) 

53class Rule: 

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

55 

56 id: str 

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

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

59 pattern: str | None = None 

60 _frozenset_labels: frozenset[str] = field( 

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

62 ) 

63 

64 def __post_init__(self): 

65 if self.kind == "label": 

66 object.__setattr__( 

67 self, 

68 "_frozenset_labels", 

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

70 ) 

71 

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

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

74 if self.kind == "label": 

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

76 return not self._frozenset_labels.isdisjoint(present) 

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

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

79 

80 

81@dataclass(frozen=True) 

82class GuardResult: 

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

84 

85 title: str 

86 labels: tuple[str, ...] 

87 matched: tuple[str, ...] 

88 rule_ids: tuple[str, ...] 

89 

90 @property 

91 def is_blocker(self) -> bool: 

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

93 return bool(self.matched) 

94 

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

96 return { 

97 "schema_version": GUARD_SCHEMA_VERSION, 

98 "title": self.title, 

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

100 "is_blocker": self.is_blocker, 

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

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

103 } 

104 

105 

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

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

108 

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

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

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

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

113 or narrow the bypass surface). 

114 """ 

115 raw_rules: Any = None 

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

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

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

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

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

121 

122 

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

124 rules: list[Rule] = [] 

125 seen: set[str] = set() 

126 for index, raw in enumerate(raw_rules): 

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

128 if not isinstance(raw, dict): 

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

130 rule_id = raw.get("id") 

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

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

133 rule_id = rule_id.strip() 

134 if rule_id in seen: 

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

136 seen.add(rule_id) 

137 kind = raw.get("kind") 

138 if kind == "label": 

139 labels = raw.get("labels") 

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

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

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

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

144 elif kind == "title-regex": 

145 pattern = raw.get("pattern") 

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

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

148 try: 

149 re.compile(pattern) 

150 except re.error as exc: 

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

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

153 else: 

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

155 return tuple(rules) 

156 

157 

158def evaluate(title: str, labels: tuple[str, ...] | list[str], *, 

159 rules: tuple[Rule, ...]) -> GuardResult: 

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

161 

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

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

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

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

166 """ 

167 norm_labels = tuple(labels) 

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

169 return GuardResult( 

170 title=title, 

171 labels=norm_labels, 

172 matched=matched, 

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

174 ) 

175 

176 

177def evaluate_config(title: str, labels: tuple[str, ...] | list[str], *, 

178 config: cfg.ProjectConfig | None) -> GuardResult: 

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

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