Coverage for src/keel/extensions.py: 100%
105 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Load + validate project Lego extensions snapped into named backbone slots.
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``):
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 ``pre-merge`` slot).
11* **Agent-neutral.** Each extension declares its ``agent`` (default ``inherit``).
12"""
14from __future__ import annotations
16from dataclasses import dataclass
17from pathlib import Path
18from typing import TYPE_CHECKING
20from . import yaml_helper as yaml
21from .capabilities import validate_names
22from .model import SLOTS, slot_meta
24if TYPE_CHECKING: # pragma: no cover
25 from .config import ProjectConfig
27KINDS: tuple[str, ...] = ("agentic", "command")
28EXECUTION_MODES: tuple[str, ...] = ("deterministic", "agentic", "hybrid")
29ON_FAIL: tuple[str, ...] = ("warn", "suggest", "block")
32class ExtensionError(ValueError):
33 """Raised on a malformed extension (or, in strict mode, any load problem)."""
36@dataclass(frozen=True)
37class Extension:
38 """A parsed, validated Lego piece."""
40 id: str
41 slot: str
42 kind: str
43 mode: str
44 agent: str
45 on_fail: str
46 anchorable: bool
47 run: str | None
48 prompt: str | None
49 body: str
50 source: str
51 required_capabilities: tuple[str, ...] = ()
52 optional_capabilities: tuple[str, ...] = ()
53 #: Wall-clock seconds for a ``command`` piece that is legitimately slower than the
54 #: rest. ``None`` ⇒ inherit the project's ``knobs.gate_timeout_s``.
55 timeout: int | None = None
58def split_frontmatter(text: str) -> tuple[dict, str]:
59 """Split ``---\\n…\\n---\\n<body>`` into (metadata dict, body). No fence ⇒ ({}, text)."""
60 lines = text.splitlines()
61 if not lines or lines[0].strip() != "---":
62 return {}, text
63 for i in range(1, len(lines)):
64 if lines[i].strip() == "---":
65 meta = yaml.load("\n".join(lines[1:i])) or {}
66 body = "\n".join(lines[i + 1:])
67 return meta, body
68 raise ExtensionError("unterminated frontmatter (no closing '---')")
71def parse_extension(text: str, *, source: str, expected_slot: str | None = None) -> Extension:
72 """Parse + validate one extension file's text into an :class:`Extension`."""
73 meta, body = split_frontmatter(text)
74 if not isinstance(meta, dict) or not meta:
75 raise ExtensionError(f"{source}: missing frontmatter mini-spec")
77 errors: list[str] = []
78 ext_id = meta.get("id")
79 slot = meta.get("slot")
80 kind = meta.get("kind", "agentic")
81 mode = meta.get("mode")
82 on_fail = meta.get("on_fail", "warn")
83 agent = meta.get("agent", "inherit")
84 run = meta.get("run")
85 prompt = meta.get("prompt")
86 required_capabilities = tuple(meta.get("required_capabilities", []))
87 optional_capabilities = tuple(meta.get("optional_capabilities", []))
88 timeout = meta.get("timeout")
90 if not ext_id:
91 errors.append("missing 'id'")
92 if not slot:
93 errors.append("missing 'slot'")
94 elif slot not in SLOTS:
95 errors.append(f"unknown slot {slot!r}; valid: {', '.join(SLOTS)}")
96 elif expected_slot is not None and slot != expected_slot:
97 errors.append(
98 f"slot {slot!r} does not match its registered slot ({expected_slot!r})"
99 )
101 if kind not in KINDS:
102 errors.append(f"invalid kind {kind!r}; valid: {', '.join(KINDS)}")
103 if mode is None:
104 mode = "deterministic" if kind == "command" else "agentic"
105 elif mode not in EXECUTION_MODES:
106 errors.append(f"invalid mode {mode!r}; valid: {', '.join(EXECUTION_MODES)}")
107 if on_fail not in ON_FAIL:
108 errors.append(f"invalid on_fail {on_fail!r}; valid: {', '.join(ON_FAIL)}")
109 elif on_fail == "block" and slot in SLOTS and not slot_meta(slot).may_block:
110 blocking = ", ".join(s for s in SLOTS if slot_meta(s).may_block)
111 errors.append(f"on_fail: block is only allowed in blocking slots: {blocking}")
113 if kind == "command" and not run:
114 errors.append("a 'command' extension requires a 'run' value")
115 if timeout is not None:
116 # bool is an int subclass — `timeout: true` is a typo, not a limit.
117 if isinstance(timeout, bool) or not isinstance(timeout, int) or timeout < 1:
118 errors.append(f"invalid timeout {timeout!r}; expected a positive integer (seconds)")
119 if kind != "command":
120 errors.append(f"'timeout' only applies to a 'command' extension (got kind {kind!r})")
121 if kind == "agentic" and not (prompt or body.strip()):
122 errors.append("an 'agentic' extension requires a 'prompt' value or a body")
123 errors.extend(validate_names(required_capabilities, source=f"{source}: required_capabilities"))
124 errors.extend(validate_names(optional_capabilities, source=f"{source}: optional_capabilities"))
126 if errors:
127 raise ExtensionError(f"{source}: " + "; ".join(errors))
129 return Extension(
130 id=ext_id,
131 slot=slot,
132 kind=kind,
133 mode=mode,
134 agent=agent,
135 on_fail=on_fail,
136 anchorable=bool(meta.get("anchorable", False)),
137 run=run,
138 prompt=prompt,
139 required_capabilities=required_capabilities,
140 optional_capabilities=optional_capabilities,
141 timeout=timeout,
142 body=body,
143 source=source,
144 )
147def load_extensions(
148 config: ProjectConfig, repo_root: str | Path, *, strict: bool = True
149) -> tuple[dict[str, list[Extension]], list[str]]:
150 """Load every extension referenced by ``config`` from ``repo_root``.
152 Returns ``(loaded, problems)`` where ``loaded`` maps each slot to its
153 extensions in declared order. In ``strict`` mode any problem raises
154 :class:`ExtensionError`; otherwise problems are returned (fail-soft) and the
155 offending pieces are skipped.
156 """
157 ext_dir = Path(repo_root) / config.extensions_dir
158 loaded: dict[str, list[Extension]] = {slot: [] for slot in SLOTS}
159 problems: list[str] = []
161 for slot in SLOTS:
162 for fname in config.slot(slot):
163 path = ext_dir / fname
164 try:
165 text = path.read_text(encoding="utf-8")
166 except OSError as exc:
167 problems.append(f"{slot}: cannot read {fname}: {exc.strerror or exc}")
168 continue
169 try:
170 loaded[slot].append(parse_extension(text, source=str(path), expected_slot=slot))
171 except ExtensionError as exc:
172 problems.append(str(exc))
174 if strict and problems:
175 raise ExtensionError("invalid extensions:\n - " + "\n - ".join(problems))
176 return loaded, problems