diff --git a/pyproject.toml b/pyproject.toml index caf23aa0..4ccdb2c6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,19 +49,16 @@ core = [ "rjieba>=0.2.1", "neo4j>=6.2.0", "networkx>=3.4.2", + "openai-codex>=0.1.0b3", ] dev = [ "pre-commit", "pytest>=8.0", "pytest-asyncio>=0.23", ] -benchmark = [ - "portalocker>=2.10.1", -] full = [ "reme-ai[core]", "reme-ai[dev]", - "reme-ai[benchmark]", ] [project.urls] diff --git a/reme/components/agent_wrapper/__init__.py b/reme/components/agent_wrapper/__init__.py index 1f084900..f5c52115 100644 --- a/reme/components/agent_wrapper/__init__.py +++ b/reme/components/agent_wrapper/__init__.py @@ -3,5 +3,6 @@ from .base_agent_wrapper import BaseAgentWrapper from .as_agent_wrapper import AsAgentWrapper from .cc_agent_wrapper import CcAgentWrapper, CcFileSessionStore +from .codex_agent_wrapper import CodexAgentWrapper -__all__ = ["BaseAgentWrapper", "AsAgentWrapper", "CcAgentWrapper", "CcFileSessionStore"] +__all__ = ["BaseAgentWrapper", "AsAgentWrapper", "CcAgentWrapper", "CcFileSessionStore", "CodexAgentWrapper"] diff --git a/reme/components/agent_wrapper/as_agent_wrapper.py b/reme/components/agent_wrapper/as_agent_wrapper.py index bbd6a529..658fcfd0 100644 --- a/reme/components/agent_wrapper/as_agent_wrapper.py +++ b/reme/components/agent_wrapper/as_agent_wrapper.py @@ -267,7 +267,6 @@ async def _build_agent(self, inputs: Any, **kwargs) -> tuple[Agent, Any]: if model is None: raise ValueError("AsAgentWrapper requires a bound as_llm component with a valid model.") - kwargs = self._merged_kwargs(kwargs) self._cleanup_expired_sessions() self._load_tool_env() @@ -427,6 +426,7 @@ def _event_to_chunk(cls, event: Any) -> StreamChunk | None: async def reply_stream(self, inputs: Any, **kwargs) -> AsyncGenerator[StreamChunk, None]: """Stream agent events as unified StreamChunk objects.""" + kwargs = self._merged_stream_kwargs(kwargs) agent, inputs = await self._build_agent(inputs, **kwargs) async for event in agent.reply_stream(inputs): diff --git a/reme/components/agent_wrapper/base_agent_wrapper.py b/reme/components/agent_wrapper/base_agent_wrapper.py index fd4ec2ae..9267f05e 100644 --- a/reme/components/agent_wrapper/base_agent_wrapper.py +++ b/reme/components/agent_wrapper/base_agent_wrapper.py @@ -64,11 +64,18 @@ def project_skills_root(self) -> Path: def set_output_schema(self, schema: dict | type[BaseModel]) -> "BaseAgentWrapper": """Set a JSON schema for structured output. Accepts dict or BaseModel class. Returns self for chaining.""" - if isinstance(schema, type) and issubclass(schema, BaseModel): - schema = schema.model_json_schema() - self.kwargs["output_schema"] = schema + self.kwargs["output_schema"] = self._normalize_output_schema(schema) return self + @staticmethod + def _normalize_output_schema(schema: Any) -> dict | None: + """Return a JSON-serializable output schema shared by every backend.""" + if isinstance(schema, type) and issubclass(schema, BaseModel): + return schema.model_json_schema() + if schema is None or isinstance(schema, dict): + return schema + raise TypeError("output_schema must be a JSON schema dict or BaseModel class") + def _resolve_job_tools(self, job_tools: list[str]) -> list["BaseJob"]: """Resolve job name strings to BaseJob instances via app_context.""" if not job_tools: @@ -84,7 +91,17 @@ def _resolve_job_tools(self, job_tools: list[str]) -> list["BaseJob"]: def _merged_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]: """Merge component defaults with call-time kwargs; call-time values win.""" - return {**self.kwargs, **kwargs} + merged = {**self.kwargs, **kwargs} + if "output_schema" in merged: + merged["output_schema"] = self._normalize_output_schema(merged["output_schema"]) + return merged + + def _merged_stream_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]: + """Merge stream options and reject unsupported structured output.""" + merged = self._merged_kwargs(kwargs) + if merged.get("output_schema") is not None: + raise NotImplementedError("Structured output is not supported by reply_stream()") + return merged @staticmethod def _chunk(chunk_type: ChunkEnum = ChunkEnum.CONTENT, **kwargs: Any) -> StreamChunk: diff --git a/reme/components/agent_wrapper/cc_agent_wrapper.py b/reme/components/agent_wrapper/cc_agent_wrapper.py index f27bd5a0..7402c06d 100644 --- a/reme/components/agent_wrapper/cc_agent_wrapper.py +++ b/reme/components/agent_wrapper/cc_agent_wrapper.py @@ -2,7 +2,6 @@ import json import os -import shutil from collections.abc import AsyncGenerator from dataclasses import asdict from pathlib import Path @@ -134,6 +133,7 @@ class CcAgentWrapper(BaseAgentWrapper): """Agent wrapper backed by Claude Code SDK.""" DEFAULT_DISALLOWED_TOOLS = ["WebSearch"] + SYSTEM_PROMPT_MODES = {"append", "replace"} @staticmethod def _first_non_empty(*values: Any) -> str: @@ -188,6 +188,24 @@ def _claude_code_api_env(self, kwargs: dict[str, Any]) -> dict[str, str]: env["ANTHROPIC_AUTH_TOKEN"] = api_key return env + @classmethod + def _apply_system_prompt_mode(cls, kwargs: dict[str, Any]) -> None: + """Translate the configured prompt mode into Claude SDK semantics.""" + mode = kwargs.pop("system_prompt_mode", "replace") + if mode not in cls.SYSTEM_PROMPT_MODES: + allowed = ", ".join(sorted(cls.SYSTEM_PROMPT_MODES)) + raise ValueError(f"Unknown system_prompt_mode {mode!r}; expected one of: {allowed}") + + if mode == "append" and "system_prompt" in kwargs: + prompt = kwargs["system_prompt"] + if not isinstance(prompt, str): + raise TypeError("system_prompt must be a string when system_prompt_mode='append'") + kwargs["system_prompt"] = { + "type": "preset", + "preset": "claude_code", + "append": prompt, + } + @property def session_path(self) -> Path: """Directory used for persisted Claude Code sessions.""" @@ -195,26 +213,55 @@ def session_path(self) -> Path: return self.workspace_path / "mem_session" return self.workspace_path / self.app_context.app_config.mem_session_dir - def _ensure_claude_skill_dir(self, config_dir: Path) -> None: - """Expose project skills through Claude Code skill discovery locations.""" + def _ensure_claude_skill_dir(self, config_dir: Path, skills: list[str] | str) -> None: + """Add selected project skills to Claude Code discovery locations.""" project_skills = self.project_skills_root - if not project_skills.exists(): + if not project_skills.is_dir(): + return + + if skills == "all": + skill_names = sorted(path.name for path in project_skills.iterdir() if path.is_dir()) + else: + skill_names = list(dict.fromkeys(skills)) + + for skill_name in skill_names: + if not skill_name or Path(skill_name).name != skill_name or skill_name in {".", ".."}: + raise ValueError(f"Invalid skill name: {skill_name!r}") + + sources = { + skill_name: project_skills / skill_name + for skill_name in skill_names + if (project_skills / skill_name).is_dir() + } + if not sources: return for target in (self.project_path / ".claude" / "skills", config_dir / "skills"): - target.parent.mkdir(parents=True, exist_ok=True) try: - if target.exists() or target.is_symlink(): + # Migrate directory-level links created by older ReMe versions. + if target.is_symlink(): if target.resolve() == project_skills.resolve(): - continue - if target.is_dir() and not target.is_symlink(): - shutil.rmtree(target) - else: target.unlink() + else: + self.logger.warning(f"Preserving existing Claude Code skills link: {target}") + continue + elif target.exists() and not target.is_dir(): + self.logger.warning(f"Preserving existing Claude Code skills path: {target}") + continue - target.symlink_to(project_skills, target_is_directory=True) + target.mkdir(parents=True, exist_ok=True) + for skill_name, source in sources.items(): + skill_target = target / skill_name + if skill_target.is_symlink(): + if skill_target.resolve() != source.resolve(): + self.logger.warning(f"Preserving existing Claude Code skill link: {skill_target}") + continue + if skill_target.exists(): + self.logger.warning(f"Preserving existing Claude Code skill path: {skill_target}") + continue + skill_target.symlink_to(source, target_is_directory=True) except OSError as exc: - self.logger.warning(f"Failed to link Claude Code skills directory {target}: {exc}") + self.logger.warning(f"Failed to link Claude Code skills into {target}: {exc}") @classmethod def _make_tool(cls, job: "BaseJob", tool_context_id: str | None = None): @@ -239,7 +286,7 @@ def _build_options(self, inputs: Any, stream: bool = False, **kwargs) -> Any: from claude_agent_sdk import create_sdk_mcp_server from claude_agent_sdk.types import ClaudeAgentOptions - kwargs = self._merged_kwargs(kwargs) + self._apply_system_prompt_mode(kwargs) skills = kwargs.get("skills") if isinstance(skills, str) and skills != "all": @@ -281,7 +328,7 @@ def _build_options(self, inputs: Any, stream: bool = False, **kwargs) -> Any: claude_config_dir = self.session_path / "claude_config" opts.env.setdefault("CLAUDE_CONFIG_DIR", str(claude_config_dir)) if opts.skills is not None: - self._ensure_claude_skill_dir(claude_config_dir) + self._ensure_claude_skill_dir(claude_config_dir, opts.skills) opts.session_store = opts.session_store or CcFileSessionStore(self.session_path / "claude_code") job_tools: list[str] = kwargs.get("job_tools", []) @@ -293,7 +340,7 @@ def _build_options(self, inputs: Any, stream: bool = False, **kwargs) -> Any: opts.mcp_servers["mcp_server"] = server opts.allowed_tools.extend(job.name for job in resolved_jobs) - if output_schema := kwargs.get("output_schema"): + if (output_schema := kwargs.get("output_schema")) is not None: opts.output_format = {"type": "json_schema", "schema": output_schema} if not isinstance(inputs, str): @@ -492,6 +539,7 @@ def _is_trailing_success_error(exc: Exception) -> bool: async def reply(self, inputs: Any, **kwargs) -> dict: from claude_agent_sdk import query, ResultMessage + kwargs = self._merged_kwargs(kwargs) opts = self._build_options(inputs, stream=False, **kwargs) last_msg = None @@ -507,8 +555,7 @@ async def reply(self, inputs: Any, **kwargs) -> dict: "last_message": asdict(last_msg), "result": last_msg.result, } - output_schema = kwargs.get("output_schema") or self.kwargs.get("output_schema") - if output_schema and last_msg.structured_output: + if kwargs.get("output_schema") is not None: result["structured_output"] = last_msg.structured_output return result @@ -517,6 +564,7 @@ async def reply_stream(self, inputs: Any, **kwargs) -> AsyncGenerator[StreamChun from claude_agent_sdk import query, ResultMessage, AssistantMessage, StreamEvent, UserMessage from claude_agent_sdk.types import RateLimitEvent + kwargs = self._merged_stream_kwargs(kwargs) opts = self._build_options(inputs, stream=True, **kwargs) block_ids: dict[int, str] = {} diff --git a/reme/components/agent_wrapper/codex_agent_wrapper.py b/reme/components/agent_wrapper/codex_agent_wrapper.py new file mode 100644 index 00000000..8fcdea37 --- /dev/null +++ b/reme/components/agent_wrapper/codex_agent_wrapper.py @@ -0,0 +1,495 @@ +"""Codex Python SDK backend for the unified agent wrapper.""" + +import asyncio +from collections.abc import AsyncGenerator +from contextlib import suppress +from dataclasses import fields, is_dataclass +from enum import Enum +import hashlib +import json +import os +from pathlib import Path +import sys +import tempfile +from typing import Any + +from .base_agent_wrapper import BaseAgentWrapper +from ..component_registry import R +from ...enumeration import ChunkEnum +from ...schema import StreamChunk +from ...utils.env_utils import load_env + + +@R.register("codex") +class CodexAgentWrapper(BaseAgentWrapper): + """Agent wrapper backed by the Codex Python SDK.""" + + def __init__(self, mcp_config: str | None = None, codex_home: str | Path | None = None, **kwargs): + super().__init__(**kwargs) + self.mcp_config = mcp_config + self._codex_home = codex_home + self._codex: Any | None = None + self._codex_config: Any | None = None + self._client_lock = asyncio.Lock() + self._turn_lock = asyncio.Lock() + self._mcp_snapshot_path: Path | None = None + self._thread_tool_contexts: dict[str, str] = {} + + @staticmethod + def _first_non_empty(*values: Any) -> str: + for value in values: + if isinstance(value, str) and value: + return value + return "" + + def _default_llm_credential(self) -> dict[str, Any]: + if self.app_context is None: + return {} + from ...enumeration import ComponentEnum + + llm_configs = self.app_context.app_config.components.get(ComponentEnum.AS_LLM) + if not isinstance(llm_configs, dict): + return {} + default_llm = llm_configs.get("default") + credential = getattr(default_llm, "credential", None) + return credential if isinstance(credential, dict) else {} + + @property + def session_path(self) -> Path: + """Directory used for Codex state and persisted threads.""" + if self._codex_home: + path = Path(self._codex_home).expanduser() + return path if path.is_absolute() else self.workspace_path / path + if self.app_context is None: + return self.workspace_path / "mem_session" / "codex" + return self.workspace_path / self.app_context.app_config.mem_session_dir / "codex" + + @property + def wrapper_session_path(self) -> Path: + """ReMe-owned session data, kept separate from a shared OAuth CODEX_HOME.""" + if self.app_context is None: + return self.workspace_path / "mem_session" / "codex" + return self.workspace_path / self.app_context.app_config.mem_session_dir / "codex" + + def _ensure_skills(self, skills: list[str] | str | None) -> None: + """Expose selected project skills through Codex's repo-level directory.""" + if skills is None: + return + if skills == "all": + if not self.project_skills_root.is_dir(): + raise FileNotFoundError(f"Project skills directory not found: {self.project_skills_root}") + names = sorted( + path.name + for path in self.project_skills_root.iterdir() + if path.is_dir() and (path / "SKILL.md").is_file() + ) + else: + names = [skills] if isinstance(skills, str) else list(skills) + names = list(dict.fromkeys(names)) + + target_root = self.workspace_path / ".agents" / "skills" + target_root.mkdir(parents=True, exist_ok=True) + for name in names: + if not name or Path(name).name != name or name in {".", ".."}: + raise ValueError(f"Invalid skill name: {name!r}") + source = self.project_skills_root / name + if not source.is_dir(): + raise FileNotFoundError(f"Skill directory not found: {source}") + if not (source / "SKILL.md").is_file(): + raise FileNotFoundError(f"Skill '{name}' is missing SKILL.md: {source}") + + target = target_root / name + if target.is_symlink(): + if target.resolve() == source.resolve(): + continue + raise FileExistsError(f"Codex skill conflict: {target} points to {target.resolve(strict=False)}") + if target.exists(): + raise FileExistsError(f"Codex skill conflict: {target} already exists and was preserved") + relative_source = os.path.relpath(source, target.parent) + target.symlink_to(relative_source, target_is_directory=True) + + def _explicit_mcp_config(self, kwargs: dict[str, Any]) -> str | None: + value = kwargs.get("mcp_config") if "mcp_config" in kwargs else self.mcp_config + if value is None: + return None + source = Path(str(value)).expanduser() + if source.suffix in {".yaml", ".yml", ".json"}: + if not source.is_absolute(): + source = self.workspace_path / source + return str(source.absolute()) + return str(value) + + def _effective_config_snapshot(self) -> Path: + """Create one private snapshot that remains valid for the client lifetime.""" + if self._mcp_snapshot_path is not None: + return self._mcp_snapshot_path + if self.app_context is None: + raise RuntimeError("Cannot snapshot MCP config without an app_context") + snapshot_dir = self.wrapper_session_path / "reme-mcp" + snapshot_dir.mkdir(parents=True, exist_ok=True) + fd, raw_path = tempfile.mkstemp(prefix="config-", suffix=".json", dir=snapshot_dir) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(self.app_context.app_config.model_dump(mode="json"), stream) + except BaseException: + with suppress(OSError): + os.close(fd) + Path(raw_path).unlink(missing_ok=True) + raise + self._mcp_snapshot_path = Path(raw_path) + return self._mcp_snapshot_path + + def _mcp_config_source(self, kwargs: dict[str, Any]) -> str: + return self._explicit_mcp_config(kwargs) or str(self._effective_config_snapshot()) + + def _build_client_config(self, kwargs: dict[str, Any]): + from openai_codex import CodexConfig + + credential = kwargs.get("credential") if isinstance(kwargs.get("credential"), dict) else {} + default_credential = self._default_llm_credential() + api_key = self._first_non_empty( + kwargs.get("api_key"), + credential.get("api_key"), + os.getenv("CODEX_API_KEY"), + os.getenv("OPENAI_API_KEY"), + os.getenv("LLM_API_KEY"), + default_credential.get("api_key"), + ) + base_url = self._first_non_empty( + kwargs.get("base_url"), + credential.get("base_url"), + os.getenv("CODEX_BASE_URL"), + os.getenv("OPENAI_BASE_URL"), + os.getenv("LLM_BASE_URL"), + default_credential.get("base_url"), + ) + + project_env = self.project_path / ".env" + env = load_env(project_env) if project_env.exists() else load_env() + self.session_path.mkdir(parents=True, exist_ok=True) + env["CODEX_HOME"] = str(self.session_path) + if api_key: + env["OPENAI_API_KEY"] = api_key + + overrides = list(kwargs.get("config_overrides") or []) + if base_url: + overrides.append(f"openai_base_url={json.dumps(base_url)}") + return CodexConfig( + codex_bin=kwargs.get("codex_bin"), + config_overrides=tuple(overrides), + cwd=str(self.cwd), + env=env, + client_name="reme", + client_title="ReMe", + ) + + def _mcp_server_config(self, kwargs: dict[str, Any]) -> dict[str, Any] | None: + from ..job import BackgroundJob, StreamJob + + job_names = list(kwargs.get("job_tools") or []) + if not job_names: + return None + jobs = self._resolve_job_tools(job_names) + unsupported = [job.name for job in jobs if isinstance(job, (BackgroundJob, StreamJob))] + if unsupported: + raise TypeError(f"Codex job_tools must be non-stream request jobs: {', '.join(unsupported)}") + + config_source = self._mcp_config_source(kwargs) + return { + "command": sys.executable, + "args": [ + "-m", + "reme.components.agent_wrapper.codex_mcp_server", + "--config", + config_source, + "--workspace", + str(self.workspace_path), + "--jobs", + json.dumps(job_names), + "--tool-context-id", + str(kwargs.get("tool_context_id") or ""), + ], + "cwd": str(self.project_path), + "required": True, + "enabled_tools": job_names, + "startup_timeout_sec": kwargs.get("mcp_startup_timeout", 30), + "tool_timeout_sec": kwargs.get("mcp_tool_timeout", 300), + } + + def _thread_config(self, kwargs: dict[str, Any]) -> dict[str, Any] | None: + config = dict(kwargs.get("config") or {}) + if server := self._mcp_server_config(kwargs): + servers = dict(config.get("mcp_servers") or {}) + server_key = hashlib.sha256(json.dumps(server, sort_keys=True).encode()).hexdigest()[:12] + servers[f"reme_jobs_{server_key}"] = server + config["mcp_servers"] = servers + return config or None + + @staticmethod + def _enum(enum_cls: Any, value: Any, default: Any = None) -> Any: + if value is None: + return default + return value if isinstance(value, enum_cls) else enum_cls(value) + + async def _open_thread(self, codex: Any, kwargs: dict[str, Any]): + from openai_codex import ApprovalMode, Sandbox + + resume = kwargs.get("resume") or "" + session_id = kwargs.get("session_id") or "" + if resume and session_id and resume != session_id: + raise ValueError("resume and session_id must identify the same Codex thread") + thread_id = resume or session_id + fork_session = bool(kwargs.get("fork_session", False)) + if fork_session and not thread_id: + raise ValueError("fork_session=True requires resume or session_id") + requested_tool_context = str(kwargs.get("tool_context_id") or "") + if not fork_session and thread_id in self._thread_tool_contexts: + if requested_tool_context != self._thread_tool_contexts[thread_id]: + raise ValueError("tool_context_id cannot change when resuming a Codex thread") + + common = { + "approval_mode": self._enum(ApprovalMode, kwargs.get("approval_mode"), ApprovalMode.auto_review), + "base_instructions": kwargs.get("base_instructions"), + "config": self._thread_config(kwargs), + "cwd": str(self.cwd), + "developer_instructions": kwargs.get("system_prompt"), + "model": kwargs.get("model"), + "model_provider": kwargs.get("model_provider"), + "sandbox": self._enum(Sandbox, kwargs.get("sandbox"), Sandbox.full_access), + "service_tier": kwargs.get("service_tier"), + } + if fork_session: + thread = await codex.thread_fork(thread_id, ephemeral=kwargs.get("ephemeral"), **common) + elif thread_id: + thread = await codex.thread_resume(thread_id, **common) + else: + thread = await codex.thread_start(ephemeral=kwargs.get("ephemeral"), **common) + self._thread_tool_contexts[thread.id] = requested_tool_context + return thread + + def _turn_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]: + from openai_codex import ApprovalMode, Sandbox + from openai_codex.generated.v2_all import Personality, ReasoningEffort, ReasoningSummary + + return { + "approval_mode": self._enum(ApprovalMode, kwargs.get("approval_mode")), + "cwd": str(self.cwd), + "effort": self._enum(ReasoningEffort, kwargs.get("effort")), + "model": kwargs.get("model"), + "output_schema": kwargs.get("output_schema"), + "personality": self._enum(Personality, kwargs.get("personality")), + "sandbox": self._enum(Sandbox, kwargs.get("sandbox")), + "service_tier": kwargs.get("service_tier"), + "summary": self._enum(ReasoningSummary, kwargs.get("summary")), + } + + async def _get_codex(self, kwargs: dict[str, Any]) -> Any: + """Lazily start one app-server and reject launch-config changes while it is live.""" + from openai_codex import AsyncCodex + + config = self._build_client_config(kwargs) + async with self._client_lock: + if self._codex is not None: + if config != self._codex_config: + raise RuntimeError("Codex client configuration changed; close the wrapper before reconfiguring it") + return self._codex + codex = AsyncCodex(config) + self._codex = codex + self._codex_config = config + return codex + + async def _close(self) -> None: + """Close the persistent app-server and remove its private config snapshot.""" + async with self._client_lock: + codex, self._codex = self._codex, None + self._codex_config = None + self._thread_tool_contexts.clear() + try: + if codex is not None: + await codex.close() + finally: + if self._mcp_snapshot_path is not None: + self._mcp_snapshot_path.unlink(missing_ok=True) + self._mcp_snapshot_path = None + + @classmethod + def _serialize(cls, value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", by_alias=True) + if isinstance(value, Enum): + return value.value + if is_dataclass(value): + return {field.name: cls._serialize(getattr(value, field.name)) for field in fields(value)} + if isinstance(value, dict): + return {key: cls._serialize(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [cls._serialize(item) for item in value] + return value + + async def reply(self, inputs: Any, **kwargs) -> dict: + """Run one Codex turn and return its final response.""" + if not isinstance(inputs, str): + raise NotImplementedError("Only string input is supported for Codex.") + kwargs = self._merged_kwargs(kwargs) + self._ensure_skills(kwargs.get("skills")) + async with self._turn_lock: + codex = await self._get_codex(kwargs) + thread = await self._open_thread(codex, kwargs) + result = await thread.run(inputs, **self._turn_kwargs(kwargs)) + + final_response = result.final_response or "" + response = { + "session_id": thread.id, + "last_message": final_response, + "result": final_response, + "turn": self._serialize(result), + } + if kwargs.get("output_schema") is not None: + try: + response["structured_output"] = json.loads(final_response) + except json.JSONDecodeError as exc: + raise ValueError("Codex returned invalid JSON for the requested output_schema") from exc + return response + + @classmethod + def _item_data(cls, item: Any) -> tuple[Any, str, str]: + item = item.root if hasattr(item, "root") else item + return item, getattr(item, "type", ""), getattr(item, "id", "") + + @classmethod + # pylint: disable=too-many-return-statements + def _event_to_chunks(cls, event: Any, session_id: str) -> list[StreamChunk]: + """Convert one Codex app-server notification to unified stream chunks.""" + method, payload = event.method, event.payload + if method == "turn/started": + return [cls._chunk(ChunkEnum.REPLY_START, session_id=session_id, metadata={"turn_id": payload.turn.id})] + if method == "item/agentMessage/delta": + return [cls._chunk(ChunkEnum.CONTENT, session_id=session_id, block_id=payload.item_id, chunk=payload.delta)] + if method in {"item/reasoning/summaryTextDelta", "item/reasoning/textDelta", "item/plan/delta"}: + return [cls._chunk(ChunkEnum.THINK, session_id=session_id, block_id=payload.item_id, chunk=payload.delta)] + if method in {"item/commandExecution/outputDelta", "item/fileChange/outputDelta"}: + return [ + cls._chunk( + ChunkEnum.TOOL_RESULT, + session_id=session_id, + block_id=payload.item_id, + tool_call_id=payload.item_id, + chunk=payload.delta, + ), + ] + if method == "item/mcpToolCall/progress": + return [ + cls._chunk( + ChunkEnum.TOOL_RESULT, + session_id=session_id, + block_id=payload.item_id, + tool_call_id=payload.item_id, + chunk=payload.message, + ), + ] + if method in {"item/autoApprovalReview/started", "item/autoApprovalReview/completed"}: + action = cls._serialize(payload.action) + review = cls._serialize(payload.review) + review_id = payload.review_id + target_item_id = getattr(payload, "target_item_id", None) + status = "started" if method.endswith("/started") else "completed" + decision_source = cls._serialize(getattr(payload, "decision_source", None)) + return [ + cls._chunk( + ChunkEnum.APPROVAL, + session_id=session_id, + block_id=target_item_id or review_id, + tool_call_id=target_item_id, + chunk=action, + metadata={ + "review_id": review_id, + "status": status, + "review": review, + "decision_source": decision_source, + "turn_id": payload.turn_id, + }, + ), + ] + if method in {"item/started", "item/completed"}: + item, item_type, item_id = cls._item_data(payload.item) + tool_types = {"commandExecution", "fileChange", "mcpToolCall", "dynamicToolCall", "collabAgentToolCall"} + if item_type not in tool_types: + return [] + name = getattr(item, "tool", None) or item_type + chunk_type = ChunkEnum.TOOL_CALL if method == "item/started" else ChunkEnum.TOOL_RESULT + return [ + cls._chunk( + chunk_type, + session_id=session_id, + block_id=item_id, + tool_call_id=item_id, + tool_call_name=name, + chunk=cls._serialize(item), + ), + ] + if method == "thread/tokenUsage/updated": + usage = payload.token_usage.last + data = cls._serialize(usage) + return [ + cls._chunk( + ChunkEnum.USAGE, + session_id=session_id, + chunk=data, + input_tokens=usage.input_tokens, + output_tokens=usage.output_tokens, + ), + ] + if method == "error": + return [ + cls._chunk( + ChunkEnum.ERROR, + session_id=session_id, + chunk=payload.error.message, + metadata={"will_retry": payload.will_retry}, + ), + ] + if method == "turn/completed": + turn = payload.turn + chunks = [] + if getattr(turn, "error", None): + chunks.append(cls._chunk(ChunkEnum.ERROR, session_id=session_id, chunk=turn.error.message)) + chunks.append( + cls._chunk( + ChunkEnum.REPLY_END, + session_id=session_id, + metadata={ + "turn_id": turn.id, + "status": getattr(turn.status, "value", str(turn.status)), + "duration_ms": turn.duration_ms, + }, + ), + ) + return chunks + return [] + + async def reply_stream(self, inputs: Any, **kwargs) -> AsyncGenerator[StreamChunk, None]: + """Stream Codex app-server notifications as unified chunks.""" + if not isinstance(inputs, str): + raise NotImplementedError("Only string input is supported for Codex.") + kwargs = self._merged_stream_kwargs(kwargs) + self._ensure_skills(kwargs.get("skills")) + async with self._turn_lock: + codex = await self._get_codex(kwargs) + thread = await self._open_thread(codex, kwargs) + turn = await thread.turn(inputs, **self._turn_kwargs(kwargs)) + stream = turn.stream() + completed = False + try: + async for event in stream: + if event.method == "turn/completed": + completed = True + for chunk in self._event_to_chunks(event, thread.id): + yield chunk + finally: + if not completed: + try: + await turn.interrupt() + except Exception as exc: # pylint: disable=broad-exception-caught + self.logger.warning(f"Failed to interrupt Codex turn {turn.id}: {exc}") + await stream.aclose() diff --git a/reme/components/agent_wrapper/codex_mcp_server.py b/reme/components/agent_wrapper/codex_mcp_server.py new file mode 100644 index 00000000..16e43704 --- /dev/null +++ b/reme/components/agent_wrapper/codex_mcp_server.py @@ -0,0 +1,97 @@ +"""FastMCP STDIO bridge that exposes selected ReMe jobs to Codex.""" + +import argparse +from contextlib import asynccontextmanager +import json +from pathlib import Path +from typing import Any + +from fastmcp import FastMCP +from fastmcp.exceptions import ToolError +from fastmcp.tools import FunctionTool + +from ...config import resolve_app_config +from ...reme import ReMe + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Expose selected ReMe jobs over FastMCP STDIO") + parser.add_argument("--config", default="default", help="ReMe config name or file path") + parser.add_argument("--workspace", required=True, help="ReMe workspace directory") + parser.add_argument("--jobs", required=True, help="JSON array of ReMe job names") + parser.add_argument("--tool-context-id", default="", help="Context id injected into every job call") + return parser.parse_args() + + +def _load_job_names(raw: str) -> list[str]: + value = json.loads(raw) + if not isinstance(value, list) or not all(isinstance(name, str) and name for name in value): + raise ValueError("--jobs must be a JSON array of non-empty strings") + return value + + +def _make_tool(job: Any, tool_context_id: str) -> FunctionTool: + async def execute_tool(**kwargs): + if tool_context_id: + if "tool_context_id" in kwargs: + raise ToolError("tool_context_id is managed by the Codex agent wrapper") + kwargs["tool_context_id"] = tool_context_id + response = await job(**kwargs) + if not response.success: + raise ToolError(str(response.answer)) + return response.answer + + return FunctionTool( + name=job.name, + description=job.description, + fn=execute_tool, + parameters=job.parameters or {}, + ) + + +def build_server(app: ReMe, job_names: list[str], tool_context_id: str = "") -> FastMCP: + """Build a STDIO server backed by a dedicated ReMe Application.""" + + @asynccontextmanager + async def lifespan(_server): + await app.start() + try: + yield + finally: + await app.close() + + server = FastMCP(name="reme-codex-tools", lifespan=lifespan) + for name in job_names: + job = app.context.jobs.get(name) + if job is None: + raise KeyError(f"Job '{name}' not found") + server.add_tool(_make_tool(job, tool_context_id)) + return server + + +def main() -> None: + """Load ReMe and serve the requested jobs over STDIO.""" + args = _parse_args() + job_names = _load_job_names(args.jobs) + config = resolve_app_config( + config=args.config, + workspace_dir=str(Path(args.workspace).absolute()), + enable_logo=False, + log_to_console=False, + log_to_file=False, + log_config=False, + ) + # The bridge needs ordinary jobs available for nested job references, but + # must not start workspace watchers or cron loops in this short-lived child. + config["jobs"] = { + name: job_config + for name, job_config in (config.get("jobs") or {}).items() + if job_config.get("backend") not in {"background", "cron"} + } + app = ReMe(**config) + server = build_server(app, job_names, args.tool_context_id) + server.run(transport="stdio", show_banner=False) + + +if __name__ == "__main__": + main() diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 8b65573b..517c9d04 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -672,6 +672,20 @@ components: api_key: ${CLAUDE_CODE_API_KEY:-} base_url: ${CLAUDE_CODE_BASE_URL:-https://dashscope.aliyuncs.com/apps/anthropic} permission_mode: bypassPermissions + system_prompt_mode: replace + codex: + backend: codex + model: ${CODEX_MODEL_NAME:-} + api_key: ${CODEX_API_KEY:-} + base_url: ${CODEX_BASE_URL:-} + approval_mode: auto_review + sandbox: full-access + codex_oauth: + backend: codex + model: ${CODEX_MODEL_NAME:-} + codex_home: ${CODEX_HOME:-~/.codex} + approval_mode: auto_review + sandbox: full-access file_graph: default: diff --git a/reme/enumeration/chunk_enum.py b/reme/enumeration/chunk_enum.py index 32678bd5..4e7d6499 100644 --- a/reme/enumeration/chunk_enum.py +++ b/reme/enumeration/chunk_enum.py @@ -6,7 +6,7 @@ class ChunkEnum(str, Enum): """Enumeration of possible chunk categories for stream processing. - Covers both AgentScope and Claude Code SDK streaming protocols: + Covers AgentScope, Claude Code SDK, and Codex streaming protocols: AgentScope events -> ChunkEnum mapping: ReplyStartEvent -> REPLY_START @@ -28,6 +28,9 @@ class ChunkEnum(str, Enum): ToolResultBlock -> TOOL_RESULT ResultMessage -> USAGE + DONE ResultMessage.is_error -> ERROR + + Codex app-server notifications follow the same lifecycle categories; + thread ids are session ids and item ids correlate tool calls/results. """ # Lifecycle markers @@ -42,6 +45,7 @@ class ChunkEnum(str, Enum): # Tool interaction TOOL_CALL = "tool_call" TOOL_RESULT = "tool_result" + APPROVAL = "approval" # Metadata & terminal USAGE = "usage" diff --git a/reme/schema/stream_chunk.py b/reme/schema/stream_chunk.py index 3d006ecb..e74dc6f4 100644 --- a/reme/schema/stream_chunk.py +++ b/reme/schema/stream_chunk.py @@ -10,7 +10,7 @@ class StreamChunk(BaseModel): """A single chunk in a unified streaming response sequence. - Carries full information from both AgentScope and Claude Code SDK + Carries full information from AgentScope, Claude Code SDK, and Codex backends. Optional fields are ``None`` by default so that simple text-only streams (e.g. plain CONTENT deltas) stay lightweight. @@ -20,8 +20,7 @@ class StreamChunk(BaseModel): dict/list for structured data (tool-call JSON, usage stats, etc.). done: Terminal marker. True only for the final DONE chunk. - session_id: Session identifier (AgentScope: agent.state.session_id; - Claude Code: ResultMessage.session_id). + session_id: Backend session identifier (Codex uses its thread id). block_id: Content-block identifier for matching start/delta/end sequences (both backends assign block IDs). tool_call_id: Tool-call identifier for correlating call deltas diff --git a/tests/integration/test_codex_agent_wrapper.py b/tests/integration/test_codex_agent_wrapper.py new file mode 100644 index 00000000..2cc45a11 --- /dev/null +++ b/tests/integration/test_codex_agent_wrapper.py @@ -0,0 +1,241 @@ +"""Opt-in live coverage for the Codex app-server wrapper. + +Run with ``REME_CODEX_INTEGRATION=1``. These tests consume the caller's active +API-key or Codex OAuth account and are intentionally excluded from normal CI. +""" + +# pylint: disable=protected-access + +import asyncio +import os +from pathlib import Path +import subprocess +from types import SimpleNamespace + +import pytest +from pydantic import BaseModel + +from reme.components.agent_wrapper.codex_agent_wrapper import CodexAgentWrapper +from reme.enumeration import ChunkEnum, ComponentEnum +from reme.schema import ApplicationConfig, Response + +pytestmark = pytest.mark.skipif( + os.getenv("REME_CODEX_INTEGRATION") != "1", + reason="set REME_CODEX_INTEGRATION=1 to run live Codex tests", +) + + +class _StructuredResult(BaseModel): + marker: str + + +class _CustomJob: + name = "only_custom" + description = "Return the fixed marker CUSTOM_JOB_OK." + parameters = {"type": "object", "properties": {}, "additionalProperties": False} + + async def __call__(self, **_kwargs): + return Response(answer="CUSTOM_JOB_OK") + + +class _DraftJob: + def __init__(self, name: str, parameters: dict): + self.name = name + self.description = f"Live tool-context contract job: {name}" + self.parameters = parameters + + +def _child_pids(root_pid: int) -> set[int]: + result = subprocess.run( + ["ps", "-axo", "pid=,ppid="], + check=True, + capture_output=True, + text=True, + ) + children: dict[int, list[int]] = {} + for line in result.stdout.splitlines(): + pid, parent = (int(value) for value in line.split()) + children.setdefault(parent, []).append(pid) + found: set[int] = set() + pending = list(children.get(root_pid, [])) + while pending: + pid = pending.pop() + if pid not in found: + found.add(pid) + pending.extend(children.get(pid, [])) + return found + + +@pytest.mark.asyncio +async def test_live_codex_reply_stream_tools_skills_resume_fork_approval_and_close(tmp_path): + """Exercise the live Codex wrapper contract when explicitly enabled.""" + workspace = tmp_path / "workspace" + workspace.mkdir() + project_skill = Path(__file__).resolve().parents[2] / "skills" / "reme_memory" + skills_root = workspace / "skills" + skills_root.mkdir() + (skills_root / "reme_memory").symlink_to(project_skill, target_is_directory=True) + + app_config = ApplicationConfig( + workspace_dir=str(workspace), + mem_session_dir="sessions", + enable_logo=False, + log_to_console=False, + log_to_file=False, + service={"backend": "mcp"}, + jobs={ + "only_custom": { + "backend": "base", + "description": _CustomJob.description, + "parameters": _CustomJob.parameters, + "steps": [], + }, + "add_live_draft": { + "backend": "base", + "description": "Append text to the current tool context.", + "parameters": { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, + "steps": [{"backend": "add_draft_step"}], + }, + "read_live_draft": { + "backend": "base", + "description": "Read text accumulated in the current tool context.", + "parameters": {"type": "object", "properties": {}}, + "steps": [{"backend": "read_all_draft_step"}], + }, + }, + components={ComponentEnum.AS_LLM: {}}, + ) + context = SimpleNamespace( + app_config=app_config, + jobs={ + "only_custom": _CustomJob(), + "add_live_draft": _DraftJob( + "add_live_draft", + { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, + ), + "read_live_draft": _DraftJob( + "read_live_draft", + {"type": "object", "properties": {}}, + ), + }, + ) + codex_home = os.getenv("REME_CODEX_HOME") + if codex_home is None and not any(os.getenv(name) for name in ("CODEX_API_KEY", "OPENAI_API_KEY", "LLM_API_KEY")): + codex_home = str(Path.home() / ".codex") + wrapper = CodexAgentWrapper(app_context=context, codex_home=codex_home) + + await wrapper.start() + first = await wrapper.reply( + "Call add_live_draft with text STATE_OK, then reply FIRST_TURN_OK.", + job_tools=["add_live_draft", "read_live_draft"], + tool_context_id="live-context", + ) + assert "FIRST_TURN_OK" in first["last_message"] + + resumed = await wrapper.reply( + "Call read_live_draft. Include its exact result and RESUME_OK in your answer.", + resume=first["session_id"], + job_tools=["add_live_draft", "read_live_draft"], + tool_context_id="live-context", + ) + assert resumed["session_id"] == first["session_id"] + assert "RESUME_OK" in resumed["last_message"] + assert "STATE_OK" in resumed["last_message"] + + isolated = await wrapper.reply( + "Call read_live_draft. If it is empty, reply ISOLATED_OK.", + job_tools=["add_live_draft", "read_live_draft"], + tool_context_id="other-context", + ) + assert "ISOLATED_OK" in isolated["last_message"] + assert "STATE_OK" not in isolated["last_message"] + + resumed_after_isolated_context = await wrapper.reply( + "Call read_live_draft again and include its exact result.", + resume=first["session_id"], + job_tools=["add_live_draft", "read_live_draft"], + tool_context_id="live-context", + ) + assert "STATE_OK" in resumed_after_isolated_context["last_message"] + + forked = await wrapper.reply( + "Reply with exactly FORK_OK.", + resume=first["session_id"], + fork_session=True, + tool_context_id="fork-context", + ) + assert forked["session_id"] != first["session_id"] + assert "FORK_OK" in forked["last_message"] + + structured = await wrapper.reply( + "Return marker STRUCTURED_OK.", + output_schema=_StructuredResult, + ) + assert structured["structured_output"] == {"marker": "STRUCTURED_OK"} + + tool_result = await wrapper.reply( + "Call the only_custom tool once, then include its result in your answer.", + job_tools=["only_custom"], + tool_context_id="tool-context", + ) + assert "CUSTOM_JOB_OK" in tool_result["last_message"] + + skill_result = await wrapper.reply( + "Use the reme_memory skill. Read only its instructions and reply SKILL_OK; do not run its scripts or CLI.", + skills=["reme_memory"], + ) + assert "SKILL_OK" in skill_result["last_message"] + assert (workspace / ".agents" / "skills" / "reme_memory").is_symlink() + + outside_target = tmp_path / "approval-target.txt" + approval_chunks = [ + chunk + async for chunk in wrapper.reply_stream( + f"Try to write the word approved to {outside_target} using a shell command.", + approval_mode="auto_review", + sandbox="workspace-write", + ) + ] + assert approval_chunks[0].chunk_type == ChunkEnum.REPLY_START + assert approval_chunks[-1].chunk_type == ChunkEnum.REPLY_END + assert any(chunk.chunk_type == ChunkEnum.APPROVAL for chunk in approval_chunks) + + codex_proc = wrapper._codex._client._sync._proc + assert codex_proc is not None and codex_proc.poll() is None + child_pids = _child_pids(codex_proc.pid) + await wrapper.close() + await asyncio.sleep(0.2) + + assert codex_proc.poll() is not None + running_pids = { + int(line) + for line in subprocess.run( + ["ps", "-axo", "pid="], + check=True, + capture_output=True, + text=True, + ).stdout.split() + } + assert child_pids.isdisjoint(running_pids) + assert wrapper._codex is None + assert wrapper._mcp_snapshot_path is None + + second_context = SimpleNamespace(app_config=app_config, jobs=context.jobs.copy()) + second_wrapper = CodexAgentWrapper(app_context=second_context, codex_home=codex_home) + await second_wrapper.start() + isolated_application = await second_wrapper.reply( + "Call read_live_draft. If it is empty, reply NEW_APPLICATION_OK.", + job_tools=["read_live_draft"], + tool_context_id="live-context", + ) + await second_wrapper.close() + assert "NEW_APPLICATION_OK" in isolated_application["last_message"] + assert "STATE_OK" not in isolated_application["last_message"] diff --git a/tests/unit/test_cc_agent_wrapper.py b/tests/unit/test_cc_agent_wrapper.py new file mode 100644 index 00000000..671de056 --- /dev/null +++ b/tests/unit/test_cc_agent_wrapper.py @@ -0,0 +1,174 @@ +"""Tests for the Claude Code agent wrapper.""" + +from dataclasses import replace +from pathlib import Path + +import pytest + +from reme.components.agent_wrapper.as_agent_wrapper import AsAgentWrapper +from reme.components.agent_wrapper.cc_agent_wrapper import CcAgentWrapper +from reme.components.application_context import ApplicationContext +from reme.config import resolve_app_config + +# pylint: disable=protected-access + + +def _wrapper(tmp_path: Path) -> CcAgentWrapper: + return CcAgentWrapper(app_context=ApplicationContext(workspace_dir=str(tmp_path))) + + +def _skill_roots(tmp_path: Path) -> tuple[Path, Path]: + return ( + tmp_path / ".claude" / "skills", + tmp_path / "mem_session" / "claude_config" / "skills", + ) + + +def test_ensure_claude_skill_dir_adds_selected_skills_without_replacing_existing(tmp_path): + """Selected workspace skills are added while unrelated Claude skills remain.""" + project_skills = tmp_path / "skills" + (project_skills / "one").mkdir(parents=True) + (project_skills / "two").mkdir() + config_dir = tmp_path / "mem_session" / "claude_config" + + for root in _skill_roots(tmp_path): + existing = root / "existing" + existing.mkdir(parents=True) + (existing / "SKILL.md").write_text("existing", encoding="utf-8") + + _wrapper(tmp_path)._ensure_claude_skill_dir(config_dir, ["one"]) + + for root in _skill_roots(tmp_path): + assert (root / "one").is_symlink() + assert (root / "one").resolve() == (project_skills / "one").resolve() + assert not (root / "two").exists() + assert (root / "existing" / "SKILL.md").read_text(encoding="utf-8") == "existing" + + +def test_ensure_claude_skill_dir_all_adds_each_project_skill(tmp_path): + """The all selector creates child links instead of replacing the skills root.""" + project_skills = tmp_path / "skills" + (project_skills / "one").mkdir(parents=True) + (project_skills / "two").mkdir() + config_dir = tmp_path / "mem_session" / "claude_config" + + _wrapper(tmp_path)._ensure_claude_skill_dir(config_dir, "all") + + for root in _skill_roots(tmp_path): + assert root.is_dir() + assert not root.is_symlink() + assert {path.name for path in root.iterdir()} == {"one", "two"} + + +def test_ensure_claude_skill_dir_migrates_old_directory_link(tmp_path): + """A legacy link to the whole project skills directory is migrated safely.""" + project_skills = tmp_path / "skills" + (project_skills / "one").mkdir(parents=True) + config_dir = tmp_path / "mem_session" / "claude_config" + legacy_root = tmp_path / ".claude" / "skills" + legacy_root.parent.mkdir(parents=True) + legacy_root.symlink_to(project_skills, target_is_directory=True) + + _wrapper(tmp_path)._ensure_claude_skill_dir(config_dir, ["one"]) + + assert legacy_root.is_dir() + assert not legacy_root.is_symlink() + assert (legacy_root / "one").resolve() == (project_skills / "one").resolve() + + +def test_ensure_claude_skill_dir_rejects_paths_as_skill_names(tmp_path): + """Skill selectors cannot escape the project skills directory.""" + (tmp_path / "skills").mkdir() + + with pytest.raises(ValueError, match="Invalid skill name"): + _wrapper(tmp_path)._ensure_claude_skill_dir(tmp_path / "config", ["../outside"]) + + +def test_system_prompt_mode_replace_preserves_current_behavior(tmp_path): + """Replace mode passes a string system prompt directly to the SDK.""" + opts = _wrapper(tmp_path)._build_options( + "hello", + system_prompt="custom prompt", + system_prompt_mode="replace", + ) + + assert opts.system_prompt == "custom prompt" + + +def test_system_prompt_mode_append_uses_claude_code_preset(tmp_path): + """Append mode retains Claude Code's preset and appends the custom prompt.""" + opts = _wrapper(tmp_path)._build_options( + "hello", + system_prompt="custom prompt", + system_prompt_mode="append", + ) + + assert opts.system_prompt == { + "type": "preset", + "preset": "claude_code", + "append": "custom prompt", + } + + +def test_system_prompt_mode_rejects_unknown_value(tmp_path): + """Invalid prompt modes fail with a clear configuration error.""" + with pytest.raises(ValueError, match="Unknown system_prompt_mode"): + _wrapper(tmp_path)._build_options("hello", system_prompt_mode="merge") + + +def test_default_claude_code_system_prompt_mode_is_replace(): + """The built-in configuration preserves the existing replacement behavior.""" + config = resolve_app_config(log_config=False) + + assert config["components"]["agent_wrapper"]["claude_code"]["system_prompt_mode"] == "replace" + + +def test_build_options_accepts_empty_output_schema(tmp_path): + """An empty schema remains a valid structured-output request.""" + opts = _wrapper(tmp_path)._build_options("hello", output_schema={}) + + assert opts.output_format == {"type": "json_schema", "schema": {}} + + +@pytest.mark.asyncio +async def test_reply_preserves_falsy_structured_output(tmp_path, monkeypatch): + """Falsy structured output is returned instead of being discarded.""" + from claude_agent_sdk import ResultMessage + + message = ResultMessage( + subtype="success", + duration_ms=1, + duration_api_ms=1, + is_error=False, + num_turns=1, + session_id="session-1", + result="{}", + structured_output={"placeholder": True}, + ) + + async def query(**_kwargs): + yield replace(message, structured_output={}) + + monkeypatch.setattr("claude_agent_sdk.query", query) + + result = await _wrapper(tmp_path).reply("hello", output_schema={}) + + assert "structured_output" in result + assert result["structured_output"] == {} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "wrapper_factory", + [ + _wrapper, + lambda tmp_path: AsAgentWrapper(as_llm="", app_context=ApplicationContext(workspace_dir=str(tmp_path))), + ], +) +@pytest.mark.parametrize("schema", [{}, {"type": "object"}]) +async def test_reply_stream_rejects_output_schema(tmp_path, wrapper_factory, schema): + """Streaming wrappers reject structured-output schemas consistently.""" + wrapper = wrapper_factory(tmp_path) + + with pytest.raises(NotImplementedError, match="Structured output is not supported"): + await anext(wrapper.reply_stream("hello", output_schema=schema)) diff --git a/tests/unit/test_codex_agent_wrapper.py b/tests/unit/test_codex_agent_wrapper.py new file mode 100644 index 00000000..0dc835c0 --- /dev/null +++ b/tests/unit/test_codex_agent_wrapper.py @@ -0,0 +1,730 @@ +"""Unit tests for the Codex agent wrapper and its FastMCP bridge.""" + +# pylint: disable=missing-class-docstring,missing-function-docstring,protected-access + +import asyncio +from dataclasses import dataclass +import json +from pathlib import Path +import stat +import sys +from types import SimpleNamespace + +import pytest +from openai_codex.generated.v2_all import TokenUsageBreakdown +from pydantic import BaseModel + +from reme.components.agent_wrapper.codex_agent_wrapper import CodexAgentWrapper +from reme.components.agent_wrapper.codex_mcp_server import _load_job_names, _make_tool, build_server +from reme.components.job import BackgroundJob +from reme.config import resolve_app_config +from reme.enumeration import ChunkEnum, ComponentEnum +from reme.schema import ApplicationConfig, Response + + +class _Job: + def __init__(self, name="search"): + self.name = name + self.description = "Search memory" + self.parameters = { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + } + self.calls = [] + + async def __call__(self, **kwargs): + self.calls.append(kwargs) + return Response(answer=f"found:{kwargs['query']}") + + +def _wrapper(tmp_path, **kwargs): + job = _Job() + config = SimpleNamespace( + workspace_dir=str(tmp_path), + mem_session_dir="mem_session", + components={ComponentEnum.AS_LLM: {}}, + model_dump=lambda **_kwargs: { + "workspace_dir": str(tmp_path), + "enable_logo": False, + "log_to_console": False, + "log_to_file": False, + "jobs": {}, + "components": {}, + }, + ) + context = SimpleNamespace(app_config=config, jobs={job.name: job}) + return CodexAgentWrapper(app_context=context, **kwargs), job + + +def test_mcp_config_uses_stdio_bridge_and_selected_jobs(tmp_path): + wrapper, _job = _wrapper(tmp_path, mcp_config="custom.yaml") + + config = wrapper._mcp_server_config( # pylint: disable=protected-access + {"job_tools": ["search"], "tool_context_id": "ctx-1"}, + ) + + assert config["command"] + assert config["enabled_tools"] == ["search"] + assert "reme.components.agent_wrapper.codex_mcp_server" in config["args"] + assert config["args"][config["args"].index("--config") + 1] == str(tmp_path / "custom.yaml") + assert config["args"][config["args"].index("--tool-context-id") + 1] == "ctx-1" + + +def test_thread_config_preserves_other_mcp_servers(tmp_path): + wrapper, _job = _wrapper(tmp_path) + config = wrapper._thread_config( # pylint: disable=protected-access + { + "job_tools": ["search"], + "config": {"mcp_servers": {"docs": {"url": "https://example.test/mcp"}}}, + }, + ) + + assert "docs" in config["mcp_servers"] + assert len(config["mcp_servers"]) == 2 + assert next(name for name in config["mcp_servers"] if name != "docs").startswith("reme_jobs_") + + +def test_mcp_config_rejects_background_jobs(tmp_path): + wrapper, _job = _wrapper(tmp_path) + wrapper.app_context.jobs["watch"] = BackgroundJob(name="watch", app_context=wrapper.app_context) + + with pytest.raises(TypeError, match="non-stream request jobs"): + wrapper._mcp_server_config({"job_tools": ["watch"]}) + + +def test_bridge_tool_injects_tool_context_id(): + async def run(): + job = _Job() + tool = _make_tool(job, "ctx-1") + result = await tool.run({"query": "alpha"}) + assert job.calls == [{"query": "alpha", "tool_context_id": "ctx-1"}] + assert "found:alpha" in str(result.content) + + asyncio.run(run()) + + +def test_bridge_rejects_caller_tool_context_id(): + async def run(): + job = _Job() + tool = _make_tool(job, "ctx-1") + with pytest.raises(Exception, match="managed by the Codex agent wrapper"): + await tool.run({"query": "alpha", "tool_context_id": "caller"}) + + asyncio.run(run()) + + +def test_build_server_registers_only_selected_jobs(): + app = SimpleNamespace( + context=SimpleNamespace(jobs={"one": _Job("one"), "two": _Job("two")}), + start=lambda: None, + close=lambda: None, + ) + + async def run(): + server = build_server(app, ["two"]) + tools = await server.list_tools(run_middleware=False) + assert [tool.name for tool in tools] == ["two"] + + asyncio.run(run()) + + +def test_load_job_names_validates_json_array(): + assert _load_job_names('["one", "two"]') == ["one", "two"] + with pytest.raises(ValueError, match="JSON array"): + _load_job_names('{"one": true}') + + +@pytest.mark.asyncio +async def test_stdio_bridge_starts_and_lists_selected_job(tmp_path): + from fastmcp import Client + from fastmcp.client import StdioTransport + + config_path = tmp_path / "bridge.json" + config_path.write_text( + json.dumps( + { + "service": {"backend": "mcp"}, + "workspace_dir": str(tmp_path / "workspace"), + "jobs": { + "empty": { + "backend": "base", + "description": "Return an empty response", + "parameters": {"type": "object", "properties": {}}, + "steps": [], + }, + }, + }, + ), + encoding="utf-8", + ) + transport = StdioTransport( + command=sys.executable, + args=[ + "-m", + "reme.components.agent_wrapper.codex_mcp_server", + "--config", + str(config_path), + "--workspace", + str(tmp_path / "workspace"), + "--jobs", + '["empty"]', + ], + cwd=str(Path(__file__).resolve().parents[2]), + ) + + async with Client(transport, timeout=10) as client: + tools = await client.list_tools() + + assert [tool.name for tool in tools] == ["empty"] + + +@pytest.mark.asyncio +async def test_stdio_bridge_stdout_is_protocol_clean(tmp_path): + config_path = tmp_path / "bridge.json" + config_path.write_text( + json.dumps( + { + "service": {"backend": "mcp"}, + "workspace_dir": str(tmp_path / "workspace"), + "jobs": { + "empty": { + "backend": "base", + "description": "Empty", + "parameters": {"type": "object", "properties": {}}, + "steps": [], + }, + }, + }, + ), + encoding="utf-8", + ) + proc = await asyncio.create_subprocess_exec( + sys.executable, + "-m", + "reme.components.agent_wrapper.codex_mcp_server", + "--config", + str(config_path), + "--workspace", + str(tmp_path / "workspace"), + "--jobs", + '["empty"]', + cwd=str(Path(__file__).resolve().parents[2]), + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + request = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "1"}, + }, + } + assert proc.stdin is not None and proc.stdout is not None and proc.stderr is not None + proc.stdin.write((json.dumps(request) + "\n").encode()) + await proc.stdin.drain() + first_line = await asyncio.wait_for(proc.stdout.readline(), timeout=10) + message = json.loads(first_line) + assert message["jsonrpc"] == "2.0" + assert message["id"] == 1 + proc.terminate() + await asyncio.wait_for(proc.wait(), timeout=10) + stdout = first_line + await proc.stdout.read() + stderr = (await proc.stderr.read()).decode() + assert b"Loading config" not in stdout + assert b"INFO" not in stdout + assert b"WARNING" not in stdout + assert b"2026-" not in stdout + assert "Failed to parse JSONRPC message" not in stderr + assert "Invalid JSON" not in stderr + + +def test_sdk_responds_to_interactive_approval_server_requests(): + from openai_codex.client import CodexClient + + client = CodexClient() + assert client._default_approval_handler( + "item/commandExecution/requestApproval", + {}, + ) == { # pylint: disable=protected-access + "decision": "accept", + } + assert client._default_approval_handler( + "item/fileChange/requestApproval", + {}, + ) == { # pylint: disable=protected-access + "decision": "accept", + } + + +def test_event_to_chunks_maps_content_usage_and_completion(): + content_event = SimpleNamespace( + method="item/agentMessage/delta", + payload=SimpleNamespace(item_id="item-1", delta="hello"), + ) + usage = TokenUsageBreakdown( + cachedInputTokens=1, + inputTokens=3, + outputTokens=5, + reasoningOutputTokens=2, + totalTokens=8, + ) + usage_event = SimpleNamespace( + method="thread/tokenUsage/updated", + payload=SimpleNamespace(token_usage=SimpleNamespace(last=usage)), + ) + completed_event = SimpleNamespace( + method="turn/completed", + payload=SimpleNamespace( + turn=SimpleNamespace(id="turn-1", status=SimpleNamespace(value="completed"), duration_ms=10, error=None), + ), + ) + + content = CodexAgentWrapper._event_to_chunks(content_event, "thread-1") # pylint: disable=protected-access + usage_chunks = CodexAgentWrapper._event_to_chunks(usage_event, "thread-1") # pylint: disable=protected-access + completed = CodexAgentWrapper._event_to_chunks(completed_event, "thread-1") # pylint: disable=protected-access + + assert content[0].chunk_type == ChunkEnum.CONTENT + assert content[0].chunk == "hello" + assert usage_chunks[0].chunk_type == ChunkEnum.USAGE + assert usage_chunks[0].input_tokens == 3 + assert usage_chunks[0].output_tokens == 5 + assert completed[0].chunk_type == ChunkEnum.REPLY_END + assert completed[0].metadata["status"] == "completed" + + +@dataclass +class _TurnResult: + final_response: str + status: str = "completed" + + +def test_reply_returns_thread_id_and_structured_output(tmp_path, monkeypatch): + wrapper, _job = _wrapper(tmp_path) + + class FakeThread: + id = "thread-1" + + async def run(self, inputs, **kwargs): + assert inputs == "answer" + assert kwargs["output_schema"] == {"type": "object"} + return _TurnResult(final_response=json.dumps({"ok": True})) + + class FakeCodex: + def __init__(self, _config): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, _exc_type, _exc, _tb): + return None + + async def close(self): + return None + + async def thread_start(self, **_kwargs): + return FakeThread() + + monkeypatch.setattr("openai_codex.AsyncCodex", FakeCodex) + monkeypatch.setattr("reme.components.agent_wrapper.codex_agent_wrapper.load_env", lambda *_args: {}) + + result = asyncio.run(wrapper.reply("answer", output_schema={"type": "object"})) + + assert result["session_id"] == "thread-1" + assert result["structured_output"] == {"ok": True} + + +def test_codex_skills_add_all_without_deleting_existing_content(tmp_path): + wrapper, _job = _wrapper(tmp_path) + for name in ("reme_memory", "qwenpaw_memory"): + source = tmp_path / "skills" / name + source.mkdir(parents=True) + (source / "SKILL.md").write_text(f"# {name}", encoding="utf-8") + existing = tmp_path / ".agents" / "skills" / "user_skill" + existing.mkdir(parents=True) + marker = existing / "marker" + marker.write_text("keep", encoding="utf-8") + + wrapper._ensure_skills("all") # pylint: disable=protected-access + + assert marker.read_text(encoding="utf-8") == "keep" + for name in ("reme_memory", "qwenpaw_memory"): + target = tmp_path / ".agents" / "skills" / name + assert target.is_symlink() + assert target.resolve() == (tmp_path / "skills" / name).resolve() + + +def test_codex_skills_support_single_name_and_are_idempotent(tmp_path): + wrapper, _job = _wrapper(tmp_path) + source = tmp_path / "skills" / "one" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("# one", encoding="utf-8") + + wrapper._ensure_skills("one") # pylint: disable=protected-access + wrapper._ensure_skills(["one"]) # pylint: disable=protected-access + + assert (tmp_path / ".agents" / "skills" / "one").resolve() == source.resolve() + + +@pytest.mark.parametrize("kind", ["directory", "external_link"]) +def test_codex_skills_preserve_conflicts(tmp_path, kind): + wrapper, _job = _wrapper(tmp_path) + source = tmp_path / "skills" / "one" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("# one", encoding="utf-8") + target = tmp_path / ".agents" / "skills" / "one" + target.parent.mkdir(parents=True) + if kind == "directory": + target.mkdir() + (target / "marker").write_text("keep", encoding="utf-8") + else: + external = tmp_path / "external" + external.mkdir() + target.symlink_to(external, target_is_directory=True) + + with pytest.raises(FileExistsError, match="Codex skill conflict"): + wrapper._ensure_skills("one") # pylint: disable=protected-access + assert target.exists() + + +@pytest.mark.parametrize("create_dir", [False, True]) +def test_codex_skills_reject_missing_or_invalid_skill(tmp_path, create_dir): + wrapper, _job = _wrapper(tmp_path) + if create_dir: + (tmp_path / "skills" / "missing_manifest").mkdir(parents=True) + name = "missing_manifest" + else: + name = "missing" + with pytest.raises(FileNotFoundError): + wrapper._ensure_skills(name) # pylint: disable=protected-access + + +def test_codex_skills_do_not_modify_codex_home(tmp_path): + codex_home = tmp_path / "codex-home" + marker = codex_home / "skills" / "marker" + marker.parent.mkdir(parents=True) + marker.write_text("keep", encoding="utf-8") + wrapper, _job = _wrapper(tmp_path, codex_home=codex_home) + source = tmp_path / "skills" / "one" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("# one", encoding="utf-8") + + wrapper._ensure_skills("one") # pylint: disable=protected-access + + assert marker.read_text(encoding="utf-8") == "keep" + + +def test_effective_mcp_config_snapshot_is_private_and_removed_on_close(tmp_path): + wrapper, _job = _wrapper(tmp_path) + config = wrapper._mcp_server_config({"job_tools": ["search"]}) # pylint: disable=protected-access + snapshot = Path(config["args"][config["args"].index("--config") + 1]) + assert snapshot.exists() + assert stat.S_IMODE(snapshot.stat().st_mode) == 0o600 + assert json.loads(snapshot.read_text(encoding="utf-8"))["workspace_dir"] == str(tmp_path) + + async def close_started_wrapper(): + await wrapper.start() + await wrapper.close() + + asyncio.run(close_started_wrapper()) + assert not snapshot.exists() + + +@pytest.mark.asyncio +async def test_effective_snapshot_exposes_parent_only_custom_job(tmp_path): + from fastmcp import Client + from fastmcp.client import StdioTransport + + app_config = ApplicationConfig( + workspace_dir=str(tmp_path), + enable_logo=False, + log_to_console=False, + log_to_file=False, + service={"backend": "mcp"}, + jobs={ + "only_custom": { + "backend": "base", + "description": "Parent-only inline job", + "parameters": {"type": "object", "properties": {}}, + "steps": [], + }, + "referenced_helper": { + "backend": "base", + "description": "A normal job that custom jobs may reference.", + "parameters": {"type": "object", "properties": {}}, + "steps": [], + }, + }, + ) + job = _Job("only_custom") + context = SimpleNamespace(app_config=app_config, jobs={"only_custom": job}) + wrapper = CodexAgentWrapper(app_context=context) + server_config = wrapper._mcp_server_config({"job_tools": ["only_custom"]}) # pylint: disable=protected-access + transport = StdioTransport( + command=server_config["command"], + args=server_config["args"], + cwd=server_config["cwd"], + ) + + await wrapper.start() + async with Client(transport, timeout=10) as client: + tools = await client.list_tools() + snapshot = Path(server_config["args"][server_config["args"].index("--config") + 1]) + assert snapshot.exists() + assert set(json.loads(snapshot.read_text(encoding="utf-8"))["jobs"]) == { + "only_custom", + "referenced_helper", + } + await wrapper.close() + + assert [tool.name for tool in tools] == ["only_custom"] + assert not snapshot.exists() + + +@pytest.mark.asyncio +async def test_open_thread_defaults_to_full_access(tmp_path): + from openai_codex import ApprovalMode, Sandbox + + wrapper, _job = _wrapper(tmp_path) + observed = {} + + class FakeCodex: + async def thread_start(self, **kwargs): + observed.update(kwargs) + return SimpleNamespace(id="thread-1") + + await wrapper._open_thread(FakeCodex(), {}) # pylint: disable=protected-access + + assert observed["approval_mode"] == ApprovalMode.auto_review + assert observed["sandbox"] == Sandbox.full_access + + +@pytest.mark.asyncio +async def test_resume_reuses_tool_context_and_rejects_context_change(tmp_path): + wrapper, _job = _wrapper(tmp_path) + + class FakeCodex: + async def thread_start(self, **_kwargs): + return SimpleNamespace(id="thread-1") + + async def thread_resume(self, _thread_id, **_kwargs): + return SimpleNamespace(id="thread-1") + + await wrapper._open_thread(FakeCodex(), {"tool_context_id": "ctx-a"}) # pylint: disable=protected-access + await wrapper._open_thread( # pylint: disable=protected-access + FakeCodex(), + {"resume": "thread-1", "tool_context_id": "ctx-a"}, + ) + with pytest.raises(ValueError, match="cannot change"): + await wrapper._open_thread( # pylint: disable=protected-access + FakeCodex(), + {"resume": "thread-1", "tool_context_id": "ctx-b"}, + ) + + +class _StructuredModel(BaseModel): + ok: bool + + +@pytest.mark.parametrize("schema", [_StructuredModel(ok=True), str]) +def test_output_schema_rejects_instances_and_arbitrary_classes(tmp_path, schema): + wrapper, _job = _wrapper(tmp_path) + with pytest.raises(TypeError, match="JSON schema dict or BaseModel class"): + wrapper._merged_kwargs({"output_schema": schema}) # pylint: disable=protected-access + + +def test_output_schema_normalizes_model_class_and_preserves_dict(tmp_path): + wrapper, _job = _wrapper(tmp_path) + schema = {"type": "object", "properties": {"ok": {"type": "boolean"}}} + + assert ( + wrapper._merged_kwargs({"output_schema": _StructuredModel})["output_schema"] # pylint: disable=protected-access + == _StructuredModel.model_json_schema() + ) + assert ( + wrapper._merged_kwargs({"output_schema": schema})["output_schema"] is schema + ) # pylint: disable=protected-access + + +@pytest.mark.asyncio +async def test_reply_normalizes_schema_and_reuses_persistent_client(tmp_path, monkeypatch): + wrapper, _job = _wrapper(tmp_path) + clients = [] + observed_schemas = [] + close_count = 0 + + class FakeThread: + id = "thread-1" + + async def run(self, _inputs, **kwargs): + observed_schemas.append(kwargs["output_schema"]) + return _TurnResult(final_response=json.dumps({"ok": True})) + + class FakeCodex: + def __init__(self, _config): + clients.append(self) + + async def __aenter__(self): + return self + + async def close(self): + nonlocal close_count + close_count += 1 + + async def thread_start(self, **_kwargs): + return FakeThread() + + monkeypatch.setattr("openai_codex.AsyncCodex", FakeCodex) + monkeypatch.setattr("reme.components.agent_wrapper.codex_agent_wrapper.load_env", lambda *_args: {}) + + await wrapper.start() + result = await wrapper.reply("first", output_schema=_StructuredModel) + await wrapper.close() + + assert result["structured_output"] == {"ok": True} + assert observed_schemas == [_StructuredModel.model_json_schema()] + assert len(clients) == 1 + assert close_count == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("schema", [{}, _StructuredModel]) +async def test_reply_stream_rejects_output_schema(tmp_path, schema): + wrapper, _job = _wrapper(tmp_path) + + with pytest.raises(NotImplementedError, match="Structured output is not supported"): + await anext(wrapper.reply_stream("answer", output_schema=schema)) + + +@pytest.mark.asyncio +async def test_reply_stream_interrupts_turn_when_consumer_closes_early(tmp_path, monkeypatch): + wrapper, _job = _wrapper(tmp_path) + stream_closed = False + interrupt_count = 0 + + class FakeTurn: + id = "turn-1" + + async def stream(self): + nonlocal stream_closed + try: + yield SimpleNamespace( + method="turn/started", + payload=SimpleNamespace(turn=SimpleNamespace(id=self.id)), + ) + await asyncio.Event().wait() + finally: + stream_closed = True + + async def interrupt(self): + nonlocal interrupt_count + interrupt_count += 1 + + class FakeThread: + id = "thread-1" + + async def turn(self, _inputs, **_kwargs): + return FakeTurn() + + async def get_codex(_kwargs): + return SimpleNamespace() + + async def open_thread(_codex, _kwargs): + return FakeThread() + + monkeypatch.setattr(wrapper, "_get_codex", get_codex) + monkeypatch.setattr(wrapper, "_open_thread", open_thread) + + stream = wrapper.reply_stream("answer") + first = await anext(stream) + assert first.chunk_type == ChunkEnum.REPLY_START + await stream.aclose() + + assert interrupt_count == 1 + assert stream_closed + + +@pytest.mark.asyncio +async def test_persistent_client_rejects_launch_config_changes(tmp_path, monkeypatch): + wrapper, _job = _wrapper(tmp_path) + + class FakeCodex: + def __init__(self, _config): + pass + + async def close(self): + return None + + monkeypatch.setattr("openai_codex.AsyncCodex", FakeCodex) + monkeypatch.setattr("reme.components.agent_wrapper.codex_agent_wrapper.load_env", lambda *_args: {}) + + await wrapper.start() + first = await wrapper._get_codex({"api_key": "one"}) # pylint: disable=protected-access + assert await wrapper._get_codex({"api_key": "one"}) is first # pylint: disable=protected-access + with pytest.raises(RuntimeError, match="configuration changed"): + await wrapper._get_codex({"api_key": "two"}) # pylint: disable=protected-access + await wrapper.close() + + +@pytest.mark.parametrize("review_status", ["approved", "denied"]) +def test_event_to_chunks_maps_approval_started_and_completed(review_status): + action = {"type": "futureApprovalAction", "value": "preserved"} + review = {"status": review_status, "rationale": "policy"} + started = SimpleNamespace( + method="item/autoApprovalReview/started", + payload=SimpleNamespace( + action=action, + review=review, + review_id="review-1", + target_item_id="item-1", + turn_id="turn-1", + ), + ) + completed = SimpleNamespace( + method="item/autoApprovalReview/completed", + payload=SimpleNamespace( + action=action, + review=review, + review_id="review-1", + target_item_id="item-1", + turn_id="turn-1", + decision_source={"type": "guardian"}, + ), + ) + + started_chunk = CodexAgentWrapper._event_to_chunks(started, "thread-1")[0] # pylint: disable=protected-access + completed_chunk = CodexAgentWrapper._event_to_chunks(completed, "thread-1")[0] # pylint: disable=protected-access + + assert started_chunk.chunk_type == ChunkEnum.APPROVAL + assert started_chunk.chunk == action + assert started_chunk.metadata["status"] == "started" + assert completed_chunk.metadata["status"] == "completed" + assert completed_chunk.metadata["review"]["status"] == review_status + assert completed_chunk.metadata["decision_source"] == {"type": "guardian"} + + +def test_codex_home_expands_user_directory(tmp_path): + wrapper, _job = _wrapper(tmp_path, codex_home="~/.codex") + assert wrapper.session_path == Path.home() / ".codex" + + +def test_named_default_mcp_config_remains_supported(tmp_path): + wrapper, _job = _wrapper(tmp_path, mcp_config="default") + assert wrapper._mcp_config_source({}) == "default" # pylint: disable=protected-access + + +def test_default_config_provides_codex_oauth_wrapper(monkeypatch): + monkeypatch.delenv("CODEX_HOME", raising=False) + config = resolve_app_config(log_config=False) + oauth = config["components"]["agent_wrapper"]["codex_oauth"] + codex = config["components"]["agent_wrapper"]["codex"] + assert oauth["backend"] == "codex" + assert oauth["codex_home"] == "~/.codex" + assert oauth["sandbox"] == "full-access" + assert "api_key" not in oauth + assert codex["sandbox"] == "full-access"