fix: keep Codex MCP identity outside native settings

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Codex MCP entries no longer emit the unsupported `id` setting; reinstalling corrects recorded APM-owned entries while preserving registry identity, user settings, and ownership safeguards. — by @Pybsama (#3157)
+
 ## [0.33.0] - 2026-10-02
 
 ### Added

docs/src/content/docs/consumer/install-mcp-servers.md

@@ -287,6 +287,20 @@ apm install --target codex --mcp local-dev --url http://localhost:3000/mcp
 
 This writes the endpoint to the Codex MCP configuration.
 
+Codex MCP entries use its [native configuration schema](https://developers.openai.com/codex/config-schema.json).
+APM keeps registry IDs in TOML comments on `command`, `url`, or the whole
+inline entry for duplicate detection, not in an unsupported `id` setting. Keep those comments when editing generated entries.
+After upgrading APM, run `apm install --only mcp --target codex` to remove old
+`id` settings from entries recorded as APM-owned in `apm.lock.yaml`; add
+`--global` for user-scope configuration. Other settings and user-owned entries
+are preserved. Inline MCP containers being written, and owned inline entries
+being repaired, may expand into regular TOML tables; their values and comments
+are retained. A legacy lockfile without per-target ownership can adopt only
+self-defined entries that exactly match its saved baseline. Missing ownership
+records, edited legacy entries, and unrelated user entries require manual
+review; regeneration does not clean every `id` in the file or change Codex's
+project trust settings.
+
 `--transport` is inferred when omitted: a `--url` implies a remote
 transport, a post-`--` command implies `stdio`. The mutually-exclusive
 combinations (`--url` plus stdio command, `--header` without `--url`,

packages/apm-guide/.apm/skills/apm-usage/commands.md

@@ -107,6 +107,12 @@ lock state may skip the write, so repeating that install is not a migration.
 
 ### Registry MCP runtime variables
 
+Codex keeps APM registry identity in TOML comments instead of an unsupported
+native `id` setting. After upgrading, `apm install --only mcp --target codex`
+repairs recorded APM-owned entries; add `--global` for user scope. Unowned or
+edited legacy entries need manual review. Keep generated identity comments;
+inline containers being written may expand while preserving values and comments.
+
 For registry MCP runtime variables, `apm install` prompts once for a required
 non-secret default and accepts an override; secret defaults remain hidden.
 Non-secret values resolve every matching `{variable}` launcher reference,

src/apm_cli/adapters/client/codex.py

@@ -1,5 +1,6 @@
 """OpenAI Codex CLI implementation of MCP client adapter."""
 
+import json
 import logging
 import os
 import re
@@ -8,6 +9,7 @@
 
 import tomlkit
 from tomlkit.exceptions import TOMLKitError
+from tomlkit.items import InlineTable, Item, String, Table
 
 from ...registry.client import SimpleRegistryClient
 from ...registry.integration import RegistryIntegration
@@ -31,6 +33,7 @@
 # ``_resolve_variable_placeholders`` untouched.
 # See https://learn.chatgpt.com/docs/extend/mcp?surface=cli
 _CODEX_BEARER_HEADER = "authorization"
+_CODEX_REGISTRY_ID_COMMENT = "# apm-registry-id: "
 # Reuse the canonical placeholder syntax so the two spellings APM accepts
 # (``${VAR}`` and ``${env:VAR}``) cannot drift apart here. The auth scheme is
 # case-insensitive per RFC 7235; Codex always writes it as ``Bearer``.
@@ -102,29 +105,132 @@ def get_config_path(self) -> str:
         """
         return str(self._get_codex_dir() / "config.toml")
 
+    @staticmethod
+    def _registry_id_item(server_config: object) -> Item | None:
+        """Locate a comment that survives standard, dotted and inline TOML layouts.
+
+        Inline values cannot contain comments, so annotate the whole inline
+        entry. Other layouts expose command/url scalar trivia even when their
+        parent tables are implicit or split across out-of-order fragments.
+        """
+        if isinstance(server_config, InlineTable):
+            return server_config
+        if isinstance(server_config, dict):
+            for key in ("command", "url"):
+                value = server_config.get(key)
+                if isinstance(value, String):
+                    return value
+        return None
+
+    @classmethod
+    def get_registry_id(cls, server_config: dict) -> str | None:
+        """Read APM identity metadata without exposing it as a Codex setting.
+
+        Keep reading legacy native IDs for conflict detection; their presence
+        alone is not ownership evidence and never authorizes a migration.
+        """
+        legacy_id = server_config.get("id")
+        if isinstance(legacy_id, str) and legacy_id:
+            return legacy_id
+        item = cls._registry_id_item(server_config)
+        if item is None:
+            return None
+        comment = item.trivia.comment
+        if not comment.startswith(_CODEX_REGISTRY_ID_COMMENT):
+            return None
+        try:
+            registry_id, _ = json.JSONDecoder().raw_decode(
+                comment[len(_CODEX_REGISTRY_ID_COMMENT) :]
+            )
+        except ValueError:
+            return None
+        return registry_id if isinstance(registry_id, str) and registry_id else None
+
+    @classmethod
+    def _annotate_registry_id(cls, server_config: dict, registry_id: str) -> None:
+        """Keep UUID matching metadata in an escaped, round-trip TOML comment."""
+        item = cls._registry_id_item(server_config)
+        if item is None:
+            raise ValueError("Codex MCP server must have a string command or url")
+        original_comment = item.trivia.comment
+        suffix = f" {original_comment}" if original_comment else ""
+        item.comment(_CODEX_REGISTRY_ID_COMMENT[2:] + json.dumps(registry_id) + suffix)
+
+    @staticmethod
+    def _expand_inline_table(entry: InlineTable) -> Table:
+        """Expand an inline container without discarding its values or comments.
+
+        Nested inline comments are invalid TOML, and deleting a middle key can
+        leave stray separators in tomlkit. Expand only containers being written
+        or owned entries being repaired; keep unrelated child items intact.
+        """
+        table = tomlkit.table(is_super_table=False)
+        for key, value in entry.items():
+            table.add(key, value)
+        if entry.trivia.comment:
+            table.comment(entry.trivia.comment)
+        return table
+
+    def _write_config(self, config: dict) -> None:
+        """Validate serialized TOML before atomically replacing the native file."""
+        serialized = tomlkit.dumps(config)
+        tomlkit.parse(serialized)
+        config_path = Path(self.get_config_path())
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+        atomic_write_text(config_path, serialized, new_file_mode=0o600)
+        _log.debug("Codex config written to %s", config_path)
+
+    def migrate_legacy_managed_servers(self, managed_names: set[str]) -> set[str]:
+        """Remove unsupported IDs only from entries with recorded APM ownership."""
+        if not managed_names:
+            return set()
+        config = self.get_current_config()
+        if config is None:
+            return set()
+        servers = config.get("mcp_servers", {})
+        migrated: set[str] = set()
+        for name in sorted(managed_names):
+            config_key = self._determine_config_key(name, None)
+            entry = servers.get(config_key)
+            if not isinstance(entry, dict) or "id" not in entry:
+                continue
+            registry_id = entry["id"]
+            if not isinstance(registry_id, str):
+                continue
+            if isinstance(servers, InlineTable):
+                servers = self._expand_inline_table(servers)
+                config["mcp_servers"] = servers
+            if isinstance(entry, InlineTable):
+                entry = self._expand_inline_table(entry)
+                servers[config_key] = entry
+            if registry_id:
+                self._annotate_registry_id(entry, registry_id)
+            del entry["id"]
+            migrated.add(name)
+        if migrated:
+            self._write_config(config)
+        return migrated
+
     def update_config(self, config_updates):
         """Update the Codex CLI MCP configuration.
 
         Args:
             config_updates (dict): Configuration updates to apply.
         """
-        config_path = Path(self.get_config_path())
         current_config = self.get_current_config()
         if current_config is None:
             return False
 
         # Ensure mcp_servers section exists
         if "mcp_servers" not in current_config:
             current_config["mcp_servers"] = {}
+        elif isinstance(current_config["mcp_servers"], InlineTable):
+            current_config["mcp_servers"] = self._expand_inline_table(current_config["mcp_servers"])
 
         # Apply updates to mcp_servers section
         current_config["mcp_servers"].update(config_updates)
 
-        # Ensure directory exists
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-
-        atomic_write_text(config_path, tomlkit.dumps(current_config), new_file_mode=0o600)
-        _log.debug("Codex config written to %s", config_path)
+        self._write_config(current_config)
         return True
 
     def get_current_config(self):
@@ -199,6 +305,12 @@ def configure_mcp_server(
             if server_config is None:
                 return False
 
+            # Registry identity is APM metadata, not a supported native field.
+            registry_id = server_info.get("id")
+            if isinstance(registry_id, str) and registry_id:
+                server_config = tomlkit.item(server_config)
+                self._annotate_registry_id(server_config, registry_id)
+
             # Update configuration using the chosen key
             if not self.update_config({config_key: server_config}):
                 return False
@@ -227,12 +339,11 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No
         if runtime_vars is None:
             runtime_vars = {}
 
-        # Default configuration structure with registry ID for conflict detection
+        # Only fields accepted by Codex belong in the native server mapping.
         config = {
             "command": "unknown",
             "args": [],
             "env": {},
-            "id": server_info.get("id", ""),  # Add registry UUID for conflict detection
         }
 
         # Self-defined stdio deps carry raw command/args. Route ``env`` and
@@ -313,10 +424,7 @@ def _process_stdio_arg(arg):
                 )
                 return None
 
-            remote_config = {
-                "url": remote_url,
-                "id": server_info.get("id", ""),
-            }
+            remote_config = {"url": remote_url}
             http_headers: dict[str, str] = {}
             env_http_headers: dict[str, str] = {}
             bearer_token_env_var = ""

src/apm_cli/core/conflict_detector.py

@@ -43,11 +43,14 @@ def check_server_exists(
 
                 # Check if any existing server has the same UUID
                 for existing_name, existing_config in existing_servers.items():  # noqa: B007
-                    if (
-                        isinstance(existing_config, dict)
-                        and existing_config.get("id") == server_uuid
-                    ):
-                        return True
+                    if isinstance(existing_config, dict):
+                        existing_id = existing_config.get("id")
+                        if self.adapter.target_name == "codex":
+                            from ..adapters.client.codex import CodexClientAdapter
+
+                            existing_id = CodexClientAdapter.get_registry_id(existing_config)
+                        if existing_id == server_uuid:
+                            return True
             elif server_info:
                 # MCP Registry v0.1 entries carry no stable id; match by name
                 # without further registry lookups.

src/apm_cli/install/mcp/ownership.py

@@ -105,6 +105,20 @@ def adopt_legacy_mcp_target_servers(
                     exc_info=True,
                 )
                 continue
-            if any(existing.get(name) == expected for existing in existing_configs):
+            native_baselines = [expected]
+            candidates = [existing.get(name) for existing in existing_configs]
+            if runtime == "codex":
+                from apm_cli.adapters.client.codex import CodexClientAdapter
+
+                # Earlier Codex renders added an empty id to self-defined
+                # entries. A nonempty ID, including its new comment encoding,
+                # still disqualifies a native entry from this exact baseline.
+                native_baselines.append({**expected, "id": ""})
+                candidates = [
+                    entry
+                    for entry in candidates
+                    if isinstance(entry, dict) and not CodexClientAdapter.get_registry_id(entry)
+                ]
+            if any(entry in native_baselines for entry in candidates):
                 adopted.setdefault(runtime, set()).add(name)
     return adopted

src/apm_cli/integration/mcp_integrator_install.py

@@ -1233,6 +1233,13 @@ def run_mcp_install(  # noqa: PLR0913
             else:
                 managed_target_servers[target].intersection_update(current_names)
 
+    if "codex" in target_runtimes and managed_target_servers is not None:
+        from apm_cli.adapters.client.codex import CodexClientAdapter
+
+        CodexClientAdapter(
+            project_root=project_root, user_scope=user_scope
+        ).migrate_legacy_managed_servers(managed_target_servers.get("codex", set()))
+
     # Use the new registry operations module for better server detection
     configured_count = 0
 

src/apm_cli/registry/operations.py

@@ -153,11 +153,13 @@ def _get_installed_server_ids(
                                     installed_ids.add(server_id)
 
                     elif runtime == "codex":
+                        from ..adapters.client.codex import CodexClientAdapter
+
                         # Codex stores servers as mcp_servers.{name} sections in config.toml
                         mcp_servers = config.get("mcp_servers", {})
                         for server_name, server_config in mcp_servers.items():  # noqa: B007
                             if isinstance(server_config, dict):
-                                server_id = server_config.get("id")
+                                server_id = CodexClientAdapter.get_registry_id(server_config)
                                 if server_id:
                                     installed_ids.add(server_id)
 

tests/integration/test_codex_mcp_schema_lifecycle.py

@@ -0,0 +1,69 @@
+"""Real CLI engine regeneration keeps Codex MCP ownership and user settings."""
+
+import os
+from pathlib import Path
+
+import pytest
+import tomlkit
+
+from apm_cli.utils.yaml_io import load_yaml
+from tests.utils.apm_lifecycle_runner import ApmLifecycleRunner
+from tests.utils.isolated_apm_environment import IsolatedApmEnvironment
+
+pytestmark = pytest.mark.e2e
+
+
+@pytest.mark.parametrize("user_scope", [False, True])
+def test_codex_install_repairs_owned_legacy_ids_and_converges(
+    tmp_path: Path, apm_engine_command: tuple[str, ...], user_scope: bool
+) -> None:
+    """The same install command repairs a prior release's output in either scope."""
+    isolated = IsolatedApmEnvironment.create(tmp_path / "scenario", base_env=dict(os.environ))
+    environment = isolated.subprocess_env()
+    environment["CODEX_HOME"] = str(isolated.home / "custom-codex")
+    project = isolated.config_root if user_scope else isolated.work_root
+    project.joinpath("apm.yml").write_text(
+        "name: codex-schema-test\nversion: 1.0.0\ntargets: [codex]\n"
+        "dependencies:\n  mcp:\n"
+        "    - name: managed-stdio\n      registry: false\n      transport: stdio\n"
+        "      command: echo\n      args: [hello]\n"
+        "    - name: managed-http\n      registry: false\n      transport: http\n"
+        "      url: https://example.test/mcp\n",
+        encoding="utf-8",
+    )
+    config_path = (
+        Path(environment["CODEX_HOME"]) if user_scope else project / ".codex"
+    ) / "config.toml"
+    config_path.parent.mkdir(parents=True, exist_ok=True)
+    unrelated = '# personal settings\nmodel = "gpt-5"\n[mcp_servers.personal] # user note\ncommand = "custom"\nid = "personal-id"\n'
+    config_path.write_text(unrelated, encoding="utf-8")
+    args = ("install", "--only", "mcp", "--target", "codex")
+    if user_scope:
+        args += ("--global",)
+    runner = ApmLifecycleRunner(apm_engine_command)
+
+    result = runner.run(args, cwd=isolated.work_root, env=environment)
+    assert result.returncode == 0, result.stdout + result.stderr
+    native = tomlkit.parse(config_path.read_text(encoding="utf-8"))
+    for name in ("managed-stdio", "managed-http"):
+        assert "id" not in native["mcp_servers"][name]
+        native["mcp_servers"][name]["id"] = ""
+    # Seed precisely the obsolete native field; keep the real install's ledger.
+    config_path.write_text(tomlkit.dumps(native), encoding="utf-8")
+    lockfile = project / "apm.lock.yaml"
+    ownership = load_yaml(lockfile)["mcp_target_servers"]
+    assert ownership == {"codex": ["managed-http", "managed-stdio"]}
+
+    result = runner.run(args, cwd=isolated.work_root, env=environment)
+    assert result.returncode == 0, result.stdout + result.stderr
+    corrected = config_path.read_text(encoding="utf-8")
+    native = tomlkit.parse(corrected)
+    assert unrelated in corrected
+    for name in ("managed-stdio", "managed-http"):
+        assert "id" not in native["mcp_servers"][name]
+    assert load_yaml(lockfile)["mcp_target_servers"] == ownership
+
+    result = runner.run(args, cwd=isolated.work_root, env=environment)
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert config_path.read_text(encoding="utf-8") == corrected
+    assert load_yaml(lockfile)["mcp_target_servers"] == ownership