Coverage for little_loops / adapters / core.py: 87%
126 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -0500
1"""Host-parameterized adapter core for skill/command/agent emission.
3Defines the :class:`HostEmitter` Protocol and a registry-backed
4:func:`resolve_emitter` factory. Concrete emitters live in sibling modules
5(``codex.py``, ``omp.py``); this module owns only the shared interface,
6registry, shared helpers, and traversal functions.
8See FEAT-2391 for the full design.
9"""
11from __future__ import annotations
13import importlib
14import sys
15from pathlib import Path
16from typing import Protocol, cast, runtime_checkable
18import yaml
21class AdapterError(Exception):
22 """Raised when a host emitter cannot fulfil the request."""
25@runtime_checkable
26class HostEmitter(Protocol):
27 """Protocol for host-specific output emitters.
29 Implementations convert ll skill/command/agent metadata into the target
30 host's discovery format. Structurally matched — any class with ``name``
31 and the three ``emit_*`` methods satisfies this Protocol without explicit
32 subclassing. ``@runtime_checkable`` enables ``isinstance`` registry checks.
33 """
35 name: str
37 def emit_skill(self, skill_meta: dict) -> str: ...
38 def emit_command(self, cmd_meta: dict) -> str: ...
39 def emit_agent(self, agent_meta: dict) -> str: ...
42# Lazy-import registry: host name → (module_path, class_name).
43# Concrete modules import from this module (AdapterError, helpers), so we must
44# not import them at the module level — only resolve on demand via importlib.
45_EMITTER_MAP: dict[str, tuple[str, str]] = {
46 "codex": ("little_loops.adapters.codex", "CodexEmitter"),
47 "omp": ("little_loops.adapters.omp", "OmpEmitter"),
48}
51def resolve_emitter(host: str) -> HostEmitter:
52 """Return a :class:`HostEmitter` instance for *host*.
54 Args:
55 host: One of ``"codex"``, ``"omp"``.
57 Returns:
58 A :class:`HostEmitter` ready to emit skills, commands, and agents.
60 Raises:
61 AdapterError: if *host* is not registered.
62 """
63 entry = _EMITTER_MAP.get(host)
64 if entry is None:
65 raise AdapterError(
66 f"Host {host!r} is not registered. Available: {sorted(_EMITTER_MAP)}."
67 )
68 module_path, cls_name = entry
69 module = importlib.import_module(module_path)
70 return cast(HostEmitter, getattr(module, cls_name)())
73# ---------------------------------------------------------------------------
74# Shared frontmatter helpers
75# ---------------------------------------------------------------------------
78def _read_frontmatter(text: str) -> dict | None:
79 """Parse YAML frontmatter from *text*. Returns None on any failure."""
80 if not text.startswith("---"):
81 return None
82 end = text.find("---", 3)
83 if end == -1:
84 return None
85 try:
86 fm = yaml.safe_load(text[3:end]) or {}
87 except yaml.YAMLError:
88 return None
89 return fm if isinstance(fm, dict) else None
92def _extract_body(text: str) -> str:
93 """Return body text after the closing ``---`` of frontmatter."""
94 if not text.startswith("---"):
95 return ""
96 end = text.find("---", 3)
97 if end == -1:
98 return ""
99 after_fm = text[end + 3:]
100 if after_fm.startswith("\n"):
101 after_fm = after_fm[1:]
102 return after_fm
105def _is_model_invocation_disabled(fm: dict) -> bool:
106 """Return True if *fm* has ``disable-model-invocation`` set to a truthy value.
108 Handles both native YAML booleans and stringified values (``"true"``,
109 ``"yes"``, ``"1"``) for compatibility with ``parse_skill_frontmatter``
110 which returns a flat ``dict[str, str]``.
111 """
112 val = fm.get("disable-model-invocation")
113 if val is None:
114 return False
115 if isinstance(val, bool):
116 return val
117 return str(val).strip().lower() in {"true", "yes", "1"}
120# ---------------------------------------------------------------------------
121# Shared traversal functions
122# ---------------------------------------------------------------------------
125def process_skills(
126 emitter: HostEmitter,
127 skills_dir: Path,
128 apply: bool,
129 quiet: bool,
130) -> tuple[int, int, int]:
131 """Walk *skills_dir*, apply the ``disable-model-invocation`` filter, and
132 call ``emitter.emit_skill`` for each eligible skill.
134 Returns:
135 ``(adapted, skipped, errors)`` counts.
136 """
137 adapted = skipped = errors = 0
139 for skill_md in sorted(skills_dir.glob("*/SKILL.md")):
140 skill_name = skill_md.parent.name
141 try:
142 content = skill_md.read_text()
143 except OSError as exc:
144 if not quiet:
145 print(f" ERROR {skill_name}: cannot read: {exc}", file=sys.stderr)
146 errors += 1
147 continue
149 fm = _read_frontmatter(content) or {}
150 if _is_model_invocation_disabled(fm):
151 if not quiet:
152 print(f" SKIP {skill_name}: disable-model-invocation: true")
153 skipped += 1
154 continue
156 result = emitter.emit_skill(
157 {
158 "skill_name": skill_name,
159 "skill_path": skill_md,
160 "content": content,
161 "fm": fm,
162 "apply": apply,
163 "quiet": quiet,
164 }
165 )
166 if result == "adapted":
167 adapted += 1
168 elif result == "skipped":
169 skipped += 1
170 else:
171 errors += 1
173 return adapted, skipped, errors
176def process_commands(
177 emitter: HostEmitter,
178 commands_dir: Path,
179 output_dir: Path,
180 apply: bool,
181 quiet: bool,
182) -> tuple[int, int, int]:
183 """Walk *commands_dir*, apply the ``disable-model-invocation`` filter, and
184 call ``emitter.emit_command`` for each eligible command.
186 Args:
187 output_dir: Host-specific destination for synthesized skill wrappers
188 (e.g. ``skills/`` for Codex).
190 Returns:
191 ``(adapted, skipped, errors)`` counts.
192 """
193 adapted = skipped = errors = 0
195 if not commands_dir.exists():
196 return adapted, skipped, errors
198 for cmd_md in sorted(commands_dir.glob("*.md")):
199 stem = cmd_md.stem
200 label = f"ll-{stem}"
201 try:
202 content = cmd_md.read_text()
203 except OSError as exc:
204 if not quiet:
205 print(f" ERROR {label}: cannot read: {exc}", file=sys.stderr)
206 errors += 1
207 continue
209 fm = _read_frontmatter(content)
210 if fm is None:
211 if not quiet:
212 print(f" SKIP {label}: no parseable frontmatter")
213 skipped += 1
214 continue
216 if _is_model_invocation_disabled(fm):
217 if not quiet:
218 print(f" SKIP {label}: disable-model-invocation: true")
219 skipped += 1
220 continue
222 result = emitter.emit_command(
223 {
224 "stem": stem,
225 "cmd_path": cmd_md,
226 "content": content,
227 "fm": fm,
228 "output_dir": output_dir,
229 "apply": apply,
230 "quiet": quiet,
231 }
232 )
233 if result == "adapted":
234 adapted += 1
235 elif result == "skipped":
236 skipped += 1
237 else:
238 errors += 1
240 return adapted, skipped, errors
243def process_agents(
244 emitter: HostEmitter,
245 agents_dir: Path,
246 output_dir: Path,
247 apply: bool,
248 quiet: bool,
249 only: str | None = None,
250) -> tuple[int, int, int]:
251 """Walk *agents_dir* and call ``emitter.emit_agent`` for each agent file.
253 Agents whose stem does not match *only* (when set) are silently skipped
254 and NOT counted in any return bucket.
256 Args:
257 output_dir: Host-specific destination for emitted artefacts
258 (e.g. ``.codex/agents/`` for Codex).
259 only: If non-None, restrict processing to the single agent with this
260 stem name.
262 Returns:
263 ``(adapted, skipped, errors)`` counts.
264 """
265 adapted = skipped = errors = 0
267 for agent_md in sorted(agents_dir.glob("*.md")):
268 agent_name = agent_md.stem
270 if only is not None and agent_name != only:
271 continue # silently dropped, not counted
273 try:
274 content = agent_md.read_text()
275 except OSError as exc:
276 if not quiet:
277 print(f" ERROR {agent_name}: cannot read: {exc}", file=sys.stderr)
278 errors += 1
279 continue
281 fm = _read_frontmatter(content) or {}
283 result = emitter.emit_agent(
284 {
285 "agent_name": agent_name,
286 "agent_path": agent_md,
287 "content": content,
288 "fm": fm,
289 "output_dir": output_dir,
290 "apply": apply,
291 "quiet": quiet,
292 }
293 )
294 if result == "adapted":
295 adapted += 1
296 elif result == "skipped":
297 skipped += 1
298 else:
299 errors += 1
301 return adapted, skipped, errors