feat(agent): add Codex agent wrapper and ReMe MCP bridge

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]

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"]

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):

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:

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,33 +188,80 @@ 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."""
         if self.app_context is None:
             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] = {}