Coverage for little_loops / adapters / codex.py: 85%
232 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -0500
1"""CodexEmitter: host adapter for the Codex CLI.
3Migrated from ``cli/adapt_skills_for_codex.py`` and
4``cli/adapt_agents_for_codex.py`` under FEAT-2391.
5"""
7from __future__ import annotations
9import re
10import sys
11from pathlib import Path
13import yaml
15from little_loops.adapters.core import AdapterError, _extract_body # noqa: F401
17__all__ = ["CodexEmitter"]
19_MARKER = "# generated by ll-adapt"
20_MAX_SHORT_DESC = 80
21_FM_CLOSE_RE = re.compile(r"\n---\s*\n")
23# Recognised ll-generated markers for idempotency; includes the legacy value so
24# existing files are not treated as user-authored and can be updated in place.
25_LL_GENERATED_MARKERS = (
26 _MARKER,
27 "# generated by ll-adapt-agents-for-codex",
28)
30_READ_ONLY_TOOLS = frozenset({"Read", "Glob", "Grep", "WebFetch", "WebSearch"})
31_WRITE_TOOLS = frozenset({"Edit", "Write", "Bash", "NotebookEdit"})
34# ---------------------------------------------------------------------------
35# Skill helpers (migrated from adapt_skills_for_codex.py)
36# ---------------------------------------------------------------------------
39def _extract_skill_short_desc(text: str) -> str:
40 """Parse skill/command content and return first non-empty description line, ≤80 chars."""
41 if not text.startswith("---"):
42 return ""
43 end = text.find("---", 3)
44 if end == -1:
45 return ""
46 fm_raw = text[3:end]
47 try:
48 fm = yaml.safe_load(fm_raw) or {}
49 except yaml.YAMLError:
50 return ""
51 desc = fm.get("description", "") or ""
52 if not isinstance(desc, str):
53 return ""
54 for line in desc.splitlines():
55 stripped = line.strip()
56 if stripped:
57 return stripped[:_MAX_SHORT_DESC]
58 return ""
61def _insert_skill_fields(content: str, name: str, short_desc: str) -> tuple[str, bool]:
62 """Insert ``name:`` and ``metadata.short-description:`` into SKILL.md frontmatter.
64 Uses targeted string manipulation — no yaml roundtrip — to preserve
65 existing frontmatter formatting. Returns ``(new_content, changed)``.
66 """
67 if not content.startswith("---\n"):
68 return content, False
70 m = _FM_CLOSE_RE.search(content[3:])
71 if not m:
72 return content, False
74 fm_text = content[4:3 + m.start()]
75 after = content[3 + m.start():]
77 changed = False
79 if not re.search(r"^name\s*:", fm_text, re.MULTILINE):
80 fm_text = f"name: {name}\n" + fm_text
81 changed = True
83 if "short-description:" not in fm_text:
84 if re.search(r"^metadata\s*:", fm_text, re.MULTILINE):
85 fm_text = re.sub(
86 r"^(metadata\s*:.*)$",
87 lambda mtch: mtch.group(0) + f"\n short-description: {short_desc}",
88 fm_text,
89 flags=re.MULTILINE,
90 count=1,
91 )
92 else:
93 fm_text += f"\nmetadata:\n short-description: {short_desc}"
94 changed = True
96 return f"---\n{fm_text}{after}", changed
99def _title_case(slug: str) -> str:
100 """Convert a kebab-case slug to title-cased display name."""
101 return " ".join(word.capitalize() for word in slug.replace("-", " ").split())
104def _make_openai_yaml_content(display_name: str, short_desc: str) -> str:
105 """Return ``agents/openai.yaml`` content for the Codex Skills API."""
106 return (
107 f'interface:\n'
108 f' display_name: "{display_name}"\n'
109 f' short_description: "{short_desc}"\n'
110 )
113def _synthesized_skill_md(stem: str, description: str) -> str:
114 """Build a minimal synthesized SKILL.md for a bridged command."""
115 short_desc = ""
116 for line in description.splitlines():
117 stripped = line.strip()
118 if stripped:
119 short_desc = stripped[:_MAX_SHORT_DESC]
120 break
122 if "\n" in description.strip():
123 indented = "\n".join(f" {line}" if line else "" for line in description.splitlines())
124 desc_block = f"description: |\n{indented}"
125 else:
126 desc_block = f"description: {description.strip()}"
128 return (
129 f"---\n"
130 f"name: ll-{stem}\n"
131 f"{desc_block}\n"
132 f"metadata:\n"
133 f" short-description: {short_desc}\n"
134 f"---\n"
135 f"\n"
136 f"# {_title_case(stem)}\n"
137 f"\n"
138 f"Bridged from `commands/{stem}.md` for Codex Skills API discovery.\n"
139 f"See the source command file for the full prompt body.\n"
140 )
143# ---------------------------------------------------------------------------
144# Agent helpers (migrated from adapt_agents_for_codex.py)
145# ---------------------------------------------------------------------------
148def _extract_agent_short_desc(text: str) -> str:
149 """Return first non-empty description line from agent frontmatter, ≤80 chars."""
150 if not text.startswith("---"):
151 return ""
152 end = text.find("---", 3)
153 if end == -1:
154 return ""
155 try:
156 fm = yaml.safe_load(text[3:end]) or {}
157 except yaml.YAMLError:
158 return ""
159 desc = fm.get("description", "") or ""
160 if not isinstance(desc, str):
161 return ""
162 for line in desc.splitlines():
163 stripped = line.strip()
164 if stripped:
165 return stripped[:_MAX_SHORT_DESC]
166 return ""
169# ---------------------------------------------------------------------------
170# Rich TOML field derivation (ENH-2121 absorbed into FEAT-2391)
171# ---------------------------------------------------------------------------
174def _derive_sandbox_mode(tools: list | None) -> str | None:
175 """Derive Codex ``sandbox_mode`` from the agent's tools list.
177 Returns ``"read-only"`` when all tools are known read-only operations,
178 ``"write-to-cwd"`` when any write tool is present, or ``None`` when the
179 tools list is absent (field omitted; Codex inherits from the session).
180 """
181 if not tools:
182 return None
183 tool_set = set(tools)
184 if not (tool_set & _WRITE_TOOLS):
185 return "read-only"
186 return "write-to-cwd"
189def _derive_mcp_servers(tools: list | None) -> list[str] | None:
190 """Derive Codex ``mcp_servers`` from MCP tool references in the tools list.
192 Returns a deduplicated list of server names extracted from ``mcp__<server>__*``
193 patterns, or ``None`` when no MCP tools are present.
194 """
195 if not tools:
196 return None
197 servers: list[str] = []
198 for tool in tools:
199 if isinstance(tool, str) and tool.startswith("mcp__"):
200 parts = tool.split("__")
201 if len(parts) >= 2:
202 server = parts[1]
203 if server not in servers:
204 servers.append(server)
205 return servers if servers else None
208def _format_agent_toml(
209 name: str,
210 description: str,
211 model: str,
212 body: str,
213 fm: dict,
214) -> str:
215 """Return TOML content for ``.codex/agents/<name>.toml``.
217 Emits the four core fields plus optional rich fields derived from *fm*
218 (``sandbox_mode``, ``mcp_servers``) when source data is available.
219 """
220 escaped_desc = description.replace("\\", "\\\\").replace('"', '\\"')
221 escaped_model = model.replace("\\", "\\\\").replace('"', '\\"')
222 safe_body = body.replace('"""', '\\"\\"\\"')
223 if not safe_body.endswith("\n"):
224 safe_body += "\n"
226 result = (
227 f"{_MARKER}\n"
228 f'name = "{name}"\n'
229 f'description = "{escaped_desc}"\n'
230 f'model = "{escaped_model}"\n'
231 )
233 # Resolve tools from frontmatter for rich-field derivation
234 tools = fm.get("tools")
235 if isinstance(tools, str):
236 try:
237 tools = yaml.safe_load(tools) or []
238 except yaml.YAMLError:
239 tools = []
241 sandbox_mode = _derive_sandbox_mode(tools)
242 if sandbox_mode:
243 result += f'sandbox_mode = "{sandbox_mode}"\n'
245 mcp_servers = _derive_mcp_servers(tools)
246 if mcp_servers:
247 servers_str = ", ".join(f'"{s}"' for s in mcp_servers)
248 result += f"mcp_servers = [{servers_str}]\n"
250 result += (
251 f"\n"
252 f'developer_instructions = """\n'
253 f"{safe_body}"
254 f'"""\n'
255 )
256 return result
259# ---------------------------------------------------------------------------
260# CodexEmitter
261# ---------------------------------------------------------------------------
264class CodexEmitter:
265 """Output emitter for the Codex host (``--host codex``).
267 Each ``emit_*`` method receives a metadata dict built by the core traversal
268 functions and returns ``"adapted"``, ``"skipped"``, or ``"error"``.
269 """
271 name = "codex"
273 def emit_skill(self, skill_meta: dict) -> str:
274 """Add Codex Skills API frontmatter to one SKILL.md."""
275 skill_name: str = skill_meta["skill_name"]
276 skill_path: Path = skill_meta["skill_path"]
277 content: str = skill_meta["content"]
278 apply: bool = skill_meta["apply"]
279 quiet: bool = skill_meta["quiet"]
281 short_desc = _extract_skill_short_desc(content)
282 if not short_desc:
283 if not quiet:
284 print(f" SKIP {skill_name}: no description found")
285 return "skipped"
287 new_text, skill_changed = _insert_skill_fields(content, skill_name, short_desc)
288 openai_yaml = skill_path.parent / "agents" / "openai.yaml"
289 yaml_exists = openai_yaml.exists()
291 if not skill_changed and yaml_exists:
292 if not quiet:
293 print(f" SKIP {skill_name}: already adapted")
294 return "skipped"
296 if apply:
297 if skill_changed:
298 skill_path.write_text(new_text)
299 if not yaml_exists:
300 openai_yaml.parent.mkdir(exist_ok=True)
301 openai_yaml.write_text(_make_openai_yaml_content(_title_case(skill_name), short_desc))
302 if not quiet:
303 print(f" APPLY {skill_name}: {short_desc[:50]}")
304 else:
305 if not quiet:
306 print(f" DRY {skill_name}: {short_desc[:50]}")
308 return "adapted"
310 def emit_command(self, cmd_meta: dict) -> str:
311 """Bridge one command into a Codex-discoverable skills/ll-<stem>/ wrapper."""
312 stem: str = cmd_meta["stem"]
313 label = f"ll-{stem}"
314 content: str = cmd_meta["content"]
315 fm: dict = cmd_meta["fm"]
316 output_dir: Path = cmd_meta["output_dir"]
317 apply: bool = cmd_meta["apply"]
318 quiet: bool = cmd_meta["quiet"]
320 description = fm.get("description", "") or ""
321 if not isinstance(description, str) or not description.strip():
322 if not quiet:
323 print(f" SKIP {label}: no description in frontmatter")
324 return "skipped"
326 short_desc = _extract_skill_short_desc(content)
327 if not short_desc:
328 if not quiet:
329 print(f" SKIP {label}: empty short description")
330 return "skipped"
332 out_dir = output_dir / f"ll-{stem}"
333 out_skill_md = out_dir / "SKILL.md"
334 out_openai_yaml = out_dir / "agents" / "openai.yaml"
336 if out_skill_md.exists() and out_openai_yaml.exists():
337 if not quiet:
338 print(f" SKIP {label}: already adapted")
339 return "skipped"
341 if apply:
342 out_dir.mkdir(parents=True, exist_ok=True)
343 if not out_skill_md.exists():
344 out_skill_md.write_text(_synthesized_skill_md(stem, description))
345 if not out_openai_yaml.exists():
346 out_openai_yaml.parent.mkdir(exist_ok=True)
347 out_openai_yaml.write_text(_make_openai_yaml_content(_title_case(stem), short_desc))
348 if not quiet:
349 print(f" APPLY {label}: {short_desc[:50]}")
350 else:
351 if not quiet:
352 print(f" DRY {label}: {short_desc[:50]}")
354 return "adapted"
356 def emit_agent(self, agent_meta: dict) -> str:
357 """Emit ``.codex/agents/<name>.toml`` for one agent definition."""
358 agent_name: str = agent_meta["agent_name"]
359 content: str = agent_meta["content"]
360 fm: dict = agent_meta["fm"]
361 output_dir: Path = agent_meta["output_dir"]
362 apply: bool = agent_meta["apply"]
363 quiet: bool = agent_meta["quiet"]
365 if not fm:
366 if not quiet:
367 print(f" SKIP {agent_name}: no parseable frontmatter")
368 return "skipped"
370 name = str(fm.get("name") or agent_name)
371 model = str(fm.get("model") or "")
372 short_desc = _extract_agent_short_desc(content)
373 if not short_desc:
374 if not quiet:
375 print(f" SKIP {agent_name}: no description found")
376 return "skipped"
378 body = _extract_body(content)
379 out_toml = output_dir / f"{agent_name}.toml"
380 new_content = _format_agent_toml(name, short_desc, model, body, fm)
382 if out_toml.exists():
383 existing = out_toml.read_text()
384 if not any(existing.startswith(m) for m in _LL_GENERATED_MARKERS):
385 if not quiet:
386 print(f" SKIP {agent_name}: user-authored file (no marker)")
387 return "skipped"
388 if existing == new_content:
389 if not quiet:
390 print(f" SKIP {agent_name}: already up to date")
391 return "skipped"
393 if apply:
394 output_dir.mkdir(parents=True, exist_ok=True)
395 out_toml.write_text(new_content)
396 if not quiet:
397 print(f" APPLY {agent_name}: {short_desc[:50]}")
398 else:
399 if not quiet:
400 print(f" DRY {agent_name}: {short_desc[:50]}")
402 return "adapted"