diff --git a/src/coder_eval/agents/codex_agent.py b/src/coder_eval/agents/codex_agent.py index 9be3274..3114331 100644 --- a/src/coder_eval/agents/codex_agent.py +++ b/src/coder_eval/agents/codex_agent.py @@ -7,6 +7,7 @@ import os import shlex import shutil +import stat import tempfile import time from collections.abc import Callable @@ -65,6 +66,8 @@ "Read": "shell", "Grep": "shell", "Glob": "shell", + "WebFetch": "web_search", + "WebSearch": "web_search", } # Approval mode — the SAME for every permission mode (no per-mode mapping). @@ -169,6 +172,7 @@ # ("StopIteration interacts badly with generators…") that escapes ``except # StopIteration`` — masking the real turn-failure reason on the stream-end path. _STREAM_DONE = object() +_WEB_FETCH_ACTION_TYPES = frozenset({"openpage", "findinpage", "open_page", "find_in_page"}) def _ms_to_dt(ms: int | None) -> datetime: @@ -184,6 +188,34 @@ def _status_value(status: Any) -> str: return str(value).lower() if value is not None else "" +def _web_search_action(root: Any) -> Any: + """Return a web-search item's concrete action, unwrapping the SDK RootModel.""" + action = getattr(root, "action", None) + return getattr(action, "root", action) + + +def _web_search_tool_name(root: Any) -> str: + """Classify page navigation as WebFetch while retaining searches as WebSearch.""" + action_type = _status_value(getattr(_web_search_action(root), "type", None)) + return "WebFetch" if action_type in _WEB_FETCH_ACTION_TYPES else "WebSearch" + + +def _web_search_parameters(root: Any) -> dict[str, Any]: + """Extract useful search/page-action inputs from a Codex webSearch item.""" + action = _web_search_action(root) + parameters: dict[str, Any] = {} + if action is not None: + for key in ("url", "pattern", "query", "queries"): + value = getattr(action, key, None) + if value not in (None, "", []): + parameters[key] = value + if "query" not in parameters and "queries" not in parameters: + query = getattr(root, "query", None) + if query not in (None, ""): + parameters["query"] = query + return parameters + + def _fresh_input_tokens(raw_input: int, cached: int) -> int: """The fresh (uncached) prompt slice = tokens written to cache this call. @@ -209,7 +241,7 @@ def _message_uncached_input(m: AssistantMessage) -> int: # supported"), so this is a fixed constant, not an operator knob. _CODEX_WIRE_API = "responses" -# Login-shell profile files generated into the per-task HOME (see +# Login-shell profile files generated into the per-start isolated HOME (see # _setup_login_shell_home). ``.bash_profile`` is what ``bash -l`` reads; # ``.profile`` covers ``sh``/``dash`` login shells. The zsh trio covers macOS, # where zsh is the default shell: ``.zshenv`` runs in EVERY zsh, ``.zprofile`` @@ -445,7 +477,7 @@ def on_item_started(self, notification: Any) -> None: tool_id = item_id or f"{root_type}_{self.next_sequence}" self.seq_by_id[tool_id] = self.next_sequence start_tel = CommandTelemetry( - tool_name=self._agent._tool_name(root_type), + tool_name=self._agent._tool_name(root_type, root), tool_id=tool_id, timestamp=datetime.now(), parameters=self._agent._tool_parameters(root, root_type), @@ -471,9 +503,16 @@ def on_item_completed(self, notification: Any) -> None: tool_id = getattr(root, "id", None) or f"{root_type}_{self.next_sequence}" seq = self.seq_by_id.get(tool_id, self.next_sequence) # This tool is now resolved — drop it from the orphan set. - self.open_tools.pop(tool_id, None) + provisional = self.open_tools.pop(tool_id, None) telemetry, is_error = self._agent._telemetry_for_item(root, root_type, tool_id, seq) + # Some Codex items (notably webSearch) acquire their concrete action + # only at item/completed. Upgrade the ToolStart telemetry object in + # place so callbacks retaining that reference see the final canonical + # name/parameters too. + if provisional is not None and telemetry is not None: + provisional.tool_name = telemetry.tool_name + provisional.parameters.update(telemetry.parameters) if telemetry: self.commands.append(telemetry) self.emit.on_event( @@ -482,7 +521,7 @@ def on_item_completed(self, notification: Any) -> None: turn_id=self.turn_id, tool=telemetry or CommandTelemetry( - tool_name=self._agent._tool_name(root_type), + tool_name=self._agent._tool_name(root_type, root), tool_id=tool_id, timestamp=datetime.now(), sequence_number=seq, @@ -654,7 +693,9 @@ def __init__( self.thread: Any = None self.working_directory: Path | None = None self._env_path_prepend: list[str] = [] + self._runtime_root: Path | None = None self._login_shell_home: Path | None = None + self._isolated_codex_home: Path | None = None # _state / _iteration / _iteration_was_incremented / pending_turn lifecycle # bookkeeping lives on the Agent base class (shared defaults + helpers). self._log = PrefixedAdapter(logger, {"prefix": instance_name}) @@ -683,6 +724,11 @@ async def start( """ self.working_directory = Path(working_directory) self._env_path_prepend = list(env_path_prepend or []) + # start() is retried by the orchestrator. Reap the previous SDK process + # before replacing its runtime root so it cannot keep using a deleted + # HOME/CODEX_HOME. + self._close_client() + self.thread = None self._setup_login_shell_home() self._state = AgentState.WORKING @@ -691,13 +737,12 @@ async def start( # Build CodexConfig with environment variables for custom API configuration env_override = self._build_codex_env() - config = CodexConfig(env=env_override) if env_override else None + config = CodexConfig( + env=env_override, + config_overrides=self._build_codex_config_overrides(), + ) - # Initialize the Codex client (context manager compatible). Close any - # prior client first: start() is driven through execute_with_retry, so - # a retried start would otherwise orphan the previous app-server - # subprocess + reader threads (reaped only at final cleanup). - self._close_client() + # Initialize the Codex client (context manager compatible). self.codex_client = Codex(config=config) self._log.debug("Codex client initialized") @@ -720,8 +765,12 @@ async def start( self._log_config_enforcement() except ImportError as e: + self._close_client() + self._cleanup_login_shell_home() raise RuntimeError("Codex SDK not installed. Install with: pip install 'coder-eval[codex]'") from e except Exception as e: + self._close_client() + self._cleanup_login_shell_home() raise RuntimeError(f"Failed to initialize Codex client: {e}") from e async def communicate( @@ -1081,30 +1130,54 @@ def _build_codex_env(self) -> dict[str, str] | None: env[path_key] = os.pathsep.join([*self._env_path_prepend, os.environ.get(path_key, "")]) self._log.debug(f"PATH prepend: {os.pathsep.join(self._env_path_prepend)}") if self._login_shell_home is not None: - # Point login shells at the generated profile dir (see - # _setup_login_shell_home) while pinning codex state (auth, rollout - # sessions) to its real location — _codex_home() reads the same - # resolution for sub-agent rollout recovery, so both sides agree. + # Point the app-server and login shells at the per-start runtime + # (see _setup_login_shell_home). CODEX_HOME contains only isolated + # run state plus an optional auth.json bridge; host config, plugins, + # skills, and sessions are deliberately absent. _codex_home() reads + # this same directory for sub-agent rollout recovery. # HOME steers bash/sh; ZDOTDIR steers zsh (the macOS default # shell), which ignores HOME for dotfile selection when it is set. env["HOME"] = str(self._login_shell_home) env["ZDOTDIR"] = str(self._login_shell_home) - # The binary hard-errors on an explicitly set CODEX_HOME that does - # not exist (unset, it materializes the ~/.codex default itself) — - # hosts that auth via CODEX_API_KEY never ran `codex login`, so - # the dir may not exist yet. Create it before pinning. + if self._is_windows(): + # Codex also uses USERPROFILE for global discovery on Windows. + # Tool subprocesses get the host values back through + # shell_environment_policy (see _build_codex_config_overrides). + env["USERPROFILE"] = str(self._login_shell_home) codex_home = self._codex_home() codex_home.mkdir(parents=True, exist_ok=True) env["CODEX_HOME"] = str(codex_home) return env if env else None + @staticmethod + def _is_windows() -> bool: + return os.name == "nt" + + def _build_codex_config_overrides(self) -> tuple[str, ...]: + """Restore host home variables only inside Windows tool subprocesses. + + The app-server keeps isolated HOME/USERPROFILE so Codex cannot discover + user-global skills or config. Windows shells do not have POSIX startup + profiles where we can restore the original homes, so Codex's supported + shell environment policy supplies them only to spawned tools. Values + are JSON-quoted, which is valid TOML basic-string syntax and safely + handles Windows separators, quotes, and control characters. + """ + if not self._is_windows(): + return () + restore = {key: os.environ[key] for key in ("HOME", "USERPROFILE") if os.environ.get(key)} + if not restore: + return () + entries = ", ".join(f"{key} = {json.dumps(value)}" for key, value in restore.items()) + return (f"shell_environment_policy.set={{ {entries} }}",) + @staticmethod def _login_shell_profiles_supported() -> bool: """Whether to generate login-shell profile shims (POSIX shells only).""" return os.name == "posix" def _setup_login_shell_home(self) -> None: - """Create a per-task HOME whose profiles restore the mock PATH prepend. + """Create a per-start runtime with isolated HOME and CODEX_HOME. Codex issues every shell command through the user's default shell as a login shell: ``bash -lc`` on Linux, ``zsh -lc`` on macOS (its default @@ -1114,17 +1187,17 @@ def _setup_login_shell_home(self) -> None: which unconditionally RESETS PATH, silently dropping the mock-CLI prepend passed via the app-server environment, so bare commands resolve to the REAL CLIs (real-tenant contamination). The per-user - dotfiles are sourced AFTER that chain, so a generated per-task HOME - (wired up in ``_build_codex_env``; codex state stays in CODEX_HOME) - gets the last word and re-prepends the mock dirs: + dotfiles are sourced AFTER that chain, so generated profiles in the + isolated HOME (wired up in ``_build_codex_env``) get the last word and + optionally re-prepend the mock dirs: ``.bash_profile``/``.profile`` for bash/sh (selected via env HOME) and ``.zshenv``/``.zprofile``/``.zshrc`` for zsh (selected via env ZDOTDIR; ``.zshrc`` also feeds codex's shell snapshot, which sources - it explicitly). Per-task rather than the user's real dotfiles so - parallel tasks with different mocks cannot collide. No-op without mock - dirs or on non-POSIX hosts: Windows codex shells through PowerShell - (``-NoProfile``) or ``cmd /c``, neither of which re-sources a profile - chain that resets PATH, so the plain env prepend survives there as-is. + it explicitly). The runtime is created even without mock dirs and on + non-POSIX hosts because HOME/CODEX_HOME isolation is also the skill + provenance boundary: user config, plugins, skills, and historical + sessions must not enter an eval. Only a validated host ``auth.json`` is + exposed when CODEX_API_KEY is absent. Bash uses the env HOME only to PICK the profile file; the generated profile's first act is to export the ORIGINAL home back, so the @@ -1135,36 +1208,87 @@ def _setup_login_shell_home(self) -> None: keeps it - ZDOTDIR stays exported). """ self._cleanup_login_shell_home() - if not (self._env_path_prepend and self._login_shell_profiles_supported()): - return original_home = os.environ.get("HOME", "") # Where the user's REAL zsh dotfiles live: their own ZDOTDIR when set, # else their home (zsh's fallback). original_zdotdir = os.environ.get("ZDOTDIR", "") or original_home # The profile only ever executes under a POSIX shell, so the PATH # separator is ':' regardless of the host building it. - quoted_prepend = shlex.quote(":".join(self._env_path_prepend)) - export_line = f'export PATH={quoted_prepend}:"$PATH"' - # Track the dir BEFORE writing so a failed write can't orphan it — - # the except below (and any later cleanup) always sees it. - home = Path(tempfile.mkdtemp(prefix="coder-eval-codex-home-")) + export_line = "" + if self._env_path_prepend: + quoted_prepend = shlex.quote(":".join(self._env_path_prepend)) + export_line = f'export PATH={quoted_prepend}:"$PATH"' + # Track the root BEFORE creating children so any failure is cleaned. + runtime_root = Path(tempfile.mkdtemp(prefix="coder-eval-codex-runtime-")) + self._runtime_root = runtime_root + home = runtime_root / "home" + codex_home = runtime_root / "codex" self._login_shell_home = home + self._isolated_codex_home = codex_home try: - for name in _LOGIN_PROFILE_NAMES: - content = self._login_profile_content( - name, - original_home, - export_line, - original_zdotdir=original_zdotdir, - generated_home=str(home), - ) - # newline="\n": the profile must stay LF-only no matter which host - # builds it, or bash sees literal \r at end of line. - (home / name).write_text(content, encoding="utf-8", newline="\n") + # POSIX gets explicit owner-only modes. Windows chmod does not model + # ACLs; there the fresh mkdtemp tree inherits the current user's + # temp-directory ACL, and no path is reused or shared. + runtime_root.chmod(0o700) + home.mkdir(mode=0o700) + codex_home.mkdir(mode=0o700) + self._copy_host_auth(codex_home) + if self._login_shell_profiles_supported(): + for name in _LOGIN_PROFILE_NAMES: + content = self._login_profile_content( + name, + original_home, + export_line, + original_zdotdir=original_zdotdir, + generated_home=str(home), + ) + # newline="\n": the profile must stay LF-only no matter which host + # builds it, or bash sees literal \r at end of line. + (home / name).write_text(content, encoding="utf-8", newline="\n") except Exception: self._cleanup_login_shell_home() raise - self._log.debug(f"Login-shell mock-PATH home: {home}") + self._log.debug("Codex isolated runtime: %s", runtime_root) + + def _copy_host_auth(self, isolated_codex_home: Path) -> None: + """Copy only the host auth file into the isolated CODEX_HOME. + + A copy, never a symlink or hard link, ensures SDK token refreshes and + atomic replacements cannot mutate host credentials. On platforms with + ``O_NOFOLLOW``, opening the source rejects a symlink swap; ``fstat`` + verifies the opened object itself is a regular file on every platform. + The exclusive target uses 0600 from creation on POSIX, avoiding a + permissions window before the final defensive chmod. On Windows it + remains protected by the fresh runtime tree's inherited user ACL. + """ + if os.getenv("CODEX_API_KEY"): + return + source = self._host_codex_home() / "auth.json" + try: + source = source.resolve(strict=True) + except OSError: + return + source_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + source_fd = os.open(source, source_flags) + except OSError: + return + try: + if not stat.S_ISREG(os.fstat(source_fd).st_mode): + return + target = isolated_codex_home / "auth.json" + + def private_opener(path: str, flags: int) -> int: + return os.open(path, flags, 0o600) + + with os.fdopen(source_fd, "rb") as source_file: + source_fd = -1 + with open(target, "xb", opener=private_opener) as target_file: + shutil.copyfileobj(source_file, target_file) + target.chmod(0o600) + finally: + if source_fd >= 0: + os.close(source_fd) @staticmethod def _login_profile_content( @@ -1229,15 +1353,20 @@ def _login_profile_content( f"if [ -r {orig}/.profile ]; then . {orig}/.profile", "fi", ] - lines.append(export_line) + if export_line: + lines.append(export_line) return "\n".join(lines) + "\n" def _cleanup_login_shell_home(self) -> None: - """Remove the generated per-task HOME (best-effort, idempotent).""" - home, self._login_shell_home = self._login_shell_home, None - if home is None: - return - shutil.rmtree(home, ignore_errors=True) + """Remove the per-start runtime (best-effort, idempotent).""" + runtime_root, self._runtime_root = self._runtime_root, None + login_home, self._login_shell_home = self._login_shell_home, None + self._isolated_codex_home = None + if runtime_root is not None: + shutil.rmtree(runtime_root, ignore_errors=True) + elif login_home is not None: + # Compatibility for callers/tests that injected a standalone home. + shutil.rmtree(login_home, ignore_errors=True) def _build_thread_options(self) -> dict[str, Any]: """Build thread_start options from agent config. @@ -1506,8 +1635,10 @@ def _flush() -> None: return rebuilt @staticmethod - def _tool_name(root_type: str | None) -> str: + def _tool_name(root_type: str | None, root: Any = None) -> str: """Friendly tool label for a Codex item type (falls back to the raw type).""" + if root_type == "webSearch": + return _web_search_tool_name(root) return _TOOL_ITEM_NAMES.get(root_type or "", root_type or "Tool") def _tool_parameters(self, root: Any, root_type: str | None) -> dict[str, Any]: @@ -1538,7 +1669,7 @@ def _tool_parameters(self, root: Any, root_type: str | None) -> dict[str, Any]: params["arguments"] = args return params if root_type == "webSearch": - return {"query": getattr(root, "query", "")} + return _web_search_parameters(root) if root_type == "imageView": return {"path": str(getattr(root, "path", ""))} if root_type == "imageGeneration": @@ -1580,7 +1711,7 @@ def _extract_generic_telemetry( duration_ms = getattr(root, "duration_ms", None) return ( CommandTelemetry( - tool_name=self._tool_name(root_type), + tool_name=self._tool_name(root_type, root), tool_id=tool_id, timestamp=datetime.now(), duration_ms=float(duration_ms) if duration_ms is not None else None, @@ -1605,7 +1736,8 @@ def _summarize_tool_item(root: Any, root_type: str | None) -> str: messages = [s.message for s in states.values() if getattr(s, "message", None)] return f"collab {op}: {'; '.join(messages)[:200]}" if messages else f"collab {op}" if root_type == "webSearch": - return f"query: {getattr(root, 'query', '')}" + parameters = _web_search_parameters(root) + return "; ".join(f"{key}: {value}" for key, value in parameters.items()) if root_type in ("mcpToolCall", "dynamicToolCall"): return f"{getattr(root, 'server', '') or getattr(root, 'namespace', '')}:{getattr(root, 'tool', '')}".strip( ":" @@ -1733,10 +1865,14 @@ async def _recover_subagent_tool_calls( self._log.debug("CodexAgent: sub-agent recovery failed for %s: %s", thread_id, exc) @staticmethod - def _codex_home() -> Path: - """Codex data directory (rollouts live under ``/sessions``).""" + def _host_codex_home() -> Path: + """Host Codex data directory, consulted only for ``auth.json``.""" return Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex")) + def _codex_home(self) -> Path: + """This agent's Codex data directory (rollouts under ``sessions``).""" + return self._isolated_codex_home or self._host_codex_home() + async def _await_rollout_file(self, home: Path, thread_id: str, *, attempts: int = 20) -> Path | None: """Locate a thread's rollout file, polling briefly for the async flush. diff --git a/tests/test_codex_agent.py b/tests/test_codex_agent.py index 95ee117..98488ad 100644 --- a/tests/test_codex_agent.py +++ b/tests/test_codex_agent.py @@ -471,9 +471,11 @@ def test_get_state_returns_current_state(): # without a live SDK. These mirror the notification shapes the real stream emits. # --------------------------------------------------------------------------- +import json # noqa: E402 import os # noqa: E402 import shlex # noqa: E402 import shutil # noqa: E402 +import stat # noqa: E402 import subprocess # noqa: E402 import tempfile # noqa: E402 import time # noqa: E402 @@ -1053,6 +1055,33 @@ async def test_spawn_nests_subagent_message_and_records_tool_calls(self, monkeyp assert nested[0].content_blocks[0].text == "5050" assert nested[0].model == "gpt-5.5" + async def test_subagent_rollout_recovery_reads_isolated_codex_home(self, monkeypatch, tmp_path): + monkeypatch.setenv("CODEX_HOME", str(tmp_path / "host-codex")) + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX)) + agent._setup_login_shell_home() + isolated = agent._codex_home() + seen: list[tuple[Path, str]] = [] + + async def capture_home(home, thread_id, *, attempts=20): + seen.append((home, thread_id)) + return None + + monkeypatch.setattr(agent, "_await_rollout_file", capture_home) + try: + await agent._recover_subagent_tool_calls( + [("child-thread", "spawn-call", "gpt-5.6")], + {}, + [], + [], + SimpleNamespace(on_event=lambda _event: None), + "task", + "turn", + ) + assert seen == [(isolated, "child-thread")] + assert isolated != tmp_path / "host-codex" + finally: + agent._cleanup_login_shell_home() + async def test_wait_only_records_no_subagent(self, monkeypatch, tmp_path): monkeypatch.setenv("CODEX_HOME", str(tmp_path)) # A wait with no preceding spawn (defensive) must not invent a sub-agent. @@ -1350,6 +1379,40 @@ async def test_mcp_and_websearch_items_become_tool_calls(self): mcp_tel = next(c for c in record.commands if c.tool_name == "Mcp") assert "read_file" in (mcp_tel.result_summary or "") + @pytest.mark.parametrize( + ("action", "expected_parameters"), + [ + ( + SimpleNamespace(type="openPage", url="https://example.com/docs"), + {"url": "https://example.com/docs"}, + ), + ( + SimpleNamespace(type="findInPage", url="https://example.com/docs", pattern="guardrails"), + {"url": "https://example.com/docs", "pattern": "guardrails"}, + ), + ], + ) + async def test_web_page_actions_become_webfetch(self, action, expected_parameters): + web = SimpleNamespace( + type="webSearch", + id="ws_fetch", + query="", + action=SimpleNamespace(root=action), + ) + notifications = [ + _item_notification("item/started", web), + _item_notification("item/completed", web), + _turn_completed(), + ] + agent = _started_agent(parse_agent_config(type=AgentKind.CODEX), notifications) + + record = await agent.communicate("open the documentation") + + assert len(record.commands) == 1 + telemetry = record.commands[0] + assert telemetry.tool_name == "WebFetch" + assert telemetry.parameters == expected_parameters + async def test_failed_mcp_call_records_error(self): mcp = SimpleNamespace( type="mcpToolCall", @@ -1551,10 +1614,10 @@ class TestLoginShellMockPathHome: ``/etc/zprofile``'s path_helper), which unconditionally RESETS PATH — silently dropping the mock-CLI prepend passed via the app-server env, so bare commands resolve to the REAL CLIs (real-tenant contamination). The - agent therefore generates a per-task HOME whose ``.bash_profile``/ + agent therefore generates a per-start isolated HOME whose ``.bash_profile``/ ``.profile`` (bash/sh) and ``.zshenv``/``.zprofile``/``.zshrc`` (zsh) run AFTER that chain and restore the prepend; ``_build_codex_env`` points HOME - and ZDOTDIR at it and pins CODEX_HOME so codex state stays put.""" + and ZDOTDIR at it and pins CODEX_HOME to an isolated sibling directory.""" ALL_PROFILE_NAMES = (".bash_profile", ".profile", ".zshenv", ".zprofile", ".zshrc") ZSH_PROFILE_NAMES = (".zshenv", ".zprofile", ".zshrc") @@ -1728,55 +1791,176 @@ def test_generated_profiles_are_lf_only(self, monkeypatch, tmp_path): finally: agent._cleanup_login_shell_home() - def test_no_login_home_without_mock_dirs(self, monkeypatch): + def test_runtime_home_exists_without_mock_dirs_and_has_no_empty_path_export(self, monkeypatch, tmp_path): self._force_posix(monkeypatch) + monkeypatch.setenv("HOME", str(tmp_path / "orig-home")) agent = self._agent_with_prepend([]) agent._setup_login_shell_home() - assert agent._login_shell_home is None + try: + home = agent._login_shell_home + assert agent._runtime_root is not None + assert home is not None and home.parent == agent._runtime_root + assert agent._codex_home().parent == agent._runtime_root + for name in self.ALL_PROFILE_NAMES: + content = (home / name).read_text(encoding="utf-8") + assert "export PATH=" not in content + finally: + agent._cleanup_login_shell_home() - def test_no_login_home_on_unsupported_platform(self, monkeypatch): + def test_runtime_home_exists_on_unsupported_platform_without_profiles(self, monkeypatch): self._force_posix(monkeypatch, supported=False) agent = self._agent_with_prepend(["/sandbox/mocks"]) agent._setup_login_shell_home() - assert agent._login_shell_home is None + try: + home = agent._login_shell_home + assert home is not None and home.is_dir() + assert not any((home / name).exists() for name in self.ALL_PROFILE_NAMES) + finally: + agent._cleanup_login_shell_home() def test_build_codex_env_sets_home_and_pins_codex_home(self, monkeypatch, tmp_path): monkeypatch.delenv("CODEX_API_KEY", raising=False) monkeypatch.setenv("PATH", "/parent/bin") + monkeypatch.setenv("HOME", str(tmp_path / "host-home")) agent = self._agent_with_prepend(["/sandbox/mocks"]) - agent._login_shell_home = tmp_path / "login-home" + agent._setup_login_shell_home() - env = agent._build_codex_env() - assert env is not None - assert env["HOME"] == str(tmp_path / "login-home") - # zsh (macOS default shell) picks its dotfiles via ZDOTDIR, not HOME. - assert env["ZDOTDIR"] == str(tmp_path / "login-home") - # Codex state (auth, rollout sessions) must NOT move with HOME — the - # harness reads the same _codex_home() for sub-agent rollout recovery. - assert env["CODEX_HOME"] == str(agent._codex_home()) - - def test_build_codex_env_creates_missing_codex_home(self, monkeypatch, tmp_path): - """The codex binary hard-errors on an explicitly set CODEX_HOME that - does not exist (unset, it materializes the ~/.codex default itself). - Runners that auth via CODEX_API_KEY never ran ``codex login``, so the - dir may not exist — pinning it must create it first.""" + try: + env = agent._build_codex_env() + assert env is not None + root = agent._runtime_root + assert root is not None + assert Path(env["HOME"]).parent == root + assert Path(env["ZDOTDIR"]).parent == root + assert Path(env["CODEX_HOME"]).parent == root + assert env["CODEX_HOME"] == str(agent._codex_home()) + finally: + agent._cleanup_login_shell_home() + + def test_runtime_codex_home_exposes_only_host_auth(self, monkeypatch, tmp_path): monkeypatch.delenv("CODEX_API_KEY", raising=False) - monkeypatch.setenv("CODEX_HOME", str(tmp_path / "state" / ".codex")) - agent = self._agent_with_prepend(["/sandbox/mocks"]) - agent._login_shell_home = tmp_path / "login-home" + host_codex_home = tmp_path / "host-codex" + (host_codex_home / "skills" / "stale").mkdir(parents=True) + (host_codex_home / "skills" / "stale" / "SKILL.md").write_text("stale", encoding="utf-8") + (host_codex_home / "plugins").mkdir() + (host_codex_home / "config.toml").write_text("model = 'wrong'", encoding="utf-8") + (host_codex_home / "auth.json").write_text('{"token":"secret"}', encoding="utf-8") + monkeypatch.setenv("CODEX_HOME", str(host_codex_home)) + agent = self._agent_with_prepend([]) + agent._setup_login_shell_home() - env = agent._build_codex_env() - assert env is not None - assert Path(env["CODEX_HOME"]).is_dir() + try: + isolated = agent._codex_home() + assert isolated.parent == agent._runtime_root + auth_copy = isolated / "auth.json" + assert auth_copy.read_text(encoding="utf-8") == '{"token":"secret"}' + assert not auth_copy.is_symlink() + if os.name == "posix": + assert stat.S_IMODE(agent._runtime_root.stat().st_mode) == 0o700 + assert stat.S_IMODE(agent._login_shell_home.stat().st_mode) == 0o700 + assert stat.S_IMODE(isolated.stat().st_mode) == 0o700 + assert stat.S_IMODE(auth_copy.stat().st_mode) == 0o600 + assert not (isolated / "config.toml").exists() + assert not (isolated / "plugins").exists() + assert not (isolated / "skills").exists() + + auth_copy.write_text('{"token":"mutated"}', encoding="utf-8") + assert (host_codex_home / "auth.json").read_text(encoding="utf-8") == '{"token":"secret"}' + auth_copy.unlink() + auth_copy.write_text('{"token":"replacement"}', encoding="utf-8") + assert (host_codex_home / "auth.json").read_text(encoding="utf-8") == '{"token":"secret"}' + finally: + agent._cleanup_login_shell_home() - def test_build_codex_env_without_login_home_leaves_home_alone(self, monkeypatch): + def test_api_key_runtime_does_not_expose_host_auth(self, monkeypatch, tmp_path): monkeypatch.setenv("CODEX_API_KEY", "k") - agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX)) - env = agent._build_codex_env() - assert env is not None - assert "HOME" not in env - assert "ZDOTDIR" not in env - assert "CODEX_HOME" not in env + host_codex_home = tmp_path / "host-codex" + host_codex_home.mkdir() + (host_codex_home / "auth.json").write_text('{"token":"host"}', encoding="utf-8") + monkeypatch.setenv("CODEX_HOME", str(host_codex_home)) + agent = self._agent_with_prepend([]) + agent._setup_login_shell_home() + + try: + assert not (agent._codex_home() / "auth.json").exists() + finally: + agent._cleanup_login_shell_home() + + def test_host_auth_copy_does_not_attempt_symlink(self, monkeypatch, tmp_path): + monkeypatch.delenv("CODEX_API_KEY", raising=False) + host_codex_home = tmp_path / "host-codex" + host_codex_home.mkdir() + (host_codex_home / "auth.json").write_text('{"token":"host"}', encoding="utf-8") + monkeypatch.setenv("CODEX_HOME", str(host_codex_home)) + monkeypatch.setattr( + Path, + "symlink_to", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("auth must never be symlinked")), + ) + agent = self._agent_with_prepend([]) + agent._setup_login_shell_home() + + try: + exposed = agent._codex_home() / "auth.json" + assert exposed.is_file() + assert not exposed.is_symlink() + assert exposed.read_text(encoding="utf-8") == '{"token":"host"}' + finally: + agent._cleanup_login_shell_home() + + @pytest.mark.skipif( + os.name == "nt" or not hasattr(os, "symlink"), + reason="host auth symlink compatibility requires POSIX symlink support", + ) + def test_symlinked_host_auth_is_resolved_but_isolated_auth_is_still_a_copy(self, monkeypatch, tmp_path): + monkeypatch.delenv("CODEX_API_KEY", raising=False) + credential_store = tmp_path / "credential-store" + credential_store.mkdir() + source = credential_store / "codex-auth.json" + source.write_text('{"token":"host"}', encoding="utf-8") + host_codex_home = tmp_path / "host-codex" + host_codex_home.mkdir() + (host_codex_home / "auth.json").symlink_to(source) + monkeypatch.setenv("CODEX_HOME", str(host_codex_home)) + agent = self._agent_with_prepend([]) + agent._setup_login_shell_home() + + try: + auth_copy = agent._codex_home() / "auth.json" + assert auth_copy.read_text(encoding="utf-8") == '{"token":"host"}' + assert not auth_copy.is_symlink() + finally: + agent._cleanup_login_shell_home() + + def test_same_name_user_skill_is_absent_but_task_local_skill_remains(self, monkeypatch, tmp_path): + """The app-server HOME must not expose a stale user skill, while the + task-local .agents/skills link remains Codex-discoverable from cwd.""" + self._force_posix(monkeypatch) + host_home = tmp_path / "host-home" + stale = host_home / ".agents" / "skills" / "guardrails" + stale.mkdir(parents=True) + (stale / "SKILL.md").write_text("stale user skill", encoding="utf-8") + plugin = tmp_path / "plugin" + current = plugin / "guardrails" + current.mkdir(parents=True) + (current / "SKILL.md").write_text("task-local skill", encoding="utf-8") + workspace = tmp_path / "workspace" + workspace.mkdir() + monkeypatch.setenv("HOME", str(host_home)) + + agent = self._agent_with_prepend([]) + agent.working_directory = workspace + agent._setup_login_shell_home() + agent._setup_skills(str(plugin)) + try: + env = agent._build_codex_env() + assert env is not None + isolated_user_skill = Path(env["HOME"]) / ".agents" / "skills" / "guardrails" / "SKILL.md" + task_skill = workspace / ".agents" / "skills" / "guardrails" / "SKILL.md" + assert not isolated_user_skill.exists() + assert task_skill.read_text(encoding="utf-8") == "task-local skill" + finally: + agent._cleanup_login_shell_home() def test_setup_is_rerunnable_and_cleanup_removes_dir(self, monkeypatch, tmp_path): self._force_posix(monkeypatch) @@ -1784,15 +1968,16 @@ def test_setup_is_rerunnable_and_cleanup_removes_dir(self, monkeypatch, tmp_path agent = self._agent_with_prepend(["/sandbox/mocks"]) agent._setup_login_shell_home() - first = agent._login_shell_home + first = agent._runtime_root assert first is not None agent._setup_login_shell_home() # retried start() must not leak the old dir - second = agent._login_shell_home + second = agent._runtime_root assert second is not None assert not first.exists() agent._cleanup_login_shell_home() assert agent._login_shell_home is None + assert agent._runtime_root is None assert not second.exists() def test_failed_profile_write_rolls_back_temp_home(self, monkeypatch, tmp_path): @@ -1821,8 +2006,29 @@ def tracking_mkdtemp(*args, **kwargs): agent._setup_login_shell_home() assert agent._login_shell_home is None + assert agent._runtime_root is None assert created and not Path(created[0]).exists() + def test_failed_runtime_permission_hardening_rolls_back_temp_root(self, monkeypatch): + agent = self._agent_with_prepend([]) + created: list[Path] = [] + real_mkdtemp = tempfile.mkdtemp + + def tracking_mkdtemp(*args, **kwargs): + path = Path(real_mkdtemp(*args, **kwargs)) + created.append(path) + return str(path) + + monkeypatch.setattr("coder_eval.agents.codex_agent.tempfile.mkdtemp", tracking_mkdtemp) + monkeypatch.setattr(Path, "chmod", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("chmod failed"))) + + with pytest.raises(OSError, match="chmod failed"): + agent._setup_login_shell_home() + + assert agent._runtime_root is None + assert agent._login_shell_home is None + assert created and not created[0].exists() + def test_kill_sync_cleans_login_home(self, monkeypatch, tmp_path): """The watchdog's terminal kill path must not leak the temp HOME.""" self._force_posix(monkeypatch) @@ -1831,13 +2037,14 @@ def test_kill_sync_cleans_login_home(self, monkeypatch, tmp_path): agent.codex_client = SimpleNamespace(close=lambda: None) agent._setup_login_shell_home() - home = agent._login_shell_home - assert home is not None + root = agent._runtime_root + assert root is not None agent.kill_sync() assert agent._login_shell_home is None - assert not home.exists() + assert agent._runtime_root is None + assert not root.exists() async def test_start_creates_and_stop_cleans_login_home(self, monkeypatch, tmp_path): """start() must compose the pieces: generated HOME + pinned CODEX_HOME @@ -1860,18 +2067,114 @@ def fake_codex(**kwargs): await agent.start(str(tmp_path), env_path_prepend=["/sandbox/mocks"]) home = agent._login_shell_home + root = agent._runtime_root assert home is not None and (home / ".bash_profile").is_file() + assert root is not None config = captured.get("config") assert config is not None and config.env is not None assert config.env["HOME"] == str(home) assert config.env["ZDOTDIR"] == str(home) assert config.env["CODEX_HOME"] == str(agent._codex_home()) + assert Path(config.env["CODEX_HOME"]).parent == root path_key = next(k for k in config.env if k.upper() == "PATH") assert config.env[path_key].startswith("/sandbox/mocks") await agent.stop() assert agent._login_shell_home is None - assert not home.exists() + assert agent._runtime_root is None + assert not root.exists() + + async def test_windows_app_server_isolates_home_but_tool_policy_restores_host_homes(self, monkeypatch, tmp_path): + import openai_codex + + self._force_posix(monkeypatch, supported=False) + monkeypatch.setattr(CodexAgent, "_is_windows", staticmethod(lambda: True)) + original_home = 'C:\\Users\\Eval "Runner"' + original_profile = "D:\\Profiles\\Eval\\line\nbreak" + monkeypatch.setenv("HOME", original_home) + monkeypatch.setenv("USERPROFILE", original_profile) + monkeypatch.setenv("CODEX_HOME", str(tmp_path / "missing-host-codex")) + monkeypatch.delenv("CODEX_API_KEY", raising=False) + captured: dict = {} + + def fake_codex(**kwargs): + captured.update(kwargs) + return SimpleNamespace(close=lambda: None) + + monkeypatch.setattr(openai_codex, "Codex", fake_codex) + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX)) + await agent.start(str(tmp_path)) + + try: + config = captured["config"] + root = agent._runtime_root + assert root is not None and config.env is not None + assert Path(config.env["HOME"]).parent == root + assert Path(config.env["USERPROFILE"]).parent == root + assert original_home not in config.env.values() + assert original_profile not in config.env.values() + assert config.config_overrides == ( + "shell_environment_policy.set=" + + "{ HOME = " + + json.dumps(original_home) + + ", USERPROFILE = " + + json.dumps(original_profile) + + " }", + ) + finally: + await agent.stop() + + async def test_retried_start_closes_old_client_before_removing_its_runtime(self, monkeypatch, tmp_path): + import openai_codex + + self._force_posix(monkeypatch) + monkeypatch.setenv("CODEX_HOME", str(tmp_path / "missing-host-codex")) + monkeypatch.delenv("CODEX_API_KEY", raising=False) + close_observations: list[bool] = [] + first_root: list[Path] = [] + call_count = 0 + + def fake_codex(**_kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + first_root.append(agent._runtime_root) + return SimpleNamespace(close=lambda: close_observations.append(first_root[0].exists())) + return SimpleNamespace(close=lambda: None) + + monkeypatch.setattr(openai_codex, "Codex", fake_codex) + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX)) + await agent.start(str(tmp_path)) + await agent.start(str(tmp_path)) + + try: + assert close_observations == [True] + assert not first_root[0].exists() + finally: + await agent.stop() + + async def test_failed_start_cleans_runtime_root(self, monkeypatch, tmp_path): + import openai_codex + + self._force_posix(monkeypatch) + created: list[Path] = [] + real_mkdtemp = tempfile.mkdtemp + + def tracking_mkdtemp(*args, **kwargs): + path = Path(real_mkdtemp(*args, **kwargs)) + created.append(path) + return str(path) + + monkeypatch.setattr("coder_eval.agents.codex_agent.tempfile.mkdtemp", tracking_mkdtemp) + monkeypatch.setattr(openai_codex, "Codex", lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("boom"))) + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX)) + + with pytest.raises(RuntimeError, match="Failed to initialize Codex client: boom"): + await agent.start(str(tmp_path)) + + assert created and not created[0].exists() + assert agent._runtime_root is None + assert agent._login_shell_home is None @pytest.mark.skipif( os.name != "posix" or not shutil.which("bash"), diff --git a/tests/test_codex_agent_unit.py b/tests/test_codex_agent_unit.py index 7cc78cb..cfaf25f 100644 --- a/tests/test_codex_agent_unit.py +++ b/tests/test_codex_agent_unit.py @@ -47,8 +47,14 @@ def test_grep_maps_to_shell(self): def test_glob_maps_to_shell(self): assert _CLAUDE_TO_CODEX_TOOL_MAP["Glob"] == "shell" + def test_webfetch_maps_to_web_search(self): + assert _CLAUDE_TO_CODEX_TOOL_MAP["WebFetch"] == "web_search" + + def test_websearch_maps_to_web_search(self): + assert _CLAUDE_TO_CODEX_TOOL_MAP["WebSearch"] == "web_search" + def test_all_tools_mapped(self): - expected_tools = {"Bash", "Write", "Edit", "Read", "Grep", "Glob"} + expected_tools = {"Bash", "Write", "Edit", "Read", "Grep", "Glob", "WebFetch", "WebSearch"} assert expected_tools.issubset(set(_CLAUDE_TO_CODEX_TOOL_MAP.keys())) @@ -104,3 +110,56 @@ def test_command_dispatch_mutates_lists_in_place(self): # A tool_use block was recorded into the open buffer (cut at the next # tokenUsage flush, not here), joinable to the command by tool_id. assert any(b.block_type == "tool_use" and b.tool_use_id == "c1" for b in state.open_blocks) + + def test_web_open_page_start_is_webfetch_with_url(self): + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX, model="gpt-5-codex")) + state, _commands, _messages = self._state(agent) + action = SimpleNamespace(root=SimpleNamespace(type="openPage", url="https://example.com/docs")) + web_root = SimpleNamespace(type="webSearch", id="ws-open", query="", action=action) + + state.on_item_started(_item_notification("item/started", web_root)) + + telemetry = state.open_tools["ws-open"] + assert telemetry.tool_name == "WebFetch" + assert telemetry.parameters == {"url": "https://example.com/docs"} + + def test_web_open_page_completion_upgrades_provisional_websearch(self): + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX, model="gpt-5-codex")) + state, commands, _messages = self._state(agent) + started = SimpleNamespace(type="webSearch", id="ws-late-action", query="", action=None) + completed = SimpleNamespace( + type="webSearch", + id="ws-late-action", + query="", + action=SimpleNamespace( + root=SimpleNamespace(type="openPage", url="https://example.com/late"), + ), + ) + + state.on_item_started(_item_notification("item/started", started)) + provisional = state.open_tools["ws-late-action"] + assert provisional.tool_name == "WebSearch" + + state.on_item_completed(_item_notification("item/completed", completed)) + + assert provisional.tool_name == "WebFetch" + assert provisional.parameters == {"url": "https://example.com/late"} + assert commands[0].tool_name == "WebFetch" + assert commands[0].parameters == {"url": "https://example.com/late"} + + def test_explicit_search_action_remains_websearch_and_preserves_queries(self): + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX, model="gpt-5-codex")) + state, _commands, _messages = self._state(agent) + action = SimpleNamespace( + root=SimpleNamespace(type="search", query="guardrails", queries=["guardrails", "uipath guardrails"]) + ) + web_root = SimpleNamespace(type="webSearch", id="ws-search", query="guardrails", action=action) + + state.on_item_started(_item_notification("item/started", web_root)) + + telemetry = state.open_tools["ws-search"] + assert telemetry.tool_name == "WebSearch" + assert telemetry.parameters == { + "query": "guardrails", + "queries": ["guardrails", "uipath guardrails"], + }