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

1"""Thin I/O: execute shell-command gates (build / lint / command extensions). 

2 

3This is the only place keel shells out for gates. It is deliberately thin and 

4**fail-soft**: a timeout or a missing binary becomes a failed :class:`CommandResult` 

5rather than an exception. The subprocess call is injectable (``_run``) so the gate 

6runner is fully unit-testable offline; agentic gates are dispatched elsewhere. 

7""" 

8 

9from __future__ import annotations 

10 

11import re 

12import subprocess # nosec B404 

13from collections.abc import Callable 

14from dataclasses import dataclass 

15from typing import TYPE_CHECKING 

16 

17from .findings import Finding 

18from .model import DEFAULT_GATE_TIMEOUT_S 

19 

20if TYPE_CHECKING: # pragma: no cover 

21 from .gates import GateSpec 

22 

23GateRunner = Callable[ 

24 ["GateSpec"], "tuple[bool, list[Finding]] | tuple[bool, list[Finding], bool]" 

25] 

26 

27_ON_FAIL_SEVERITY = {"block": "major", "suggest": "minor", "warn": "nit"} 

28 

29#: reviewdog-style errorformat: ``path:line[:col]: message`` (first hit wins). 

30#: A single multiline ``search`` replaces a per-``splitlines`` loop: ``^`` is 

31#: anchored to each line by ``re.MULTILINE``, the path classes exclude ``\n`` so a 

32#: match can never span lines, and the trailing ``(?:[:\s]|$)`` accepts the line 

33#: number at end-of-line. 

34_LOCATION_RE = re.compile( 

35 r"^[ \t]*(?P<path>[^\s\n:][^:\n]*?):(?P<line>\d+)(?::\d+)?(?:[:\s]|$)", 

36 re.MULTILINE, 

37) 

38 

39 

40def first_location(text: str) -> tuple[str | None, int | None]: 

41 """Extract the first ``path:line`` location from tool output (``(None, None)`` if none).""" 

42 m = _LOCATION_RE.search(text) 

43 return (m.group("path"), int(m.group("line"))) if m else (None, None) 

44 

45 

46@dataclass(frozen=True) 

47class CommandResult: 

48 ok: bool 

49 code: int 

50 #: ``stdout + stderr``, concatenated. Kept for the diagnostic uses that genuinely 

51 #: want both (a failing gate's message, an output tail). **Do not parse structured 

52 #: data out of this** — a command that writes progress or warnings to stderr while 

53 #: exiting 0 (git's ``warning: refname … is ambiguous``, ai-jury's ``[jury] …`` 

54 #: logs) leaves the real payload glued to noise. Parse :attr:`stdout` instead. 

55 output: str 

56 #: True when the wall-clock timeout killed the command (exit 124). A timeout is 

57 #: still a failure — ``ok`` stays False — but it carries no pass/fail verdict, so 

58 #: callers can label it distinctly instead of reporting it as a broken test. 

59 timed_out: bool = False 

60 #: Captured standard output alone. This is what parsers must read: a tool's 

61 #: machine-readable result goes here, never contaminated by stderr diagnostics. 

62 stdout: str = "" 

63 #: Captured standard error alone. 

64 stderr: str = "" 

65 

66 

67def _result(proc) -> CommandResult: 

68 out = proc.stdout or "" 

69 err = proc.stderr or "" 

70 return CommandResult(proc.returncode == 0, proc.returncode, out + err, 

71 stdout=out, stderr=err) 

72 

73 

74def run_command( 

75 cmd: str, *, cwd: str | None = None, timeout: int = DEFAULT_GATE_TIMEOUT_S, _run=subprocess.run 

76) -> CommandResult: 

77 """Run ``cmd`` in a shell, capturing output. Fail-soft on timeout/OS error.""" 

78 try: 

79 # Intentional shell boundary: cmd must come only from operator-controlled 

80 # project config or extension YAML, never from PR content or agent output. 

81 proc = _run(cmd, shell=True, cwd=cwd, capture_output=True, text=True, timeout=timeout) # nosec B604 

82 except subprocess.TimeoutExpired: 

83 return CommandResult(False, 124, f"timed out after {timeout}s", timed_out=True) 

84 except OSError as exc: 

85 return CommandResult(False, 127, str(exc), stderr=str(exc)) 

86 return _result(proc) 

87 

88 

89def run_argv( 

90 argv: list[str], *, cwd: str | None = None, timeout: int = 120, _run=subprocess.run 

91) -> CommandResult: 

92 """Run an argv list (no shell). Fail-soft on timeout/OS error. Used by git/gh wrappers.""" 

93 try: 

94 proc = _run(argv, cwd=cwd, capture_output=True, text=True, timeout=timeout) 

95 except subprocess.TimeoutExpired: 

96 return CommandResult(False, 124, f"timed out after {timeout}s", timed_out=True) 

97 except OSError as exc: 

98 return CommandResult(False, 127, str(exc), stderr=str(exc)) 

99 return _result(proc) 

100 

101 

102def _tail(text: str, n: int = 20) -> str: 

103 return "\n".join(text.strip().splitlines()[-n:]) 

104 

105 

106def command_gate_runner( 

107 repo_root: str | None = None, 

108 *, 

109 timeout: int = DEFAULT_GATE_TIMEOUT_S, 

110 _run=subprocess.run, 

111) -> GateRunner: 

112 """A :data:`keel.gates.GateRunner` that executes ``command`` gates via the shell. 

113 

114 Non-command gates (agentic / builtin like ``jury``) are not executed here — in 

115 command-only mode they pass as no-ops; the agent-dispatch layer runs those. 

116 

117 ``timeout`` is the fallback wall-clock limit for a gate that carries none of its 

118 own; a :attr:`~keel.gates.GateSpec.timeout` resolved by 

119 :func:`~keel.gates.plan_gates` always wins. A gate killed by that limit is 

120 reported as a **timeout** rather than a failure: it still blocks (``ok`` is 

121 False and the severity is unchanged), but the message says the command never 

122 produced a verdict instead of implying a test broke. 

123 """ 

124 

125 def runner(spec: GateSpec) -> tuple[bool, list[Finding], bool, bool]: 

126 if spec.kind != "command" or not spec.run: 

127 # Not executed here — the agent-dispatch layer runs agentic gates. Flagged 

128 # `not_run` so this can never be recorded as "ran and passed"; `ok` stays 

129 # True so a soft gate does not spuriously fail a command-only run. 

130 return True, [], False, True 

131 limit = timeout if spec.timeout is None else spec.timeout 

132 result = run_command(spec.run, cwd=repo_root, timeout=limit, _run=_run) 

133 if result.ok: 

134 return True, [], False, False 

135 severity = _ON_FAIL_SEVERITY[spec.on_fail] 

136 if result.timed_out: 

137 # No pass/fail verdict exists — do not dress the kill up as a test result. 

138 message = ( 

139 f"{spec.id} timed out after {limit}s (exit {result.code}); " 

140 "the command produced no pass/fail result. Raise the limit via " 

141 "knobs.gate_timeout_s (or this gate's timeout:) if it legitimately " 

142 "needs longer — a genuinely hanging command is still a defect." 

143 ) 

144 return False, [Finding(severity, message, spec.id)], True, False 

145 message = f"{spec.id} failed (exit {result.code})" 

146 tail = _tail(result.output) 

147 if tail: 

148 message += f": {tail}" 

149 path, line = first_location(result.output) 

150 return False, [Finding( 

151 severity, message, spec.id, 

152 path=path, line=line, anchorable=path is not None and line is not None, 

153 )], False, False 

154 

155 return runner