Coverage for src/keel/extensions.py: 100%
98 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-16 18:07 +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
20import yaml
22from .capabilities import validate_names
23from .model import SLOTS, slot_meta
25if TYPE_CHECKING: # pragma: no cover
26 from .config import ProjectConfig
28KINDS: tuple[str, ...] = ("agentic", "command")
29EXECUTION_MODES: tuple[str, ...] = ("deterministic", "agentic", "hybrid")
30ON_FAIL: tuple[str, ...] = ("warn", "suggest", "block")
33class ExtensionError(ValueError):
34 """Raised on a malformed extension (or, in strict mode, any load problem)."""
37@dataclass(frozen=True)
38class Extension:
39 """A parsed, validated Lego piece."""
41 id: str
42 slot: str
43 kind: str
44 mode: str
45 agent: str
46 on_fail: str
47 anchorable: bool
48 run: str | None
49 prompt: str | None
50 body: str
51 source: str
52 required_capabilities: tuple[str, ...] = ()
53 optional_capabilities: tuple[str, ...] = ()
56def split_frontmatter(text: str) -> tuple[dict, str]:
57 """Split ``---\\n…\\n---\\n<body>`` into (metadata dict, body). No fence ⇒ ({}, text)."""
58 lines = text.splitlines()
59 if not lines or lines[0].strip() != "---":
60 return {}, text
61 for i in range(1, len(lines)):
62 if lines[i].strip() == "---":
63 meta = yaml.safe_load("\n".join(lines[1:i])) or {}
64 body = "\n".join(lines[i + 1:])
65 return meta, body
66 raise ExtensionError("unterminated frontmatter (no closing '---')")
69def parse_extension(text: str, *, source: str, expected_slot: str | None = None) -> Extension:
70 """Parse + validate one extension file's text into an :class:`Extension`."""
71 meta, body = split_frontmatter(text)
72 if not isinstance(meta, dict) or not meta:
73 raise ExtensionError(f"{source}: missing frontmatter mini-spec")
75 errors: list[str] = []
76 ext_id = meta.get("id")
77 slot = meta.get("slot")
78 kind = meta.get("kind", "agentic")
79 mode = meta.get("mode")
80 on_fail = meta.get("on_fail", "warn")
81 agent = meta.get("agent", "inherit")
82 run = meta.get("run")
83 prompt = meta.get("prompt")
84 required_capabilities = tuple(meta.get("required_capabilities", []))
85 optional_capabilities = tuple(meta.get("optional_capabilities", []))
87 if not ext_id:
88 errors.append("missing 'id'")
89 if not slot:
90 errors.append("missing 'slot'")
91 elif slot not in SLOTS:
92 errors.append(f"unknown slot {slot!r}; valid: {', '.join(SLOTS)}")
93 elif expected_slot is not None and slot != expected_slot:
94 errors.append(
95 f"slot {slot!r} does not match its registered slot ({expected_slot!r})"
96 )
98 if kind not in KINDS:
99 errors.append(f"invalid kind {kind!r}; valid: {', '.join(KINDS)}")
100 if mode is None:
101 mode = "deterministic" if kind == "command" else "agentic"
102 elif mode not in EXECUTION_MODES:
103 errors.append(f"invalid mode {mode!r}; valid: {', '.join(EXECUTION_MODES)}")
104 if on_fail not in ON_FAIL:
105 errors.append(f"invalid on_fail {on_fail!r}; valid: {', '.join(ON_FAIL)}")
106 elif on_fail == "block" and slot in SLOTS and not slot_meta(slot).may_block:
107 blocking = ", ".join(s for s in SLOTS if slot_meta(s).may_block)
108 errors.append(f"on_fail: block is only allowed in blocking slots: {blocking}")
110 if kind == "command" and not run:
111 errors.append("a 'command' extension requires a 'run' value")
112 if kind == "agentic" and not (prompt or body.strip()):
113 errors.append("an 'agentic' extension requires a 'prompt' value or a body")
114 errors.extend(validate_names(required_capabilities, source=f"{source}: required_capabilities"))
115 errors.extend(validate_names(optional_capabilities, source=f"{source}: optional_capabilities"))
117 if errors:
118 raise ExtensionError(f"{source}: " + "; ".join(errors))
120 return Extension(
121 id=ext_id,
122 slot=slot,
123 kind=kind,
124 mode=mode,
125 agent=agent,
126 on_fail=on_fail,
127 anchorable=bool(meta.get("anchorable", False)),
128 run=run,
129 prompt=prompt,
130 required_capabilities=required_capabilities,
131 optional_capabilities=optional_capabilities,
132 body=body,
133 source=source,
134 )
137def load_extensions(
138 config: ProjectConfig, repo_root: str | Path, *, strict: bool = True
139) -> tuple[dict[str, list[Extension]], list[str]]:
140 """Load every extension referenced by ``config`` from ``repo_root``.
142 Returns ``(loaded, problems)`` where ``loaded`` maps each slot to its
143 extensions in declared order. In ``strict`` mode any problem raises
144 :class:`ExtensionError`; otherwise problems are returned (fail-soft) and the
145 offending pieces are skipped.
146 """
147 ext_dir = Path(repo_root) / config.extensions_dir
148 loaded: dict[str, list[Extension]] = {slot: [] for slot in SLOTS}
149 problems: list[str] = []
151 for slot in SLOTS:
152 for fname in config.slot(slot):
153 path = ext_dir / fname
154 try:
155 text = path.read_text(encoding="utf-8")
156 except OSError as exc:
157 problems.append(f"{slot}: cannot read {fname}: {exc.strerror or exc}")
158 continue
159 try:
160 loaded[slot].append(parse_extension(text, source=str(path), expected_slot=slot))
161 except ExtensionError as exc:
162 problems.append(str(exc))
164 if strict and problems:
165 raise ExtensionError("invalid extensions:\n - " + "\n - ".join(problems))
166 return loaded, problems