diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c1a995f25..084cbd41dc 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/src/content/docs/consumer/install-mcp-servers.md b/docs/src/content/docs/consumer/install-mcp-servers.md index dee9e9f0dc..156e4704b3 100644 --- a/docs/src/content/docs/consumer/install-mcp-servers.md +++ b/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`, diff --git a/packages/apm-guide/.apm/skills/apm-usage/commands.md b/packages/apm-guide/.apm/skills/apm-usage/commands.md index f8055348c3..0ba42ca48f 100644 --- a/packages/apm-guide/.apm/skills/apm-usage/commands.md +++ b/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, diff --git a/src/apm_cli/adapters/client/codex.py b/src/apm_cli/adapters/client/codex.py index dbf74d3b53..d0d2ac62ea 100644 --- a/src/apm_cli/adapters/client/codex.py +++ b/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,13 +105,118 @@ 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 @@ -116,15 +224,13 @@ def update_config(self, config_updates): # 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 = "" diff --git a/src/apm_cli/core/conflict_detector.py b/src/apm_cli/core/conflict_detector.py index 33fe597f2c..d8f5774467 100644 --- a/src/apm_cli/core/conflict_detector.py +++ b/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. diff --git a/src/apm_cli/install/mcp/ownership.py b/src/apm_cli/install/mcp/ownership.py index 0f53cf922d..cd8ce6404a 100644 --- a/src/apm_cli/install/mcp/ownership.py +++ b/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 diff --git a/src/apm_cli/integration/mcp_integrator_install.py b/src/apm_cli/integration/mcp_integrator_install.py index 4a18c5d789..8ed4b2026d 100644 --- a/src/apm_cli/integration/mcp_integrator_install.py +++ b/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 diff --git a/src/apm_cli/registry/operations.py b/src/apm_cli/registry/operations.py index ce4f3664fc..ac0c5bd790 100644 --- a/src/apm_cli/registry/operations.py +++ b/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) diff --git a/tests/integration/test_codex_mcp_schema_lifecycle.py b/tests/integration/test_codex_mcp_schema_lifecycle.py new file mode 100644 index 0000000000..50c99319e2 --- /dev/null +++ b/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 diff --git a/tests/unit/test_codex_adapter_compatibility.py b/tests/unit/test_codex_adapter_compatibility.py index 22f40d3074..cb5e87e5c7 100644 --- a/tests/unit/test_codex_adapter_compatibility.py +++ b/tests/unit/test_codex_adapter_compatibility.py @@ -498,7 +498,7 @@ def test_env_vars_added_to_config(self, tmp_path: Path) -> None: cfg = adapter._format_server_config({"packages": [pkg]}) assert cfg["env"].get("MY_TOKEN") == "env-value" - def test_id_added_from_server_info(self, tmp_path: Path) -> None: + def test_registry_id_is_not_a_native_setting(self, tmp_path: Path) -> None: adapter = _make_adapter(project_root=tmp_path) pkg = { "registry_name": "npm", @@ -509,7 +509,7 @@ def test_id_added_from_server_info(self, tmp_path: Path) -> None: "environment_variables": [], } cfg = adapter._format_server_config({"packages": [pkg], "id": "my-uuid-123"}) - assert cfg["id"] == "my-uuid-123" + assert "id" not in cfg # --------------------------------------------------------------------------- diff --git a/tests/unit/test_codex_adapter_phase3.py b/tests/unit/test_codex_adapter_phase3.py index a18338d53b..0dba19223c 100644 --- a/tests/unit/test_codex_adapter_phase3.py +++ b/tests/unit/test_codex_adapter_phase3.py @@ -590,7 +590,7 @@ def test_env_vars_added_to_config(self, tmp_path: Path) -> None: cfg = adapter._format_server_config({"packages": [pkg]}) assert cfg["env"].get("MY_TOKEN") == "env-value" - def test_id_added_from_server_info(self, tmp_path: Path) -> None: + def test_registry_id_is_not_a_native_setting(self, tmp_path: Path) -> None: adapter = _make_adapter(project_root=tmp_path) pkg = { "registry_name": "npm", @@ -601,7 +601,7 @@ def test_id_added_from_server_info(self, tmp_path: Path) -> None: "environment_variables": [], } cfg = adapter._format_server_config({"packages": [pkg], "id": "my-uuid-123"}) - assert cfg["id"] == "my-uuid-123" + assert "id" not in cfg # --------------------------------------------------------------------------- diff --git a/tests/unit/test_codex_mcp_native_schema.py b/tests/unit/test_codex_mcp_native_schema.py new file mode 100644 index 0000000000..e6781bb1dd --- /dev/null +++ b/tests/unit/test_codex_mcp_native_schema.py @@ -0,0 +1,540 @@ +"""Codex native output must not expose APM's registry identity as a setting.""" + +from pathlib import Path + +import pytest +import tomlkit + +from apm_cli.adapters.client.codex import CodexClientAdapter +from apm_cli.core.conflict_detector import MCPConflictDetector +from apm_cli.core.safe_installer import SafeMCPInstaller +from apm_cli.install.mcp.ownership import resolve_mcp_target_servers +from apm_cli.integration.mcp_integrator import MCPIntegrator +from apm_cli.models.dependency.mcp import MCPDependency +from apm_cli.registry.operations import MCPServerOperations + +pytestmark = pytest.mark.component + + +@pytest.fixture(params=["stdio", "http"]) +def dependency(request: pytest.FixtureRequest) -> MCPDependency: + """Use transports from #3087 without registry or server network access.""" + if request.param == "stdio": + return MCPDependency.from_dict( + { + "name": "managed", + "registry": False, + "transport": "stdio", + "command": "echo", + "args": ["hello"], + } + ) + return MCPDependency.from_dict( + { + "name": "managed", + "registry": False, + "transport": "http", + "url": "https://example.test/mcp", + } + ) + + +def test_native_output_preserves_uuid_conflicts_after_reparse( + tmp_path: Path, dependency: MCPDependency +) -> None: + """Removing native id must not cause duplicate installations under aliases.""" + adapter = CodexClientAdapter(project_root=tmp_path) + info = MCPIntegrator._build_self_defined_info(dependency) + info["id"] = 'registry-uuid"\\\n# still data' + assert adapter.configure_mcp_server("managed", server_info_cache={"managed": info}) + native_path = Path(adapter.get_config_path()) + native = tomlkit.parse(native_path.read_text(encoding="utf-8")) + assert "id" not in native["mcp_servers"]["managed"] + + # Another target write must preserve the first entry's identity annotation. + assert adapter.update_config({"other": {"command": "echo", "args": []}}) + fresh_adapter = CodexClientAdapter(project_root=tmp_path) + detector = MCPConflictDetector(fresh_adapter) + assert detector.check_server_exists("alias", server_info=info) + assert not detector.check_server_exists("different", server_info={"id": "different-id"}) + before = native_path.read_bytes() + summary = SafeMCPInstaller("codex", project_root=tmp_path).install_servers( + ["alias"], server_info_cache={"alias": info} + ) + assert summary.skipped == [{"server": "alias", "reason": "already configured"}] + assert not summary.installed and not summary.failed + assert native_path.read_bytes() == before + operations = MCPServerOperations() + assert ( + operations.check_servers_needing_installation( + ["codex"], ["alias"], project_root=tmp_path, server_info_cache={"alias": info} + ) + == [] + ) + + +@pytest.mark.parametrize("owned", [True, False]) +def test_install_corrects_only_recorded_managed_entries( + tmp_path: Path, dependency: MCPDependency, owned: bool +) -> None: + """An unchanged manifest still repairs owned legacy output, not user entries.""" + adapter = CodexClientAdapter(project_root=tmp_path) + # Write the old output independently of the formatter being repaired. + legacy = ( + 'command = "echo"\nargs = ["hello"]\nid = "legacy-uuid"\n' + if dependency.command + else 'url = "https://example.test/mcp"\nid = "legacy-uuid"\n' + ) + user_settings = '\n[mcp_servers.personal] # user note\ncommand = "custom"\nid = "personal-id"\n' + native_path = Path(adapter.get_config_path()) + native_path.parent.mkdir() + original = ( + '# user config\nmodel = "gpt-5"\n[mcp_servers.managed] # keep note\n' + + legacy + + user_settings + ) + native_path.write_text(original, encoding="utf-8") + owners = {"codex": {"managed"}} if owned else {} + + MCPIntegrator.install( + [dependency], + project_root=tmp_path, + explicit_target="codex", + stored_mcp_configs={"managed": dependency.to_dict()}, + managed_target_servers=owners, + ) + + written = native_path.read_text(encoding="utf-8") + native = tomlkit.parse(written) + assert ("id" not in native["mcp_servers"]["managed"]) is owned + assert native["mcp_servers"]["personal"]["id"] == "personal-id" + assert user_settings in written + assert native["model"] == "gpt-5" + assert "# keep note" in written + assert owners == ({"codex": {"managed"}} if owned else {}) + if owned: + assert MCPConflictDetector(adapter).check_server_exists( + "alias", server_info={"id": "legacy-uuid"} + ) + MCPIntegrator.install( + [dependency], + project_root=tmp_path, + explicit_target="codex", + stored_mcp_configs={"managed": dependency.to_dict()}, + managed_target_servers=owners, + ) + assert native_path.read_text(encoding="utf-8") == written + else: + assert written == original + + +@pytest.mark.parametrize("user_edited", [False, True]) +def test_legacy_ownership_adoption_still_requires_exact_baseline( + tmp_path: Path, dependency: MCPDependency, user_edited: bool +) -> None: + """Without per-target ownership, id alone cannot claim a user's entry.""" + adapter = CodexClientAdapter(project_root=tmp_path) + legacy = ( + {"command": "echo", "args": ["hello"], "env": {}, "id": ""} + if dependency.command + else {"url": "https://example.test/mcp", "id": ""} + ) + if user_edited: + legacy["startup_timeout_sec"] = 90 + native_path = Path(adapter.get_config_path()) + native_path.parent.mkdir() + native_path.write_text(tomlkit.dumps({"mcp_servers": {"managed": legacy}}), encoding="utf-8") + ownership = resolve_mcp_target_servers( + recorded_target_servers={}, + ownership_present=False, + server_names={"managed"}, + stored_configs={"managed": dependency.to_dict()}, + project_root=tmp_path, + user_scope=False, + ) + assert ownership == ({} if user_edited else {"codex": {"managed"}}) + + +def _write_out_of_order_config( + adapter: CodexClientAdapter, + dependency: MCPDependency, + *, + legacy_id: bool, + out_of_order: bool = True, +) -> str: + """Place another server before the first server's subtable, valid TOML.""" + native_path = Path(adapter.get_config_path()) + native_path.parent.mkdir() + header = "[mcp_servers.managed] # keep note\n" + if legacy_id: + header += 'id = "registry-uuid"\n' + transport = ( + 'command = "echo"\nargs = ["hello"]\n' + if dependency.command + else 'url = "https://example.test/mcp"\n' + ) + if not legacy_id: + transport = transport.replace("\n", ' # apm-registry-id: "registry-uuid"\n', 1) + subtable = "env" if dependency.command else "http_headers" + personal = '\n[mcp_servers.personal] # personal note\ncommand = "custom"\nid = "personal-id"\n' + nested = f'\n[mcp_servers.managed.{subtable}] # late subtable\nTEST = "value"\n' + raw = header + transport + (personal + nested if out_of_order else nested + personal) + native_path.write_text(raw, encoding="utf-8") + return raw + + +def test_out_of_order_subtable_preserves_uuid_conflicts( + tmp_path: Path, dependency: MCPDependency +) -> None: + """Reordering valid TOML subtables must not cause alias installations.""" + adapter = CodexClientAdapter(project_root=tmp_path) + original = _write_out_of_order_config(adapter, dependency, legacy_id=False) + info = {**MCPIntegrator._build_self_defined_info(dependency), "id": "registry-uuid"} + assert MCPConflictDetector(adapter).check_server_exists("alias", server_info=info) + assert ( + MCPServerOperations().check_servers_needing_installation( + ["codex"], ["alias"], project_root=tmp_path, server_info_cache={"alias": info} + ) + == [] + ) + summary = SafeMCPInstaller("codex", project_root=tmp_path).install_servers( + ["alias"], server_info_cache={"alias": info} + ) + assert summary.skipped == [{"server": "alias", "reason": "already configured"}] + assert not summary.installed and not summary.failed + assert Path(adapter.get_config_path()).read_text(encoding="utf-8") == original + + +@pytest.mark.parametrize("owned", [True, False]) +def test_out_of_order_subtable_migrates_only_owned_legacy_entry( + tmp_path: Path, dependency: MCPDependency, owned: bool +) -> None: + """ID removal must preserve an owned entry split across concrete tables.""" + adapter = CodexClientAdapter(project_root=tmp_path) + original = _write_out_of_order_config(adapter, dependency, legacy_id=True) + migrated = adapter.migrate_legacy_managed_servers({"managed"} if owned else set()) + assert migrated == ({"managed"} if owned else set()) + written = Path(adapter.get_config_path()).read_text(encoding="utf-8") + expected = original + if owned: + expected = expected.replace('id = "registry-uuid"\n', "") + anchor = 'command = "echo"' if dependency.command else 'url = "https://example.test/mcp"' + expected = expected.replace(anchor, anchor + ' # apm-registry-id: "registry-uuid"') + assert written == expected + assert MCPConflictDetector(adapter).check_server_exists( + "alias", server_info={"id": "registry-uuid"} + ) + assert adapter.migrate_legacy_managed_servers({"managed"} if owned else set()) == set() + assert Path(adapter.get_config_path()).read_text(encoding="utf-8") == written + + +@pytest.mark.parametrize("out_of_order", [False, True]) +def test_legacy_adoption_does_not_claim_registry_annotated_entry( + tmp_path: Path, dependency: MCPDependency, out_of_order: bool +) -> None: + """A nonempty UUID still disqualifies a self-defined legacy baseline match.""" + if dependency.command: + dependency.env = {"TEST": "value"} + else: + dependency.headers = {"TEST": "value"} + adapter = CodexClientAdapter(project_root=tmp_path) + original = _write_out_of_order_config( + adapter, dependency, legacy_id=False, out_of_order=out_of_order + ) + ownership = resolve_mcp_target_servers( + recorded_target_servers={}, + ownership_present=False, + server_names={"managed"}, + stored_configs={"managed": dependency.to_dict()}, + project_root=tmp_path, + user_scope=False, + ) + assert ownership == {} + assert Path(adapter.get_config_path()).read_text(encoding="utf-8") == original + + +def _write_inline_config( + adapter: CodexClientAdapter, + dependency: MCPDependency, + *, + registry_id: str | None, + comment: str = "# keep my comment", +) -> str: + """Write an inline server without converting it to a standard table.""" + fields = ( + 'command="echo", args=["hello"], env={}' + if dependency.command + else 'url="https://example.test/mcp"' + ) + if registry_id is not None: + fields += f', id="{registry_id}"' + original = ( + f"[mcp_servers]\nmanaged = {{{fields}}} {comment}\n" + 'personal = {command="custom", id="personal-id"} # user comment\n' + ) + path = Path(adapter.get_config_path()) + path.parent.mkdir() + path.write_text(original, encoding="utf-8") + return original + + +def test_inline_legacy_baseline_is_corrected_on_install( + tmp_path: Path, dependency: MCPDependency +) -> None: + """Exact legacy inline baselines must not retain the unsupported empty id.""" + adapter = CodexClientAdapter(project_root=tmp_path) + original = _write_inline_config(adapter, dependency, registry_id="") + stored_configs = {"managed": dependency.to_dict()} + ownership = resolve_mcp_target_servers( + recorded_target_servers={}, + ownership_present=False, + server_names={"managed"}, + stored_configs=stored_configs, + project_root=tmp_path, + user_scope=False, + ) + assert ownership == {"codex": {"managed"}} + MCPIntegrator.install( + [dependency], + project_root=tmp_path, + explicit_target="codex", + stored_mcp_configs=stored_configs, + managed_target_servers=ownership, + ) + written = Path(adapter.get_config_path()).read_text(encoding="utf-8") + native = tomlkit.parse(written)["mcp_servers"] + assert "id" not in native["managed"] + assert native["managed"].trivia.comment == "# keep my comment" + assert original.splitlines()[-1] in written + assert ownership == {"codex": {"managed"}} + + +@pytest.mark.parametrize("owned", [False, True]) +def test_inline_uuid_migration_preserves_conflicts_and_unowned_entries( + tmp_path: Path, dependency: MCPDependency, owned: bool +) -> None: + """Inline UUIDs remain usable for conflicts, and only owners authorize repair.""" + adapter = CodexClientAdapter(project_root=tmp_path) + original = _write_inline_config(adapter, dependency, registry_id="registry-uuid") + managed = {"managed"} if owned else set() + assert adapter.migrate_legacy_managed_servers(managed) == managed + path = Path(adapter.get_config_path()) + written = path.read_text(encoding="utf-8") + native = tomlkit.parse(written)["mcp_servers"] + assert ("id" not in native["managed"]) is owned + assert "# keep my comment" in written + assert original.splitlines()[-1] in written + if not owned: + assert written == original + info = {**MCPIntegrator._build_self_defined_info(dependency), "id": "registry-uuid"} + assert MCPConflictDetector(adapter).check_server_exists("alias", server_info=info) + assert ( + MCPServerOperations().check_servers_needing_installation( + ["codex"], ["alias"], project_root=tmp_path, server_info_cache={"alias": info} + ) + == [] + ) + summary = SafeMCPInstaller("codex", project_root=tmp_path).install_servers( + ["alias"], server_info_cache={"alias": info} + ) + assert summary.skipped == [{"server": "alias", "reason": "already configured"}] + assert not summary.installed and not summary.failed + assert adapter.migrate_legacy_managed_servers(managed) == set() + assert path.read_text(encoding="utf-8") == written + + +def test_inline_registry_comment_blocks_legacy_ownership_adoption( + tmp_path: Path, dependency: MCPDependency +) -> None: + """Moving a UUID into an inline comment cannot turn a user's server into APM's.""" + adapter = CodexClientAdapter(project_root=tmp_path) + original = _write_inline_config( + adapter, + dependency, + registry_id=None, + comment='# apm-registry-id: "user-registry-uuid" # keep my comment', + ) + ownership = resolve_mcp_target_servers( + recorded_target_servers={}, + ownership_present=False, + server_names={"managed"}, + stored_configs={"managed": dependency.to_dict()}, + project_root=tmp_path, + user_scope=False, + ) + assert ownership == {} + assert Path(adapter.get_config_path()).read_text(encoding="utf-8") == original + + +@pytest.mark.parametrize("full_root_keys", [False, True]) +@pytest.mark.parametrize("registry_id", ["", "registry-uuid"]) +def test_dotted_legacy_config_preserves_identity_and_layout( + tmp_path: Path, dependency: MCPDependency, full_root_keys: bool, registry_id: str +) -> None: + """Implicit dotted server mappings migrate without materializing new tables.""" + prefix = "mcp_servers." if full_root_keys else "" + header = "" if full_root_keys else "[mcp_servers]\n" + transport = ( + f'{prefix}managed.command = "echo" # anchor note\n' + f'{prefix}managed.args = ["hello"]\n' + f"{prefix}managed.env = {{}}\n" + if dependency.command + else f'{prefix}managed.url = "https://example.test/mcp" # anchor note\n' + ) + identity_line = f'{prefix}managed.id = "{registry_id}"\n' + original = ( + header + + transport + + identity_line + + f'{prefix}personal.command = "custom" # user note\n' + + f'{prefix}personal.id = "personal-id"\n' + ) + adapter = CodexClientAdapter(project_root=tmp_path) + path = Path(adapter.get_config_path()) + path.parent.mkdir() + path.write_text(original, encoding="utf-8") + stored = {"managed": dependency.to_dict()} + ownership = resolve_mcp_target_servers( + recorded_target_servers={"codex": {"managed"}} if registry_id else {}, + ownership_present=bool(registry_id), + server_names={"managed"}, + stored_configs=stored, + project_root=tmp_path, + user_scope=False, + ) + assert ownership == {"codex": {"managed"}} + MCPIntegrator.install( + [dependency], + project_root=tmp_path, + explicit_target="codex", + stored_mcp_configs=stored, + managed_target_servers=ownership, + ) + written = path.read_text(encoding="utf-8") + expected = original.replace(identity_line, "") + if registry_id: + expected = expected.replace( + "# anchor note", '# apm-registry-id: "registry-uuid" # anchor note' + ) + assert written == expected + assert "id" not in tomlkit.parse(written)["mcp_servers"]["managed"] + assert adapter.migrate_legacy_managed_servers({"managed"}) == set() + assert path.read_text(encoding="utf-8") == written + if registry_id: + info = {**MCPIntegrator._build_self_defined_info(dependency), "id": registry_id} + assert MCPConflictDetector(adapter).check_server_exists("alias", server_info=info) + assert ( + MCPServerOperations().check_servers_needing_installation( + ["codex"], ["alias"], project_root=tmp_path, server_info_cache={"alias": info} + ) + == [] + ) + assert ( + resolve_mcp_target_servers( + recorded_target_servers={}, + ownership_present=False, + server_names={"managed"}, + stored_configs=stored, + project_root=tmp_path, + user_scope=False, + ) + == {} + ) + + +@pytest.mark.parametrize("with_personal", [False, True]) +def test_fresh_registry_server_in_inline_parent_remains_valid_toml( + tmp_path: Path, dependency: MCPDependency, with_personal: bool +) -> None: + """Adding annotated output to an inline mcp_servers parent must stay parseable.""" + adapter = CodexClientAdapter(project_root=tmp_path) + personal = 'personal={command="custom",args=["one","two"]}' if with_personal else "" + path = Path(adapter.get_config_path()) + path.parent.mkdir() + path.write_text( + f'mcp_servers = {{{personal}}} # parent note\nmodel = "gpt-5" # model note\n', + encoding="utf-8", + ) + info = {**MCPIntegrator._build_self_defined_info(dependency), "id": "registry-uuid"} + assert adapter.configure_mcp_server("managed", server_info_cache={"managed": info}) + written = path.read_text(encoding="utf-8") + native = tomlkit.parse(written) + assert "id" not in native["mcp_servers"]["managed"] + assert adapter.get_registry_id(native["mcp_servers"]["managed"]) == "registry-uuid" + assert native["model"] == "gpt-5" + assert "# parent note" in written and "# model note" in written + if with_personal: + assert native["mcp_servers"]["personal"] == {"command": "custom", "args": ["one", "two"]} + else: + assert set(native["mcp_servers"]) == {"managed"} + + +@pytest.mark.parametrize("inline_parent", [False, True]) +@pytest.mark.parametrize("id_position", [0, 1, 3], ids=["first", "middle", "last"]) +@pytest.mark.parametrize("registry_id", ["", "registry-uuid"]) +def test_inline_id_positions_migrate_without_invalid_commas_or_nested_comments( + tmp_path: Path, + dependency: MCPDependency, + inline_parent: bool, + id_position: int, + registry_id: str, +) -> None: + """Deleting any inline id preserves both the owned server and its neighbors.""" + fields = ( + ['command="echo"', 'args=["hello"]', "enabled=false"] + if dependency.command + else ['url="https://example.test/mcp"', 'http_headers={TEST="value"}', "enabled=false"] + ) + fields.insert(id_position, f'id="{registry_id}"') + managed = "managed={" + ",".join(fields) + "}" + personal = 'personal={command="custom",id="personal-id",env={TEST="keep"}}' + original = ( + f"mcp_servers={{{managed},{personal}}} # parent note\n" + if inline_parent + else f"[mcp_servers] # parent note\n{managed} # managed note\n{personal} # personal note\n" + ) + adapter = CodexClientAdapter(project_root=tmp_path) + path = Path(adapter.get_config_path()) + path.parent.mkdir() + path.write_text(original, encoding="utf-8") + original_native = tomlkit.parse(original) + # No ownership means no layout conversion and no field removal. + assert adapter.migrate_legacy_managed_servers(set()) == set() + assert path.read_text(encoding="utf-8") == original + + assert adapter.migrate_legacy_managed_servers({"managed"}) == {"managed"} + written = path.read_text(encoding="utf-8") + native = tomlkit.parse(written)["mcp_servers"] + assert "id" not in native["managed"] + assert native["managed"]["enabled"] is False + assert native["personal"] == original_native["mcp_servers"]["personal"] + assert native["managed"] == { + key: value + for key, value in original_native["mcp_servers"]["managed"].items() + if key != "id" + } + assert "# parent note" in written + if not inline_parent: + assert "# managed note" in written and "# personal note" in written + if registry_id: + assert adapter.get_registry_id(native["managed"]) == registry_id + assert MCPConflictDetector(adapter).check_server_exists( + "alias", server_info={"id": registry_id} + ) + assert adapter.migrate_legacy_managed_servers({"managed"}) == set() + assert path.read_text(encoding="utf-8") == written + + +def test_invalid_serialization_never_replaces_native_config( + tmp_path: Path, dependency: MCPDependency, monkeypatch: pytest.MonkeyPatch +) -> None: + """A serializer failure must leave the previous configuration byte-for-byte intact.""" + adapter = CodexClientAdapter(project_root=tmp_path) + path = Path(adapter.get_config_path()) + path.parent.mkdir() + original = '[mcp_servers.personal]\ncommand = "custom" # preserve\n' + path.write_text(original, encoding="utf-8") + info = {**MCPIntegrator._build_self_defined_info(dependency), "id": "registry-uuid"} + monkeypatch.setattr("apm_cli.adapters.client.codex.tomlkit.dumps", lambda _config: "[invalid") + assert not adapter.configure_mcp_server("managed", server_info_cache={"managed": info}) + assert path.read_text(encoding="utf-8") == original