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

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

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"""

11from __future__ import annotations

13from collections.abc import Callable

14from dataclasses import dataclass

15from typing import TYPE_CHECKING

17from .findings import Finding

19if TYPE_CHECKING: # pragma: no cover

20 from .config import ProjectConfig

21 from .extensions import Extension

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

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

26#: Declarative security & SAST presets supported in ``policy_pack.presets``.

27POLICY_PACK_PRESETS: dict[str, tuple[str, str, str, str]] = {

28 # preset: (gate_id, phase, on_fail, run_cmd)

29 "gitleaks": ("gitleaks", "guard", "block", "gitleaks detect --no-git -v"),

30 "semgrep": ("semgrep", "test", "suggest", "semgrep scan"),

31 "bandit": ("bandit", "test", "suggest", "bandit -r . -ll"),

32 "trivy": ("trivy", "test", "warn", "trivy fs ."),

33}

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

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

39class GateError(ValueError):

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

43@dataclass(frozen=True)

44class GateSpec:

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

47 id: str

48 kind: str # command | agentic | builtin

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

50 on_fail: str # block | suggest | warn

51 run: str | None = None

52 prompt: str | None = None

53 agent: str = "inherit"

54 source: str = "builtin"

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

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

57 #: Resolved wall-clock limit for a ``command`` gate, in seconds. ``None`` means

58 #: the runner's own fallback applies (a spec built outside :func:`plan_gates`).

59 timeout: int | None = None

62@dataclass(frozen=True)

63class GateOutcome:

64 """Result of running one gate."""

66 gate: str

67 ok: bool

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

69 error: str | None = None

70 skipped: bool = False

71 #: True when this gate was killed by its wall-clock limit rather than returning a

72 #: verdict. Purely descriptive: a timed-out gate is still ``ok=False`` with an

73 #: unchanged severity, so it blocks the merge exactly as a failure does. Only the

74 #: label and the operator-facing explanation differ — a hanging command is a real

75 #: defect and must stay red.

76 timed_out: bool = False

77 #: True when *this runner did not execute the gate at all* — an ``agentic`` gate

78 #: reached the command-only runner, which the agent-dispatch layer runs instead.

79 #: Distinct from ``ok`` on purpose: "not my job" must never be recorded as "ran and

80 #: passed", or a blocking review gate nobody executed would authorize the merge.

81 #: ``ok`` stays True so a soft gate does not spuriously fail the run; consumers that

82 #: certify (see :func:`keel.ledger.record_gates_passed`) must refuse a *blocking*

83 #: gate that was never run.

84 not_run: bool = False

85 #: The gate's declared severity (``block`` / ``suggest`` / ``warn``), carried from

86 #: its :class:`GateSpec` so a consumer reading only outcomes can tell whether a

87 #: ``not_run`` gate was one the project required.

88 on_fail: str = "block"

91# runner(spec) -> (ok, findings[, timed_out[, not_run]]). May raise; run_gates handles

92# it fail-soft. The shorter forms stay supported for runners that cannot time out or

93# that execute every gate they are given.

94GateRunner = Callable[

95 [GateSpec],

96 "tuple[bool, list[Finding]] | tuple[bool, list[Finding], bool] "

97 "| tuple[bool, list[Finding], bool, bool]",

98]

100

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

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

103

104 Every gate that shells out gets its wall-clock ``timeout`` resolved here, so the

105 planner is the single place budgets are decided:

106

107 * ``command`` gates, most specific first — the extension's own ``timeout:``

108 frontmatter → ``knobs.gate_timeout_s`` → :data:`keel.model.DEFAULT_GATE_TIMEOUT_S`;

109 * the ``jury`` builtin, which also shells out (via ``run_argv``) —

110 ``knobs.jury_timeout_s``, kept separate because a cross-vendor panel and a test

111 suite have unrelated runtimes.

112

113 ``agentic`` gates carry ``None``: the agent-dispatch layer runs those, nothing

114 shells out for them, and a number there would advertise a limit never applied.

115 """

116 project_timeout = config.knobs.gate_timeout_s

117

118 def _timeout_for(e: Extension) -> int | None:

119 if e.kind != "command":

120 return None

121 return e.timeout if e.timeout is not None else project_timeout

122

123 specs: list[GateSpec] = []

124 presets = (

125 tuple(config.policy_pack.get("presets", ()))

126 if isinstance(config.policy_pack, dict)

127 else ()

128 )

129

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

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

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

133 required_capabilities=e.required_capabilities,

134 optional_capabilities=e.optional_capabilities,

135 timeout=_timeout_for(e)))

136

137 if "gitleaks" in presets:

138 gid, phase, on_fail, run_cmd = POLICY_PACK_PRESETS["gitleaks"]

139 specs.append(GateSpec(gid, "command", phase, on_fail, run=run_cmd,

140 source="policy_pack:preset:gitleaks", timeout=project_timeout))

141

142 for name in config.gates:

143 if name == "build":

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

145 run=config.knobs.build_gate_cmd, timeout=project_timeout))

146 elif name == "lint":

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

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

149 run=config.knobs.lint_cmd, timeout=project_timeout))

150 elif name == "jury":

151 specs.append(GateSpec("jury", "builtin", "test", "block",

152 timeout=config.knobs.jury_timeout_s))

153 else:

154 raise GateError(

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

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

157 )

158

159 for preset_name in ("semgrep", "bandit", "trivy"):

160 if preset_name in presets:

161 gid, phase, on_fail, run_cmd = POLICY_PACK_PRESETS[preset_name]

162 specs.append(GateSpec(gid, "command", phase, on_fail, run=run_cmd,

163 source=f"policy_pack:preset:{preset_name}",

164 timeout=project_timeout))

165

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

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

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

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

170 required_capabilities=e.required_capabilities,

171 optional_capabilities=e.optional_capabilities,

172 timeout=_timeout_for(e)))

173 return tuple(specs)

174

175

176def run_gates(

177 specs,

178 runner: GateRunner,

179 *,

180 fail_soft: bool = True,

181 concurrency: int = 1,

182) -> list[GateOutcome]:

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

184

185 When ``concurrency > 1``, independent gates are executed concurrently using

186 standard library ``concurrent.futures.ThreadPoolExecutor``, while preserving

187 exact deterministic outcome ordering.

188 """

189 def _run_single(spec: GateSpec) -> GateOutcome:

190 try:

191 # tuple() first: the runner contract has always been "any 2-iterable",

192 # so indexing the raw return would reject a generator that used to work.

193 result = tuple(runner(spec))

194 # Runners that cannot time out may return the 2-tuple form; runners that

195 # execute every gate they are given may omit the not-run flag.

196 ok, found = result[0], result[1]

197 timed_out = result[2] is True if len(result) > 2 else False

198 not_run = result[3] is True if len(result) > 3 else False

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

200 if not fail_soft:

201 raise

202 if spec.on_fail == "block":

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

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

205 return GateOutcome(spec.id, False, (finding,), error=str(exc),

206 on_fail=spec.on_fail)

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

208 return GateOutcome(spec.id, True, (), error=str(exc), skipped=True,

209 on_fail=spec.on_fail)

210

211 found = tuple(found)

212 if ok:

213 return GateOutcome(spec.id, True, found, not_run=not_run,

214 on_fail=spec.on_fail)

215 if not found:

216 sev = _ON_FAIL_SEVERITY[spec.on_fail]

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

218 # ok stays False for a timeout: the merge gate is unchanged, only the label.

219 # not_run rides along on this branch too: dropping it would let a future

220 # runner that reports a not-run gate as *failing* certify the merge anyway.

221 return GateOutcome(spec.id, False, found, timed_out=timed_out,

222 not_run=not_run, on_fail=spec.on_fail)

223

224 spec_list = list(specs)

225 if concurrency <= 1 or len(spec_list) <= 1:

226 return [_run_single(s) for s in spec_list]

227

228 from concurrent.futures import ThreadPoolExecutor

229

230 with ThreadPoolExecutor(max_workers=concurrency) as executor:

231 return list(executor.map(_run_single, spec_list))

232

233

234def unrun_blocking(outcomes: list[GateOutcome]) -> tuple[str, ...]:

235 """Names of ``on_fail: block`` gates this run did not execute, in outcome order."""

236 return tuple(o.gate for o in outcomes if o.not_run and o.on_fail == "block")

237

238

239def apply_recorded_results(

240 outcomes: list[GateOutcome], results: dict[str, str]

241) -> tuple[list[GateOutcome], list[str]]:

242 """Fold externally-executed gate verdicts into ``outcomes``.

243

244 ``results`` maps a gate id to ``"pass"`` or ``"fail"``. It exists because the

245 command-only runner cannot execute ``agentic`` gates — the agent-dispatch layer

246 does — and without a way to report back, such a gate stays ``not_run`` forever and

247 :func:`keel.ledger.record_gates_passed` can never certify the run. That would make

248 a blocking agentic gate a permanent merge block rather than a gate.

249

250 **Only a ``not_run`` outcome is replaced.** A gate keel executed has a measured

251 verdict, and letting a recorded one override it would turn this channel into a way

252 to certify a run whose gates were observed failing — the same fail-open this whole

253 series exists to close, arriving from the other direction. Results naming an

254 executed gate are returned in ``rejected`` so the caller can refuse loudly rather

255 than silently discard them.

256

257 A recorded result clears ``not_run``, because the gate *was* run; a ``fail``

258 additionally produces a finding at the gate's declared severity, exactly as an

259 in-process failure would. A not-run gate can be neither timed out nor skipped, so

260 the rebuilt outcome carries neither.

261

262 Returns ``(outcomes, rejected)``. Ids matching no outcome at all are left to the

263 CLI, which validates them against the plan.

264 """

265 applied: list[GateOutcome] = []

266 rejected: list[str] = []

267 for outcome in outcomes:

268 verdict = results.get(outcome.gate)

269 if verdict is None:

270 applied.append(outcome)

271 continue

272 if not outcome.not_run:

273 rejected.append(outcome.gate)

274 applied.append(outcome)

275 continue

276 if verdict == "pass":

277 applied.append(GateOutcome(outcome.gate, True, outcome.findings,

278 error=outcome.error, on_fail=outcome.on_fail))

279 continue

280 found = outcome.findings or (

281 Finding(_ON_FAIL_SEVERITY[outcome.on_fail],

282 f"gate {outcome.gate!r} failed (reported by the dispatching agent)",

283 outcome.gate),

284 )

285 applied.append(GateOutcome(outcome.gate, False, found, error=outcome.error,

286 on_fail=outcome.on_fail))

287 return applied, rejected

288

289

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

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

292 out: list[Finding] = []

293 for o in outcomes:

294 out.extend(o.findings)

295 return out