fix(codex): forward ${VAR} stdio env values with env_vars instead of writing them

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Codex stdio MCP connections written from `apm.yml` forward a `${VAR}` value declared under the same variable name through `env_vars` instead of resolving it into `config.toml`, so the secret never reaches a committed file. Other values keep resolving at install time. (#3393)
+
 ## [0.35.0] - 2026-10-22
 
 ### Added

scripts/architecture_linter/checks/mutation_mcp_target_and_scope.py

@@ -598,6 +598,7 @@ def _check_mcp_passthrough_denylist(provider: FactsProvider) -> Iterable[Violati
                 '        "bearer_token_env_var",\n'
                 '        "enabled",\n'
                 '        "env_http_headers",\n'
+                '        "env_vars",\n'
                 '        "environment",\n'
                 '        "http_headers",\n'
                 '        "id",\n'

src/apm_cli/adapters/client/base.py

@@ -391,6 +391,10 @@ def matches_managed_server_config(self, native: object, expected: dict) -> bool:
         """Compare the native connection with the adapter's rendered baseline."""
         return native == expected
 
+    def matches_legacy_managed_server_config(self, native: object, expected: dict) -> bool:
+        """Recognize a verified write from an older formatter; never a current match."""
+        return False
+
     def managed_server_config(self, native: dict) -> dict:
         """Return connection data for an actual-write fingerprint."""
         return dict(native)

src/apm_cli/adapters/client/codex.py

@@ -219,6 +219,26 @@ def matches_managed_server_config(self, native: object, expected: dict) -> bool:
             comparable.pop("id")
         return comparable == expected
 
+    def matches_legacy_managed_server_config(self, native: object, expected: dict) -> bool:
+        """Recognize a stdio entry written before ``${VAR}`` values moved to ``env_vars``.
+
+        The older formatter stored either the resolved value or the literal
+        reference, so anything else is a manual edit and stays a conflict.
+        """
+        names = expected.get("env_vars")
+        native_env = native.get("env") if isinstance(native, dict) else None
+        if not names or not isinstance(native_env, dict) or "env_vars" in native:
+            return False
+        legacy_env = dict(expected.get("env") or {})
+        for name in names:
+            value = native_env.get(name)
+            if value is None or value not in (f"${{{name}}}", os.getenv(name)):
+                return False
+            legacy_env[name] = value
+        legacy = {key: value for key, value in expected.items() if key != "env_vars"}
+        legacy["env"] = legacy_env
+        return self.matches_managed_server_config(native, legacy)
+
     def update_config(self, config_updates):
         """Update the Codex CLI MCP configuration.
 
@@ -363,17 +383,29 @@ def _format_server_config(self, server_info, env_overrides=None, runtime_vars=No
         # syntaxes (``<VAR>``, ``${VAR}``, ``${env:VAR}``) are resolved at
         # install time before being written to ~/.codex/config.toml.
         # See issue #1266.
+        #
+        # A value authored as exactly ``${VAR}`` under the key ``VAR`` is the
+        # one case that must NOT be resolved and stored: Codex reads those from
+        # the launching environment through ``env_vars`` at server start, so the
+        # secret never lands in config.toml. See issue #3393.
         raw = server_info.get("_raw_stdio")
         if raw:
-            config["command"] = raw["command"]
-            if raw.get("cwd") is not None:
-                config["cwd"] = raw["cwd"]
+            forwarded_env: list[str] = []
             resolved_env_for_args: dict = {}
+            written_env: dict = {}
             if raw.get("env"):
-                resolved_env_for_args = self._resolve_environment_variables(
-                    raw["env"], env_overrides=env_overrides
+                literal_env, forwarded_env = self._split_forwarded_stdio_env(raw["env"])
+                written_env = self._resolve_environment_variables(
+                    literal_env, env_overrides=env_overrides
                 )
-                config["env"] = resolved_env_for_args
+                # Args keep resolving against every declared name; a forwarded
+                # value is read from the environment when it is available, and
+                # never prompts for a value that only the runtime needs.
+                resolved_env_for_args = dict(written_env)
+                for name in forwarded_env:
+                    value = (env_overrides or {}).get(name) or os.getenv(name)
+                    if value:
+                        resolved_env_for_args[name] = value
                 self._warn_input_variables(raw["env"], server_info.get("name", ""), "Codex CLI")
 
             def _process_stdio_arg(arg):
@@ -383,9 +415,17 @@ def _process_stdio_arg(arg):
                     )
                 return self.normalize_project_arg(arg)
 
-            config["args"] = [_process_stdio_arg(arg) for arg in raw.get("args") or []]
-            self._merge_extra(config, server_info)
-            return config
+            # Smoke scalars and arrays before the [env] sub-table: tomlkit nests
+            # any later key inside the table it has already opened.
+            stdio_config: dict = {"command": raw["command"]}
+            if raw.get("cwd") is not None:
+                stdio_config["cwd"] = raw["cwd"]
+            stdio_config["args"] = [_process_stdio_arg(arg) for arg in raw.get("args") or []]
+            if forwarded_env:
+                stdio_config["env_vars"] = forwarded_env
+            stdio_config["env"] = written_env
+            self._merge_extra(stdio_config, server_info)
+            return stdio_config
 
         # Remote MCP handling.
         # Precedence on Codex when a server publishes BOTH a remote and a stdio
@@ -656,6 +696,31 @@ def _process_arguments(  # pylint: disable=duplicate-code  # structural similari
 
         return processed
 
+    @staticmethod
+    def _split_forwarded_stdio_env(raw_env: dict) -> tuple[dict, list[str]]:
+        """Split a self-defined stdio env block into literals and forwarded names.
+
+        Codex reads a stdio server's variables from the environment that launches
+        it when they are named in ``env_vars``, so a value written as exactly
+        ``${VAR}`` under the key ``VAR`` does not have to be resolved and stored
+        in config.toml. Everything else keeps the existing install-time
+        resolution and is written to ``[mcp_servers.<name>.env]``.
+
+        Returns:
+            tuple: ``(literal_env, forwarded_env)`` -- the entries to resolve and
+                the variable names to forward.
+        """
+        literal_env: dict = {}
+        forwarded_env: list[str] = []
+        for name, raw_value in raw_env.items():
+            if isinstance(raw_value, str):
+                reference = _ENV_VAR_RE.fullmatch(raw_value)
+                if reference and reference.group(1) == name:
+                    forwarded_env.append(name)
+                    continue
+            literal_env[name] = raw_value
+        return literal_env, forwarded_env
+
     def _process_environment_variables(self, env_vars, env_overrides=None):
         """Process environment variable definitions and resolve actual values.
 

src/apm_cli/install/mcp/ownership.py

@@ -79,7 +79,9 @@ def compare_native_mcp_connection(
                     comparison = client.project_managed_server_config(
                         native, baseline=expected, requested=requested
                     )
-                matches_baseline = client.matches_managed_server_config(comparison, expected)
+                matches_baseline = client.matches_managed_server_config(
+                    comparison, expected
+                ) or client.matches_legacy_managed_server_config(native, expected)
                 if matches_baseline and managed_fields is not None:
                     managed_fields.update(expected)
             elif native_hash is not None and isinstance(native, dict):

src/apm_cli/models/dependency/mcp.py

@@ -62,6 +62,7 @@
         "bearer_token_env_var",
         "enabled",
         "env_http_headers",
+        "env_vars",
         "environment",
         "http_headers",
         "id",

tests/unit/install/test_mcp_integrator_install_flow.py

@@ -493,14 +493,18 @@ def test_self_defined_stdio_env_placeholders_resolve_from_process_env(
 
         assert result == 1
         config_path = tmp_path / config_relpath
+        content = config_path.read_text(encoding="utf-8")
         if runtime == "claude":
-            env_block = json.loads(config_path.read_text(encoding="utf-8"))["mcpServers"][
-                "env-demo"
-            ]["env"]
-        else:
-            env_block = toml.load(config_path)["mcp_servers"]["env-demo"]["env"]
-        assert env_block["MY_TOKEN"] == "repro-secret-not-a-real-token"
-        assert env_block["LITERAL_VALUE"] == "literal-value"
+            env_block = json.loads(content)["mcpServers"]["env-demo"]["env"]
+            assert env_block["MY_TOKEN"] == "repro-secret-not-a-real-token"
+            assert env_block["LITERAL_VALUE"] == "literal-value"
+            return
+        # Codex forwards a ``${VAR}`` written under its own key from the
+        # launching environment, so the resolved secret never reaches the file.
+        server = toml.loads(content)["mcp_servers"]["env-demo"]
+        assert server["env_vars"] == ["MY_TOKEN"]
+        assert server["env"] == {"LITERAL_VALUE": "literal-value"}
+        assert "repro-secret-not-a-real-token" not in content
 
     def test_self_defined_non_stdio_env_remains_env_overrides(self) -> None:
         from apm_cli.integration.mcp_integrator import MCPIntegrator

tests/unit/install/test_mcp_ownership.py

@@ -9,6 +9,7 @@
 import pytest
 
 from apm_cli.adapters.client.claude import ClaudeClientAdapter
+from apm_cli.adapters.client.codex import CodexClientAdapter
 from apm_cli.install.errors import RequiredIntegrationError
 from apm_cli.install.mcp.ownership import (
     NativeMcpConnectionState,
@@ -427,3 +428,45 @@ def test_legacy_adoption_inspects_intellij_obsolete_config(tmp_path) -> None:
         )
 
     assert adopted == {"intellij": {"managed"}}
+
+
+_FORWARDED_STDIO = {
+    "name": "managed",
+    "registry": False,
+    "transport": "stdio",
+    "command": "python",
+    "args": ["-m", "fixture_server"],
+    "env": {"MY_TOKEN": "${MY_TOKEN}"},
+}
+
+
+def _compare_codex(tmp_path: Path, written_value: str) -> NativeMcpConnectionState:
+    """Compare a Codex entry the older formatter wrote with its current rendering."""
+    client = CodexClientAdapter(project_root=tmp_path)
+    requested = MCPIntegrator._build_self_defined_info(MCPDependency.from_dict(_FORWARDED_STDIO))
+    native = deepcopy(client.render_server_config(requested))
+    native.pop("env_vars")
+    native["env"] = {"MY_TOKEN": written_value}
+    return compare_native_mcp_connection(
+        client,
+        "managed",
+        requested_info=requested,
+        stored_config=_FORWARDED_STDIO,
+        managed=True,
+        existing_configs={"managed": native},
+    )
+
+
+@pytest.mark.parametrize("written_value", ["${MY_TOKEN}", "resolved-value"])
+def test_codex_entry_from_older_formatter_upgrades_as_managed(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, written_value: str
+) -> None:
+    monkeypatch.setenv("MY_TOKEN", "resolved-value")
+    assert _compare_codex(tmp_path, written_value) is NativeMcpConnectionState.UPDATE_MANAGED
+
+
+def test_codex_entry_with_manual_edit_stays_a_conflict(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("MY_TOKEN", "resolved-value")
+    assert _compare_codex(tmp_path, "edited-by-hand") is NativeMcpConnectionState.CONFLICT

tests/unit/test_codex_adapter_phase3.py

@@ -876,3 +876,97 @@ def test_env_var_resolved_from_os_environ(self, tmp_path: Path) -> None:
         with patch.dict(os.environ, {"MY_TOKEN": "from-env", "CI": "true"}):
             result = adapter._process_environment_variables(env_vars)
         assert result.get("MY_TOKEN") == "from-env"
+
+
+# ---------------------------------------------------------------------------
+# _raw_stdio env forwarding (#3393)
+# ---------------------------------------------------------------------------
+
+
+def _raw_stdio_server(**env: str) -> dict:
+    """Self-defined stdio dep, the shape ``apm.yml`` dependencies.mcp produce."""
+    return {
+        "name": "figma",
+        "remotes": [],
+        "_raw_stdio": {
+            "command": "npx",
+            "args": ["-y", "figma-developer-mcp", "--stdio"],
+            "env": env,
+        },
+    }
+
+
+class TestForwardedStdioEnv:
+    """A ``${VAR}`` stdio value is forwarded to Codex instead of written out."""
+
+    def test_reference_is_forwarded_instead_of_written(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        with patch.dict(os.environ, {"FIGMA_API_KEY": "example-token"}):
+            cfg = adapter._format_server_config(_raw_stdio_server(FIGMA_API_KEY="${FIGMA_API_KEY}"))
+        assert cfg["env_vars"] == ["FIGMA_API_KEY"]
+        assert "FIGMA_API_KEY" not in cfg["env"]
+        assert "example-token" not in tomlkit.dumps(cfg)
+
+    def test_reference_is_forwarded_when_variable_is_unset(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        with patch.dict(os.environ, {}, clear=False):
+            os.environ.pop("FIGMA_API_KEY", None)
+            cfg = adapter._format_server_config(_raw_stdio_server(FIGMA_API_KEY="${FIGMA_API_KEY}"))
+        assert cfg["env_vars"] == ["FIGMA_API_KEY"]
+        assert cfg["env"] == {}
+
+    def test_env_prefixed_reference_is_forwarded(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        cfg = adapter._format_server_config(_raw_stdio_server(FIGMA_API_KEY="${env:FIGMA_API_KEY}"))
+        assert cfg["env_vars"] == ["FIGMA_API_KEY"]
+        assert cfg["env"] == {}
+
+    @pytest.mark.parametrize("value", [" ${TOKEN} ", "${TOKEN}\n"])
+    def test_value_with_surrounding_whitespace_is_not_forwarded(
+        self, tmp_path: Path, value: str
+    ) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        with patch.dict(os.environ, {"TOKEN": "example-token"}):
+            cfg = adapter._format_server_config(_raw_stdio_server(TOKEN=value))
+        assert "env_vars" not in cfg
+        assert "TOKEN" in cfg["env"]
+
+    def test_extra_cannot_request_launch_environment_variables(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        server = _raw_stdio_server(LITERAL="value")
+        server["_extra"] = {"env_vars": ["PRIVATE_TOKEN"]}
+        cfg = adapter._format_server_config(server)
+        assert "env_vars" not in cfg
+
+    def test_literal_value_keeps_env_block(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        cfg = adapter._format_server_config(_raw_stdio_server(LOG_LEVEL="debug"))
+        assert "env_vars" not in cfg
+        assert cfg["env"] == {"LOG_LEVEL": "debug"}
+
+    def test_reference_under_another_key_keeps_install_time_value(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        with patch.dict(os.environ, {"FIGMA_API_KEY": "example-token"}):
+            cfg = adapter._format_server_config(_raw_stdio_server(TOKEN="${FIGMA_API_KEY}"))
+        assert "env_vars" not in cfg
+        assert cfg["env"] == {"TOKEN": "example-token"}
+
+    def test_args_still_resolve_forwarded_names_from_the_environment(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        server = _raw_stdio_server(FIGMA_API_KEY="${FIGMA_API_KEY}")
+        server["_raw_stdio"]["args"] = ["--token", "<FIGMA_API_KEY>"]
+        with patch.dict(os.environ, {"FIGMA_API_KEY": "example-token"}):
+            cfg = adapter._format_server_config(server)
+        assert cfg["args"] == ["--token", "example-token"]
+        assert cfg["env_vars"] == ["FIGMA_API_KEY"]
+
+    def test_written_config_keeps_env_vars_at_server_level(self, tmp_path: Path) -> None:
+        adapter = _make_adapter(project_root=tmp_path)
+        with patch.dict(os.environ, {"FIGMA_API_KEY": "example-token"}):
+            cfg = adapter._format_server_config(_raw_stdio_server(FIGMA_API_KEY="${FIGMA_API_KEY}"))
+        assert adapter.update_config({"figma": cfg}) is True
+        content = (tmp_path / ".codex" / "config.toml").read_text(encoding="utf-8")
+        server = tomlkit.parse(content).unwrap()["mcp_servers"]["figma"]
+        assert server["env_vars"] == ["FIGMA_API_KEY"]
+        assert "FIGMA_API_KEY" not in server.get("env", {})
+        assert "example-token" not in content