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

1"""Load + validate project Lego extensions snapped into named backbone slots. 

2 

3An extension is a small markdown file with a YAML frontmatter mini-spec plus a 

4body (prompt or command). The contract (see ``docs/proposals/keel-architecture.md``): 

5 

6* **Add-only.** An extension may only register into one of the named 

7 :data:`keel.model.SLOTS`; it can never remove/reorder/replace a backbone step. 

8* **Fail-soft.** A broken extension degrades to a no-op (``strict=False`` returns 

9 the problems instead of raising) — unless it declares itself a hard gate 

10 (``on_fail: block``, valid only in the slots :data:`keel.model.SLOT_DEFINITIONS` 

11 marks ``may_block``: ``guard``, ``tester``, ``test`` and ``pre-merge``). This named 

12 ``pre-merge`` alone until #1100 — a restriction :func:`parse_extension` never had, and 

13 the direction of error that costs a reader work: an author who wants a blocking 

14 ``guard`` believes the slot cannot block and builds a workaround for nothing. Prose 

15 cannot compute the set, so ``tests/test_docs_claims.py`` reads the names back out of 

16 this sentence and compares them to ``may_block`` — here and in every doc that 

17 restates them. 

18* **Agent-neutral.** Each extension declares its ``agent`` (default ``inherit``). 

19""" 

20 

21from __future__ import annotations 

22 

23from dataclasses import dataclass 

24from pathlib import Path 

25from typing import TYPE_CHECKING 

26 

27from . import yaml_helper as yaml 

28from .capabilities import validate_names 

29from .model import SLOTS, slot_meta 

30 

31if TYPE_CHECKING: # pragma: no cover 

32 from .config import ProjectConfig 

33 

34KINDS: tuple[str, ...] = ("agentic", "command") 

35EXECUTION_MODES: tuple[str, ...] = ("deterministic", "agentic", "hybrid") 

36ON_FAIL: tuple[str, ...] = ("warn", "suggest", "block") 

37 

38 

39class ExtensionError(ValueError): 

40 """Raised on a malformed extension (or, in strict mode, any load problem).""" 

41 

42 

43@dataclass(frozen=True) 

44class Extension: 

45 """A parsed, validated Lego piece.""" 

46 

47 id: str 

48 slot: str 

49 kind: str 

50 mode: str 

51 agent: str 

52 on_fail: str 

53 anchorable: bool 

54 run: str | None 

55 prompt: str | None 

56 body: str 

57 source: str 

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

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

60 #: Wall-clock seconds for a ``command`` piece that is legitimately slower than the 

61 #: rest. ``None`` ⇒ inherit the project's ``knobs.gate_timeout_s``. 

62 timeout: int | None = None 

63 

64 

65def split_frontmatter(text: str) -> tuple[dict, str]: 

66 """Split ``---\\n…\\n---\\n<body>`` into (metadata dict, body). No fence ⇒ ({}, text).""" 

67 lines = text.splitlines() 

68 if not lines or lines[0].strip() != "---": 

69 return {}, text 

70 for i in range(1, len(lines)): 

71 if lines[i].strip() == "---": 

72 meta = yaml.load("\n".join(lines[1:i])) or {} 

73 body = "\n".join(lines[i + 1 :]) 

74 return meta, body 

75 raise ExtensionError("unterminated frontmatter (no closing '---')") 

76 

77 

78def parse_extension(text: str, *, source: str, expected_slot: str | None = None) -> Extension: 

79 """Parse + validate one extension file's text into an :class:`Extension`.""" 

80 meta, body = split_frontmatter(text) 

81 if not isinstance(meta, dict) or not meta: 

82 raise ExtensionError(f"{source}: missing frontmatter mini-spec") 

83 

84 errors: list[str] = [] 

85 ext_id = meta.get("id") 

86 slot = meta.get("slot") 

87 kind = meta.get("kind", "agentic") 

88 mode = meta.get("mode") 

89 on_fail = meta.get("on_fail", "warn") 

90 agent = meta.get("agent", "inherit") 

91 run = meta.get("run") 

92 prompt = meta.get("prompt") 

93 required_capabilities = tuple(meta.get("required_capabilities", [])) 

94 optional_capabilities = tuple(meta.get("optional_capabilities", [])) 

95 timeout = meta.get("timeout") 

96 

97 if not ext_id: 

98 errors.append("missing 'id'") 

99 if not slot: 

100 errors.append("missing 'slot'") 

101 elif slot not in SLOTS: 

102 errors.append(f"unknown slot {slot!r}; valid: {', '.join(SLOTS)}") 

103 elif expected_slot is not None and slot != expected_slot: 

104 errors.append(f"slot {slot!r} does not match its registered slot ({expected_slot!r})") 

105 

106 if kind not in KINDS: 

107 errors.append(f"invalid kind {kind!r}; valid: {', '.join(KINDS)}") 

108 if mode is None: 

109 mode = "deterministic" if kind == "command" else "agentic" 

110 elif mode not in EXECUTION_MODES: 

111 errors.append(f"invalid mode {mode!r}; valid: {', '.join(EXECUTION_MODES)}") 

112 if on_fail not in ON_FAIL: 

113 errors.append(f"invalid on_fail {on_fail!r}; valid: {', '.join(ON_FAIL)}") 

114 elif on_fail == "block" and slot in SLOTS and not slot_meta(slot).may_block: 

115 blocking = ", ".join(s for s in SLOTS if slot_meta(s).may_block) 

116 errors.append(f"on_fail: block is only allowed in blocking slots: {blocking}") 

117 

118 if kind == "command" and not run: 

119 errors.append("a 'command' extension requires a 'run' value") 

120 if timeout is not None: 

121 # bool is an int subclass — `timeout: true` is a typo, not a limit. 

122 if isinstance(timeout, bool) or not isinstance(timeout, int) or timeout < 1: 

123 errors.append(f"invalid timeout {timeout!r}; expected a positive integer (seconds)") 

124 if kind != "command": 

125 errors.append(f"'timeout' only applies to a 'command' extension (got kind {kind!r})") 

126 if kind == "agentic" and not (prompt or body.strip()): 

127 errors.append("an 'agentic' extension requires a 'prompt' value or a body") 

128 errors.extend(validate_names(required_capabilities, source=f"{source}: required_capabilities")) 

129 errors.extend(validate_names(optional_capabilities, source=f"{source}: optional_capabilities")) 

130 

131 if errors: 

132 raise ExtensionError(f"{source}: " + "; ".join(errors)) 

133 

134 return Extension( 

135 id=ext_id, 

136 slot=slot, 

137 kind=kind, 

138 mode=mode, 

139 agent=agent, 

140 on_fail=on_fail, 

141 anchorable=bool(meta.get("anchorable", False)), 

142 run=run, 

143 prompt=prompt, 

144 required_capabilities=required_capabilities, 

145 optional_capabilities=optional_capabilities, 

146 timeout=timeout, 

147 body=body, 

148 source=source, 

149 ) 

150 

151 

152def load_extensions( 

153 config: ProjectConfig, repo_root: str | Path, *, strict: bool = True 

154) -> tuple[dict[str, list[Extension]], list[str]]: 

155 """Load every extension referenced by ``config`` from ``repo_root``. 

156 

157 Returns ``(loaded, problems)`` where ``loaded`` maps each slot to its 

158 extensions in declared order. In ``strict`` mode any problem raises 

159 :class:`ExtensionError`; otherwise problems are returned (fail-soft) and the 

160 offending pieces are skipped. 

161 """ 

162 ext_dir = Path(repo_root) / config.extensions_dir 

163 loaded: dict[str, list[Extension]] = {slot: [] for slot in SLOTS} 

164 problems: list[str] = [] 

165 

166 for slot in SLOTS: 

167 for fname in config.slot(slot): 

168 path = ext_dir / fname 

169 try: 

170 text = path.read_text(encoding="utf-8") 

171 except OSError as exc: 

172 problems.append(f"{slot}: cannot read {fname}: {exc.strerror or exc}") 

173 continue 

174 try: 

175 loaded[slot].append(parse_extension(text, source=str(path), expected_slot=slot)) 

176 except ExtensionError as exc: 

177 problems.append(str(exc)) 

178 

179 if strict and problems: 

180 raise ExtensionError("invalid extensions:\n - " + "\n - ".join(problems)) 

181 return loaded, problems