"""Codex CLI adapter. Emits Codex-native artifacts: - `.codex/skills/__/SKILL.md` (8 KB body cap; overflow into references/) — gitignored - `.codex/agents/__.toml` (markdown YAML -> TOML transform) — gitignored - `plugins//.codex-plugin/plugin.json` — committed per-plugin manifest; the Codex marketplace installs the source plugin dir directly (reads `SKILL.md` from `./skills/`). - `.agents/plugins/marketplace.json` — committed Codex marketplace registry listing every plugin. `AGENTS.md` is NOT generated by this adapter — it's the committed cross-harness context file at the repo root (read directly by Codex/Cursor/OpenCode; Claude Code reads it via `CLAUDE.md`, a symlink to `AGENTS.md`; redirected from Gemini via `gemini-extension.json` / `.gemini/settings.json`). `emit_global` validates that file fits Codex's 32 KiB cap and emits the marketplace registry. Sources: research summary by `a09b06ce7e593de86` synthesized into the plan. """ from __future__ import annotations import json from pathlib import Path from tools.adapters.base import ( WORKTREE, AgentSource, EmitResult, HarnessAdapter, PluginSource, SkillSource, ) from tools.adapters.capabilities import TOOL_NAME_MAPS, resolve_model # Fields that Codex silently ignores; stripping them is honest. _CLAUDE_ONLY_SKILL_FIELDS = { "allowed-tools", "context", "model", "hooks", "agent", "user-invocable", "disable-model-invocation", } _CLAUDE_ONLY_AGENT_FIELDS = { "color", "tools", "allowed-tools", "context", "hooks", "user-invocable", "disable-model-invocation", } # ── TOML emission (hand-rolled; no toml writer in stdlib) ──────────────────── def _escape_toml_basic(s: str) -> str: return s.replace("\\", "\\\\").replace('"', '\\"') def _escape_toml_multiline(s: str) -> str: # Triple-quoted multi-line basic strings. Escape only triple-double-quote sequences. return s.replace("\\", "\\\\").replace('"""', '\\"\\"\\"') def _toml_kv(key: str, value) -> str: if isinstance(value, bool): return f"{key} = {'true' if value else 'false'}" if isinstance(value, int): return f"{key} = {value}" s = str(value) if "\n" in s: return f'{key} = """\n{_escape_toml_multiline(s)}\n"""' return f'{key} = "{_escape_toml_basic(s)}"' # ── Skill rewriting ────────────────────────────────────────────────────────── def _filter_frontmatter(fm: dict, drop: set[str]) -> dict: return {k: v for k, v in fm.items() if k not in drop} _YAML_SPECIAL_LEADS = ( "[", "{", "*", "&", "!", "|", ">", "'", '"', "@", "`", "#", "%", ",", "?", ":", "-", ) # YAML 1.1 implicit booleans/null — must be quoted to avoid being interpreted as bool/None. # YAML 1.2 narrowed this list, but PyYAML's default is still 1.1 (and many consumers are # affected); quote conservatively. _YAML_RESERVED_WORDS = frozenset( { "true", "false", "yes", "no", "on", "off", "null", "~", "True", "False", "Yes", "No", "On", "Off", "Null", "TRUE", "FALSE", "YES", "NO", "ON", "OFF", "NULL", } ) def _yaml_scalar(value: str) -> str: """Render a string as a YAML scalar, quoting when needed to avoid ambiguity. Quotes when the value: - is empty / pure whitespace - starts with a YAML special character - contains `:` followed by whitespace (would be interpreted as a key) - contains ` #` (would be interpreted as a comment) - has leading or trailing whitespace - starts with a digit or `-`/`+` (number-like) - matches a YAML 1.1 implicit-boolean/null reserved word """ s = str(value).replace("\n", " ") needs_quote = ( s == "" or s != s.strip() or s.startswith(_YAML_SPECIAL_LEADS) or ": " in s or " #" in s or s[:1].isdigit() or s in _YAML_RESERVED_WORDS ) if needs_quote: # Use double quotes; escape embedded double-quotes and backslashes. escaped = s.replace("\\", "\\\\").replace('"', '\\"') return f'"{escaped}"' return s def _frontmatter_block(fm: dict) -> str: """Render a minimal YAML-ish frontmatter block (string/scalar fields only).""" lines = ["---"] for k, v in fm.items(): if isinstance(v, list): lines.append(f"{k}:") for item in v: lines.append(f" - {_yaml_scalar(item)}") elif isinstance(v, dict): lines.append(f"{k}:") for subk, subv in v.items(): lines.append(f" {subk}: {_yaml_scalar(subv)}") elif v is None: continue else: lines.append(f"{k}: {_yaml_scalar(v)}") lines.append("---") return "\n".join(lines) def _rewrite_body_for_codex(body: str) -> str: """Replace 'the Read tool' / 'The Bash tool' / 'the `Grep` tool' -> action verbs. Matches plugin_eval/layers/harness_portability.py's _TOOL_PROSE_PATTERN exactly: the leading article is case-insensitive, but the tool name must match exact CamelCase. This prevents `the bash tool` (referring to the shell, lowercase) from being rewritten — it would be a false-positive that the lint correctly leaves alone. Codex prompts encourage action verbs over tool-name vocabulary. """ import re mapping = TOOL_NAME_MAPS["codex"] out = body for camel, replacement in mapping.items(): # `(?i:the)` makes only the article case-insensitive; tool name stays CamelCase. out = re.sub( rf"(?i:\bthe)\s+`?{re.escape(camel)}`?\s+tool\b", replacement, out, ) return out _POINTER = ( "\n\n> Detailed reference moved to `references/details.md` to fit the Codex" " 8 KB skill body budget. Read that file when the section above is insufficient.\n" ) _POINTER_BYTES = len(_POINTER.encode("utf-8")) def _split_body_fence_aware(body: str) -> list[str]: """Split body on `## ` headings, but only when they appear OUTSIDE fenced code blocks. A fence is opened/closed by a line starting with ``` (any number of backticks ≥ 3). Headings inside a fence stay attached to the surrounding section. """ sections: list[str] = [] current: list[str] = [] in_fence = False fence_marker = "" # tracks the exact backtick run that opened the fence for line in body.splitlines(keepends=True): stripped = line.lstrip() if stripped.startswith("```"): # Count leading backticks; same count must close. marker = stripped[: len(stripped) - len(stripped.lstrip("`"))] if not in_fence: in_fence = True fence_marker = marker elif marker == fence_marker: in_fence = False fence_marker = "" if not in_fence and line.startswith("## ") and current: sections.append("".join(current)) current = [line] else: current.append(line) if current: sections.append("".join(current)) return sections def _utf8_safe_cut(encoded: bytes, cap: int) -> tuple[bytes, bytes]: """Cut `encoded` at byte index `cap`, snapping to the nearest UTF-8 codepoint boundary (and preferring a newline boundary inside that codepoint window). Returns (head, tail) where head + tail == encoded and head decodes cleanly to UTF-8. Hard guarantee: `len(head) <= cap` (the cap is the entire reason the function exists). If `cap` falls inside the very first multi-byte codepoint, head MAY be empty bytes — the caller is responsible for handling that (the cap is so small there's no way to fit any content). """ if cap <= 0: return b"", encoded if cap >= len(encoded): return encoded, b"" # Step BACKWARDS only — never walk forward past `cap`, which would violate the cap. end = cap while end > 0 and (encoded[end] & 0xC0) == 0x80: end -= 1 # Prefer a newline boundary within ~256 bytes of the cut, but only if doing so # leaves head meaningfully non-empty (don't collapse all the way to byte 0). nl = encoded.rfind(b"\n", max(0, end - 256), end) if nl > end // 2: end = nl return encoded[:end], encoded[end:] def _split_body_if_oversized(body: str, cap_bytes: int) -> tuple[str, str | None]: """If body exceeds cap, return (head, overflow). Otherwise (body, None). Two-stage split: 1. Walk `## ` section boundaries (skipping headings inside fenced code blocks); accumulate sections into `head` while we fit. 2. If `head` itself still exceeds cap (e.g. the H1 + intro is already huge), hard-cut at a UTF-8 codepoint boundary (preferring a newline) to guarantee output ≤ cap without dropping multibyte chars. The pointer note appended to `head` is accounted for in the effective cap. """ encoded = body.encode("utf-8") if len(encoded) <= cap_bytes: return body, None # Reserve room for the pointer note we'll append. effective_cap = cap_bytes - _POINTER_BYTES # _split_body_fence_aware returns sections WITH the leading `## ` already in place # for every section after the first; the first section is the pre-heading head. # Joining is just `head + "\n" + section` (no re-prepend of `## `). sections = _split_body_fence_aware(body) if body else [body] head = sections[0] if sections else body overflow_parts: list[str] = [] running = head for section in sections[1:]: candidate = running.rstrip("\n") + "\n\n" + section if len(candidate.encode("utf-8")) > effective_cap: overflow_parts.append(section) else: running = candidate # If running is still over cap (the head/H1 intro itself is too big), UTF-8-safe hard cut. if len(running.encode("utf-8")) > effective_cap: running_encoded = running.encode("utf-8") head_bytes, tail_bytes = _utf8_safe_cut(running_encoded, effective_cap) truncated_overflow = tail_bytes.decode("utf-8") overflow_parts.insert(0, truncated_overflow.lstrip("\n")) running = head_bytes.decode("utf-8") if not overflow_parts: # Edge case: hit the cap exactly. Fall back to a UTF-8-safe hard cut. head_bytes, tail_bytes = _utf8_safe_cut(encoded, effective_cap) running = head_bytes.decode("utf-8") overflow_parts = [tail_bytes.decode("utf-8")] return running + _POINTER, "\n".join(overflow_parts) # ── Adapter ────────────────────────────────────────────────────────────────── class CodexAdapter(HarnessAdapter): harness_id = "codex" # Effective cap for the body. Codex's hard limit is 8192 bytes of *injected* skill # content; we leave ~700 bytes of headroom for frontmatter + the pointer note we # add when splitting, so the emitted SKILL.md file stays under the limit. SKILL_BODY_CAP = 7400 AGENTS_MD_LINE_CAP = 150 def __init__( self, output_root: Path | None = None, repo_root: Path | None = None, ) -> None: super().__init__(output_root=output_root) # AGENTS.md is committed at the repo root, not under output_root. # Tests override repo_root to point at a temp dir. self.repo_root = repo_root if repo_root is not None else WORKTREE def emit_plugin(self, plugin: PluginSource) -> EmitResult: result = EmitResult() # Detect skill/command name collisions before emitting — Codex maps both into # `.codex/skills/__/`, so duplicates would silently overwrite. skill_names = {s.name for s in plugin.skills} command_collisions = {c.name for c in plugin.commands} & skill_names # Second-order collision: if a skill is literally named `__command`, the # `__command` suffix the adapter uses for collision-resolution would clash # with it. Detect and route the command-skill to `__cmd` instead. skill_suffix_conflicts: set[str] = set() for cmd_name in command_collisions: if f"{cmd_name}__command" in skill_names: skill_suffix_conflicts.add(cmd_name) result.warnings.append( f"plugin `{plugin.name}` has skill `{cmd_name}`, command `{cmd_name}`, " f"AND skill `{cmd_name}__command` — command-skill routed to " f"`{cmd_name}__cmd` to avoid second-order collision" ) for skill in plugin.skills: self._emit_skill(plugin, skill, result) for agent in plugin.agents: self._emit_agent(plugin, agent, result) # Codex deprecated prompts in favor of skills. Synthesize a skill from each command. for cmd in plugin.commands: collides = cmd.name in command_collisions suffix_conflict = cmd.name in skill_suffix_conflicts self._emit_command_as_skill( plugin, cmd, result, collides=collides, fallback_suffix=suffix_conflict, ) # Per-plugin Codex manifest for native marketplace install (reads source skills). self._emit_codex_plugin_manifest(plugin, result) return result def emit_global(self, plugins: list[PluginSource]) -> EmitResult: """Validate the committed root AGENTS.md and emit the Codex marketplace registry. Before May 2026 this method generated AGENTS.md from a template. AGENTS.md is now the canonical context file authored directly at the repo root (read natively by Codex / Cursor / OpenCode; Claude Code reads it via `CLAUDE.md`, a symlink to `AGENTS.md`; redirected from Gemini via `.gemini/settings.json`). The adapter validates that the committed file fits Codex's constraints (32 KiB hard cap, ~150-line table-of-contents convention) and writes `.agents/plugins/marketplace.json`. Validation reads from `self.repo_root` (defaulting to `WORKTREE`), not `self.output_root` — the canonical file lives in the source tree, and `--output-root` may point at a scratch directory for snapshot/round-trip tests. """ result = EmitResult() agents_md = self.repo_root / "AGENTS.md" if not agents_md.is_file(): result.warnings.append( "AGENTS.md is missing at the repo root — commit one. It's the canonical " "context file for Codex/Cursor/OpenCode; Claude Code reads it via " "`CLAUDE.md`, a symlink to `AGENTS.md`; Gemini CLI redirects via " "`gemini-extension.json` / `.gemini/settings.json`." ) else: content = agents_md.read_text(encoding="utf-8") line_count = len(content.splitlines()) file_bytes = len(content.encode("utf-8")) if file_bytes > 32 * 1024: result.warnings.append( f"AGENTS.md is {file_bytes} bytes; Codex hard cap is 32 KiB. " "Move detail into docs/." ) if line_count > self.AGENTS_MD_LINE_CAP: result.warnings.append( f"AGENTS.md is {line_count} lines (table-of-contents cap: " f"{self.AGENTS_MD_LINE_CAP}); move detail into docs/." ) # Codex marketplace registry: lists every local plugin so each installs # individually. Lives at the project-level `.agents/plugins/marketplace.json` # (auto-discovered when the repo is cwd; also addable via # `npx codex-marketplace add `). Entries point at the per-plugin bundles # emitted under `.codex/plugins//`. result.written.append( self.write( Path(".agents") / "plugins" / "marketplace.json", json.dumps(self._codex_marketplace(plugins), indent=2) + "\n", ) ) return result # ── Internals ────────────────────────────────────────────────────────── def _emit_skill(self, plugin: PluginSource, skill: SkillSource, result: EmitResult) -> None: skill_id = f"{plugin.name}__{skill.name}" skill_dir = Path(".codex") / "skills" / skill_id fm = _filter_frontmatter(skill.frontmatter, _CLAUDE_ONLY_SKILL_FIELDS) # Codex requires `name` to match directory exactly. fm["name"] = skill_id body = _rewrite_body_for_codex(skill.body).rstrip() + "\n" head, overflow = _split_body_if_oversized(body, self.SKILL_BODY_CAP) if overflow: # If source already has references/details.md, route overflow to _overflow.md # so the source mirror pass below doesn't clobber it. source_has_details = ( skill.references_dir is not None and (skill.references_dir / "details.md").is_file() ) overflow_rel = "_overflow.md" if source_has_details else "details.md" result.warnings.append( f"skill `{skill_id}` body exceeded {self.SKILL_BODY_CAP}B; " f"split into references/{overflow_rel}" ) result.written.append( self.write(skill_dir / "references" / overflow_rel, overflow.rstrip() + "\n") ) content = _frontmatter_block(fm) + "\n\n" + head result.written.append(self.write(skill_dir / "SKILL.md", content)) # Mirror any existing references/ assets — use binary copy so non-text assets # (PDFs, images, fonts) don't crash the run with UnicodeDecodeError. # The overflow vs source-details collision was handled above by routing # overflow to references/_overflow.md when source already has details.md. if skill.references_dir: for ref in sorted(skill.references_dir.rglob("*")): if not ref.is_file(): continue rel = ref.relative_to(skill.references_dir) result.written.append(self.mirror_file(ref, skill_dir / "references" / rel)) def _emit_agent(self, plugin: PluginSource, agent: AgentSource, result: EmitResult) -> None: agent_id = f"{plugin.name}__{agent.name}" rel = Path(".codex") / "agents" / f"{agent_id}.toml" # Map model alias; warn if the source model isn't one of the known aliases. model, warning = resolve_model("codex", agent.model) if warning: result.warnings.append(f"agent `{agent_id}`: {warning}") # Heuristic for sandbox_mode: # - If source frontmatter has NO `tools:` field at all -> agent has all tools by # default in Claude Code -> map to workspace-write (permissive). # - If source has `tools:` (even empty) and all listed tools are read-ish -> read-only. # - Otherwise (mix of read/write tools) -> workspace-write. ro_tools = {"Read", "Glob", "Grep", "WebFetch", "WebSearch"} has_tools_field = "tools" in agent.frontmatter if not has_tools_field: sandbox_mode = "workspace-write" # Claude default: unrestricted elif set(agent.tools).issubset(ro_tools): # Explicit allowlist constrained to read-only tools (or `tools: []` — even more restrictive) sandbox_mode = "read-only" else: sandbox_mode = "workspace-write" developer_instructions = _rewrite_body_for_codex(agent.body).strip() if not developer_instructions: developer_instructions = agent.description or f"{agent_id} subagent." lines = [ _toml_kv("name", agent_id), _toml_kv("description", agent.description or f"{agent.name} (from {plugin.name})"), _toml_kv("model", model), _toml_kv("sandbox_mode", sandbox_mode), _toml_kv("developer_instructions", developer_instructions), ] if agent.name in {"default", "worker", "explorer"}: result.warnings.append( f"agent `{agent.name}` collides with Codex built-in role; emitted as `{agent_id}` instead." ) result.written.append(self.write(rel, "\n".join(lines) + "\n")) def _emit_command_as_skill( self, plugin: PluginSource, cmd, result: EmitResult, *, collides: bool = False, fallback_suffix: bool = False, ) -> None: """Convert a Claude command into a Codex skill. If `collides=True` (a skill in the same plugin has the same name), the command- derived skill is emitted under a namespaced suffix `____command` so it doesn't silently overwrite the real skill. If `fallback_suffix=True` (a skill named `__command` ALSO exists — the second-order collision case), use `__cmd` instead. """ suffix = ("__cmd" if fallback_suffix else "__command") if collides else "" skill_id = f"{plugin.name}__{cmd.name}{suffix}" if collides: result.warnings.append( f"command `{cmd.name}` collides with skill `{cmd.name}` in `{plugin.name}`; " f"emitted command-skill as `{skill_id}` to avoid overwrite" ) skill_dir = Path(".codex") / "skills" / skill_id fm = { "name": skill_id, "description": cmd.description or f"Command: {cmd.name} (from {plugin.name})", } body = _rewrite_body_for_codex(cmd.body).rstrip() + "\n" head, overflow = _split_body_if_oversized(body, self.SKILL_BODY_CAP) if overflow: result.warnings.append( f"command-skill `{skill_id}` body exceeded {self.SKILL_BODY_CAP}B; split into references/details.md" ) result.written.append( self.write(skill_dir / "references" / "details.md", overflow.rstrip() + "\n") ) if cmd.argument_hint: fm["metadata"] = {"argument-hint": cmd.argument_hint} content = _frontmatter_block(fm) + "\n\n" + head result.written.append(self.write(skill_dir / "SKILL.md", content)) # ── Marketplace manifests (native install; reads source skills) ────────── def _emit_codex_plugin_manifest(self, plugin: PluginSource, result: EmitResult) -> None: """Write `plugins//.codex-plugin/plugin.json`, next to `.claude-plugin/`. The Codex marketplace installs the SOURCE plugin directory directly (`source.path: ./plugins/`), so no transformed skill tree is committed — Codex reads `SKILL.md` natively from `plugins//skills/` (`skills: "./skills/"`), exactly as the Cursor marketplace does. This keeps the repo lean: only the tiny manifest is committed, not a duplicated skill tree. Skills above Codex's 8 KB injection cap are truncated by Codex at load; the transformed, cap-split copies under `.codex/skills/` (gitignored) remain for the `~/.codex/skills` symlink recipe. """ manifest_path = Path("plugins") / plugin.name / ".codex-plugin" / "plugin.json" result.written.append( self.write( manifest_path, json.dumps(self._codex_plugin_manifest(plugin), indent=2) + "\n", ) ) def _codex_plugin_manifest(self, plugin: PluginSource) -> dict: """Per-plugin `.codex-plugin/plugin.json` (mirrors the superpowers shape).""" pj = plugin.plugin_json or {} manifest: dict = { "name": plugin.name, "version": plugin.version, "description": plugin.description or "", "skills": "./skills/", } if pj.get("author"): manifest["author"] = pj["author"] for key in ("homepage", "repository", "license", "keywords"): if pj.get(key): manifest[key] = pj[key] desc = plugin.description or plugin.name if len(desc) > 120: # Cut on a word boundary so committed manifests never render mid-word. desc = desc[:117].rsplit(" ", 1)[0].rstrip(" ,.;:!?-") + "…" manifest["interface"] = { "displayName": plugin.name.replace("-", " ").title(), "shortDescription": desc, "category": pj.get("category") or "Coding", } return manifest def _codex_marketplace(self, plugins: list[PluginSource]) -> dict: """`.agents/plugins/marketplace.json` — entries install the source plugin dirs. Schema per openai/codex `core-plugins`: each entry needs `name`, a `source` (local variant `{"source": "local", "path": ...}`), a `policy` (`installation`/`authentication`), and `category`. Entries install the source plugin dir directly; per-plugin display metadata comes from each plugin's own `.codex-plugin/plugin.json`, so we don't repeat version/description here. """ root = self._read_marketplace_root() entries = [] for p in plugins: pj = p.plugin_json or {} entries.append( { "name": p.name, "source": {"source": "local", "path": f"./plugins/{p.name}"}, "policy": {"installation": "AVAILABLE", "authentication": "ON_USE"}, "category": pj.get("category") or "Coding", } ) return { "name": root.get("name", "claude-code-workflows"), "plugins": entries, } def _read_marketplace_root(self) -> dict: """Read the source `.claude-plugin/marketplace.json` for name/owner/metadata.""" path = self.repo_root / ".claude-plugin" / "marketplace.json" if not path.is_file(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): return {} # NOTE: AGENTS.md is no longer generated by this adapter — it's committed at the # repo root as the canonical cross-harness context file. `emit_global` above # validates the committed file fits Codex's constraints (32 KiB cap, ~150-line # table-of-contents convention).