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

1"""Plan + run quality gates — built-in gates and project Lego gates, uniformly. 

2 

3A *gate* is anything that can pass/fail and produce findings: the built-in 

4``build`` / ``lint`` / ``jury`` gates (from ``project.yaml``'s ``gates:`` list), 

5plus the project's blocking-capable extension hooks. :func:`plan_gates` 

6turns a config + loaded extensions into an ordered list of :class:`GateSpec`; 

7:func:`run_gates` executes them through an injected ``runner`` with fail-soft 

8semantics, normalising everything into :class:`keel.findings.Finding`. 

9""" 

10 

11from __future__ import annotations 

12 

13from collections.abc import Callable 

14from dataclasses import dataclass 

15from typing import TYPE_CHECKING 

16 

17from .findings import Finding 

18 

19if TYPE_CHECKING: # pragma: no cover 

20 from .config import ProjectConfig 

21 from .extensions import Extension 

22 

23#: Built-in gate names accepted in ``project.yaml``'s ``gates:`` list. 

24BUILTIN_GATES: tuple[str, ...] = ("build", "lint", "jury") 

25 

26# A failed gate with no explicit findings is reported at this severity. 

27_ON_FAIL_SEVERITY: dict[str, str] = {"block": "major", "suggest": "minor", "warn": "nit"} 

28 

29 

30class GateError(ValueError): 

31 """Raised when a config references an unknown built-in gate.""" 

32 

33 

34@dataclass(frozen=True) 

35class GateSpec: 

36 """A planned gate. ``phase`` is the backbone step it runs at.""" 

37 

38 id: str 

39 kind: str # command | agentic | builtin 

40 phase: str # backbone step name, e.g. "guard", "test", or "pre-merge" 

41 on_fail: str # block | suggest | warn 

42 run: str | None = None 

43 prompt: str | None = None 

44 agent: str = "inherit" 

45 source: str = "builtin" 

46 required_capabilities: tuple[str, ...] = () 

47 optional_capabilities: tuple[str, ...] = () 

48 

49 

50@dataclass(frozen=True) 

51class GateOutcome: 

52 """Result of running one gate.""" 

53 

54 gate: str 

55 ok: bool 

56 findings: tuple[Finding, ...] = () 

57 error: str | None = None 

58 skipped: bool = False 

59 

60 

61# runner(spec) -> (ok, findings). May raise; run_gates handles it fail-soft. 

62GateRunner = Callable[[GateSpec], tuple[bool, list[Finding]]] 

63 

64 

65def plan_gates(config: ProjectConfig, loaded: dict[str, list[Extension]]) -> tuple[GateSpec, ...]: 

66 """Order gates by backbone phase: guard, built-in test gates, test hooks, pre-merge.""" 

67 specs: list[GateSpec] = [] 

68 

69 for e in loaded.get("guard", []): 

70 specs.append(GateSpec(e.id, e.kind, "guard", e.on_fail, 

71 run=e.run, prompt=e.prompt, agent=e.agent, source=e.source, 

72 required_capabilities=e.required_capabilities, 

73 optional_capabilities=e.optional_capabilities)) 

74 

75 for name in config.gates: 

76 if name == "build": 

77 specs.append(GateSpec("build", "command", "test", "block", 

78 run=config.knobs.build_gate_cmd)) 

79 elif name == "lint": 

80 if config.knobs.lint_cmd: # lint is optional 

81 specs.append(GateSpec("lint", "command", "test", "block", 

82 run=config.knobs.lint_cmd)) 

83 elif name == "jury": 

84 specs.append(GateSpec("jury", "builtin", "test", "block")) 

85 else: 

86 raise GateError( 

87 f"unknown built-in gate {name!r}; valid: {', '.join(BUILTIN_GATES)} " 

88 "(project gates belong in extension slots, not in gates:)" 

89 ) 

90 

91 for slot, phase in (("tester", "test"), ("test", "test"), ("pre-merge", "pre-merge")): 

92 for e in loaded.get(slot, []): 

93 specs.append(GateSpec(e.id, e.kind, phase, e.on_fail, 

94 run=e.run, prompt=e.prompt, agent=e.agent, source=e.source, 

95 required_capabilities=e.required_capabilities, 

96 optional_capabilities=e.optional_capabilities)) 

97 return tuple(specs) 

98 

99 

100def run_gates(specs, runner: GateRunner, *, fail_soft: bool = True) -> list[GateOutcome]: 

101 """Run each gate via ``runner``; normalise to outcomes (fail-soft by default).""" 

102 outcomes: list[GateOutcome] = [] 

103 for spec in specs: 

104 try: 

105 ok, found = runner(spec) 

106 except Exception as exc: # noqa: BLE001 - fail-soft is the contract 

107 if not fail_soft: 

108 raise 

109 if spec.on_fail == "block": 

110 # A hard gate that errors must still block (can't silently pass). 

111 finding = Finding("major", f"gate {spec.id!r} errored: {exc}", spec.id) 

112 outcomes.append(GateOutcome(spec.id, False, (finding,), error=str(exc))) 

113 else: 

114 # Soft gate broke -> degrade to a no-op (logged), never abort. 

115 outcomes.append(GateOutcome(spec.id, True, (), error=str(exc), skipped=True)) 

116 continue 

117 

118 found = tuple(found) 

119 if ok: 

120 outcomes.append(GateOutcome(spec.id, True, found)) 

121 else: 

122 if not found: 

123 sev = _ON_FAIL_SEVERITY[spec.on_fail] 

124 found = (Finding(sev, f"gate {spec.id!r} failed", spec.id),) 

125 outcomes.append(GateOutcome(spec.id, False, found)) 

126 return outcomes 

127 

128 

129def collect_findings(outcomes: list[GateOutcome]) -> list[Finding]: 

130 """Flatten all findings across gate outcomes (in outcome order).""" 

131 out: list[Finding] = [] 

132 for o in outcomes: 

133 out.extend(o.findings) 

134 return out