diff --git a/.agents/plugins/marketplace.json b/.agents/plugins/marketplace.json new file mode 100644 index 00000000..abfafe00 --- /dev/null +++ b/.agents/plugins/marketplace.json @@ -0,0 +1,12 @@ +{ + "name": "reme-marketplace", + "interface": { "displayName": "ReMe Memory" }, + "plugins": [ + { + "name": "reme", + "source": { "source": "local", "path": "./plugins/codex/reme" }, + "description": "File-native long-term memory for Codex. Recall on demand + automatic background recording via Stop hook.", + "category": "Productivity" + } + ] +} diff --git a/plugins/codex/README.md b/plugins/codex/README.md new file mode 100644 index 00000000..ce8604f2 --- /dev/null +++ b/plugins/codex/README.md @@ -0,0 +1,85 @@ +# ReMe plugin for Codex + +Connect Codex to [ReMe](https://github.com/agentscope-ai/ReMe) — file-native long-term memory +for AI agents. The plugin gives the agent **recall** (read long-term memory) and **records every +session automatically** via a Stop hook. Consolidation of daily notes into long-term `digest/` +knowledge runs server-side in ReMe. + +## What you get + +- **MCP tools** from the `reme` server: `search`, `traverse`, `daily_list`, `frontmatter_read`, + `read`, `auto_memory_codex`, and more. +- **Stop hook** (`hooks/auto_memory.py`) — when a session ends it calls ReMe's server-side + `auto_memory_codex` tool in a detached background process, passing the session id and transcript + path from Codex's hook payload. The server reads the transcript JSONL and records the durable + facts into today's daily note. Recording is fully automatic and asynchronous — the agent never + records by hand, and stopping is never delayed. Best-effort: if the server is down it logs and + gives up silently. +- **Skill** `reme-memory` — recall long-term memory before answering (semantic `search`, topological + `traverse`, state `daily_list`/`frontmatter_read`, then `read` with citations), plus a server + status check. Recording is handled silently by the Stop hook. + +## Deployment model + +The plugin **connects to a shared HTTP MCP server you start once** — it does not spawn ReMe. One +server means one set of background watchers / dream cron across all your Codex windows. + +## Prerequisites + +1. Install ReMe (Python 3.11+): + + ```bash + pip install "reme-ai[core]" + ``` + +2. Configure model credentials in a `.env` (see `example.env`): + + ```bash + EMBEDDING_API_KEY=sk-xxx + EMBEDDING_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 + LLM_API_KEY=sk-xxx + LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 + ``` + +3. Start the ReMe MCP server (one time, leave it running): + + ```bash + reme start service.backend=mcp service.transport=streamable-http + ``` + + It serves `http://127.0.0.1:2333/mcp`. To use a different port, start with + `service.port=` and update the `url` in `.mcp.json` to match. + +## Install the plugin + +```bash +codex plugin marketplace add agentscope-ai/ReMe +``` + +Then start a Codex session, type `/plugins`, find "ReMe Memory" in the marketplace tab, and install +it. Restart Codex, then confirm the `reme` MCP server tools are available (e.g. `search`, +`traverse`). The `reme-memory` skill can then recall memory and report server health. + +### Trust the plugin hook + +Codex skips hooks from non-managed plugins until you explicitly trust them. After installing, +open `/hooks` in Codex, find the **ReMe** Stop hook, review it, and mark it as trusted. Without +this step the automatic background recording is silently disabled — the skill will still be able +to recall memory, but new sessions will not be recorded. + +## Supported platforms + +- **macOS / Linux**: full support. The Stop hook double-forks so recording never blocks shutdown. +- **Windows**: supported with Python 3 on `PATH`. The hook re-spawns itself as a detached + subprocess (`CREATE_NEW_PROCESS_GROUP`) to avoid blocking. + +## Notes + +- The plugin's MCP server URL lives in `plugins/codex/reme/.mcp.json`. Keep it in sync with how you start + ReMe (host/port). The Stop hook reads this same file to find the server (override with `REME_HOST` + / `REME_PORT` env vars). +- The Stop hook needs `python3` on `PATH` (use `python` on Windows). The hook receives + `PLUGIN_ROOT` (and `CLAUDE_PLUGIN_ROOT` as a compat alias). It logs to + `plugins/codex/reme/logs/auto_memory_hook.log`. +- The MCP tool-name prefix (`mcp__reme__…`) may include the server segment depending on your Codex + version; the skill uses the `mcp__reme__*` wildcard so it works either way. diff --git a/plugins/codex/reme/.codex-plugin/plugin.json b/plugins/codex/reme/.codex-plugin/plugin.json new file mode 100644 index 00000000..6195f9e1 --- /dev/null +++ b/plugins/codex/reme/.codex-plugin/plugin.json @@ -0,0 +1,26 @@ +{ + "name": "reme", + "version": "0.1.0", + "description": "File-native long-term memory for Codex, backed by a running ReMe MCP server. A memory skill recalls long-term knowledge on demand; recording is automatic via a Stop hook that records each session in the background.", + "author": { + "name": "EconML team of Alibaba Tongyi Lab", + "email": "jinli.yl@alibaba-inc.com" + }, + "homepage": "https://reme.agentscope.io/", + "repository": "https://github.com/agentscope-ai/ReMe", + "license": "Apache-2.0", + "keywords": ["memory", "reme", "mcp", "long-term-memory", "agent"], + "skills": "./skills/", + "mcpServers": "./.mcp.json", + "hooks": "./hooks/hooks.json", + "interface": { + "displayName": "ReMe Memory", + "shortDescription": "File-native long-term memory for Codex", + "longDescription": "Persistent, file-native memory for Codex. Recall past conversations, preferences, and decisions on demand; every session is automatically recorded in the background via a Stop hook.", + "developerName": "EconML team of Alibaba Tongyi Lab", + "category": "Productivity", + "capabilities": ["Read", "Write"], + "websiteURL": "https://reme.agentscope.io/", + "brandColor": "#10A37F" + } +} diff --git a/plugins/codex/reme/.mcp.json b/plugins/codex/reme/.mcp.json new file mode 100644 index 00000000..b5d271ee --- /dev/null +++ b/plugins/codex/reme/.mcp.json @@ -0,0 +1,8 @@ +{ + "mcpServers": { + "reme": { + "type": "http", + "url": "http://127.0.0.1:2333/mcp" + } + } +} diff --git a/plugins/codex/reme/hooks/auto_memory.py b/plugins/codex/reme/hooks/auto_memory.py new file mode 100644 index 00000000..151d14dc --- /dev/null +++ b/plugins/codex/reme/hooks/auto_memory.py @@ -0,0 +1,290 @@ +#!/usr/bin/env python3 +"""ReMe Stop hook: fire-and-forget auto-memory for the current session. + +Codex runs this on the ``Stop`` event and feeds the hook payload as JSON +on stdin. We read ``session_id`` and ``transcript_path`` from it and hand those +to ReMe's server-side ``auto_memory_codex`` tool over the (already-running) MCP +server — the server reads the transcript from the given path and records the +durable facts. No messages are sent from here; the agent never has to record by +hand. + +The actual run spins up an inner agent and can take a while, so we detach +(double-fork on Unix / CREATE_NEW_PROCESS_GROUP on Windows) and return +immediately: stopping is never blocked. Any failure is logged, never +surfaced — recording is best-effort. +""" + +from __future__ import annotations + +# Many small with-statements in this script; reusing file-handle names is fine. +# pylint: disable=redefined-outer-name + +import json +import os +import sys +import urllib.error +import urllib.request +from datetime import datetime + +# auto_memory drives an inner agent; give it room. The foreground process has +# already returned by the time this matters (we are detached), so a long ceiling +# is harmless. +_CALL_TIMEOUT = 600 + + +def _plugin_root() -> str: + """Return the plugin install root. + + Codex sets ``PLUGIN_ROOT`` (and ``CLAUDE_PLUGIN_ROOT`` as a compat alias). + """ + return ( + os.environ.get("PLUGIN_ROOT") + or os.environ.get("CLAUDE_PLUGIN_ROOT") + or os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + ) + + +def _extract_text(result: dict | None) -> str: + """Return the human-readable text from an MCP JSON-RPC response. + + Works for both ``{"error": {...}}`` and ``{"result": {"content": [...]}}``. + """ + if result is None: + return "" + if "error" in result: + err = result["error"] + if isinstance(err, dict): + return err.get("message", json.dumps(err, ensure_ascii=False)) + return str(err) + r = result.get("result", {}) if isinstance(result, dict) else {} + content = r.get("content", []) if isinstance(r, dict) else [] + if isinstance(content, list): + for block in content: + if isinstance(block, dict) and block.get("type") == "text": + return (block.get("text") or "").strip() + return "" + + +def _result_status(result: dict | None) -> str: + """Classify a JSON-RPC tool result for logging. + + Returns one of: ``ok``, ``skipped``, ``error``, ``no-response``. + """ + if result is None: + return "no-response" + if "error" in result: + return "error" + r = result.get("result", {}) if isinstance(result, dict) else {} + if isinstance(r, dict) and r.get("isError"): + return "error" + if _extract_text(result).startswith("Skipped"): + return "skipped" + return "ok" + + +def _error_detail(result: dict) -> str: + return _extract_text(result) + + +def _server_url() -> str: + """ReMe MCP endpoint. Prefer the bundled .mcp.json so it stays in sync.""" + mcp_json = os.path.join(_plugin_root(), ".mcp.json") + try: + with open(mcp_json, encoding="utf-8") as fh: + url = json.load(fh)["mcpServers"]["reme"]["url"] + if url: + return url + except Exception: + pass + host = os.environ.get("REME_HOST", "127.0.0.1") + port = os.environ.get("REME_PORT", "2333") + return f"http://{host}:{port}/mcp" + + +def _log(session_id: str, status: str, detail: str = "") -> None: + try: + log_dir = os.path.join(_plugin_root(), "logs") + os.makedirs(log_dir, exist_ok=True) + stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + line = f"{stamp} session={session_id} {status}" + if detail: + line += f" {detail}" + with open(os.path.join(log_dir, "auto_memory_hook.log"), "a", encoding="utf-8") as fh: + fh.write(line + "\n") + except Exception: + pass + + +def _post(url: str, body: dict, headers: dict) -> "urllib.request.addinfourl": + data = json.dumps(body).encode("utf-8") + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + return urllib.request.urlopen(req, timeout=_CALL_TIMEOUT) + + +def _read_jsonrpc(resp) -> dict | None: + """Return the JSON-RPC envelope from a JSON or text/event-stream response.""" + ctype = resp.headers.get("content-type", "") + body = resp.read().decode("utf-8", "replace") + if "text/event-stream" in ctype: + result = None + for line in body.splitlines(): + line = line.strip() + if not line.startswith("data:"): + continue + try: + obj = json.loads(line[len("data:") :].strip()) + except json.JSONDecodeError: + continue + if isinstance(obj, dict) and ("result" in obj or "error" in obj): + result = obj + return result + try: + return json.loads(body) + except json.JSONDecodeError: + return None + + +def _mcp_call(url: str, tool: str, arguments: dict) -> dict | None: + """Minimal MCP streamable-http client: initialize -> initialized -> tools/call.""" + base = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"} + + # 1. initialize (captures the session id header) + init = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "reme-stop-hook", "version": "1.0"}, + }, + } + with _post(url, init, base) as resp: + mcp_session = resp.headers.get("mcp-session-id") + _read_jsonrpc(resp) + + headers = dict(base) + if mcp_session: + headers["mcp-session-id"] = mcp_session + + # 2. notifications/initialized (no id; 202 with empty body) + try: + with _post(url, {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}, headers) as resp: + resp.read() + except urllib.error.HTTPError: + pass + + # 3. tools/call + call = {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": tool, "arguments": arguments}} + try: + with _post(url, call, headers) as resp: + return _read_jsonrpc(resp) + except urllib.error.HTTPError as exc: + # FastMCP may return non-2xx for errors — read the body to extract + # the JSON-RPC error envelope. + return _read_jsonrpc(exc) + + +def _daemonize() -> None: + """Double-fork + setsid so the (slow) call outlives the hook and is reaped by init.""" + if os.fork() > 0: + os._exit(0) # original process returns -> hook completes, Codex stops + os.setsid() + if os.fork() > 0: + os._exit(0) + devnull = os.open(os.devnull, os.O_RDWR) + for fd in (0, 1, 2): + os.dup2(devnull, fd) + + +def main() -> None: + """Entry point: read the hook payload from stdin and record the session.""" + try: + payload = json.loads(sys.stdin.read() or "{}") + except Exception: + payload = {} + + session_id = payload.get("session_id") or "" + transcript_path = payload.get("transcript_path") or "" + + if not session_id and not transcript_path: + return # nothing to anchor a recording on + + # Detach before the slow agent run. Without fork() (e.g. Windows) we re-spawn + # this same script as a fully detached subprocess. + if hasattr(os, "fork"): + _daemonize() + else: + _spawn_detached(payload) + return + + url = _server_url() + tool = "auto_memory_codex" + arguments = {"transcript_path": transcript_path, "session_id": session_id} + try: + result = _mcp_call(url, tool, arguments) + status = _result_status(result) + if status == "error": + detail = _error_detail(result) + _log(session_id, status, detail[:500]) + elif status == "skipped": + _log(session_id, status, f"transcript_path={transcript_path}") + else: + _log(session_id, status) + except urllib.error.URLError as exc: + # Server not running / unreachable — expected when ReMe isn't started. + _log(session_id, "unreachable", str(exc.reason)) + except Exception as exc: # noqa: BLE001 - best-effort, never surface + _log(session_id, "exception", repr(exc)[:500]) + + +def _spawn_detached(payload: dict) -> None: + """Windows-compatible detachment: re-spawn as a fully detached subprocess.""" + import subprocess + import tempfile + + # Write payload to a temp file so the child can read it. + fd, tmp = tempfile.mkstemp(prefix="reme-hook-", suffix=".json") + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + json.dump(payload, fh) + except Exception: + try: + os.close(fd) + except OSError: + pass + return + + # Fire-and-forget: we intentionally don't wait for the child (no `with`). + # pylint: disable-next=consider-using-with + subprocess.Popen( + [sys.executable, __file__, "--payload-file", tmp], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + creationflags=getattr(subprocess, "DETACHED_PROCESS", 0) | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0), + close_fds=True, + ) + + +if __name__ == "__main__": + # --payload-file is the Windows detach path: the parent already wrote the + # hook payload to a temp file and re-spawned us detached. Parse the payload + # first (before the with-block closes the file), then invoke main with it + # set as stdin. + if len(sys.argv) > 2 and sys.argv[1] == "--payload-file": + payload_file = sys.argv[2] + try: + with open(payload_file, encoding="utf-8") as fh: + payload_raw = fh.read() + except Exception: + payload_raw = None + if payload_raw: + sys.stdin = __import__("io").StringIO(payload_raw) + main() + try: + os.unlink(payload_file) + except OSError: + pass + else: + main() diff --git a/plugins/codex/reme/hooks/hooks.json b/plugins/codex/reme/hooks/hooks.json new file mode 100644 index 00000000..db4772f6 --- /dev/null +++ b/plugins/codex/reme/hooks/hooks.json @@ -0,0 +1,17 @@ +{ + "description": "On stop, record this Codex session into ReMe long-term memory (background, async).", + "hooks": { + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 \"${PLUGIN_ROOT}/hooks/auto_memory.py\"", + "commandWindows": "python \"${PLUGIN_ROOT}/hooks/auto_memory.py\"", + "timeout": 30 + } + ] + } + ] + } +} diff --git a/plugins/codex/reme/skills/reme-memory/SKILL.md b/plugins/codex/reme/skills/reme-memory/SKILL.md new file mode 100644 index 00000000..3c24e50f --- /dev/null +++ b/plugins/codex/reme/skills/reme-memory/SKILL.md @@ -0,0 +1,64 @@ +--- +name: reme-memory +description: Use ReMe as file-native long-term memory in Codex. RECALL — search ReMe before answering questions about past conversations, preferences, project history, or decisions. +--- + +# ReMe Memory + +ReMe is the persistent, file-native memory layer for this agent. It stores conversations and +resources as Markdown files with frontmatter and `[[wikilinks]]`, and consolidates them into +long-term **digest** knowledge. Your job with this plugin is **recall only**. + +**Recording is fully automatic.** A Stop hook fires when the session ends and calls +`auto_memory_codex`, which reads the transcript, extracts durable facts with an LLM, and writes +them into `daily/`. Never call `auto_memory_codex`, `write`, `daily_write`, `edit`, or any other +write tool to record memory yourself. If a user asks you to remember something, just acknowledge +it — the hook will record it automatically when the session ends. + +The recall tools come from the `reme` MCP server (surfaced as `mcp__reme__…`): `search`, `traverse`, +`daily_list`, `frontmatter_read`, `read`. They are only available when the user has the server +running: + +``` +reme start service.backend=mcp service.transport=streamable-http +``` + +If the tools are missing, that server is not running — tell the user the command above instead of +guessing answers. + +## Recall (read long-term memory) + +Before answering questions about previous conversations, user preferences, project history, +decisions, or long-term context, recall from ReMe first. ReMe answers **three independent kinds of +question** — pick the mode the request needs; don't merge them into one call. Durable knowledge +lives under `digest/`, daily notes under `daily/`, external materials under `resource/`. + +1. **Semantic** (default — "what do we know about X?"): `search` with `query=""`, + `limit=5` (optional `min_score`). Hybrid vector + BM25 with one-hop wikilink expansion. +2. **Topological** ("what links to this node?"): `traverse` with `path=""`, `depth=1` + (raise to 2 only when needed), `direction=both` to walk the `[[wikilink]]` graph. +3. **State** ("what exists / what was recorded on ?"): `daily_list` with `date="YYYY-MM-DD"` + (empty = today) to list a day's notes, or `frontmatter_read` with a `path` to inspect one file's + frontmatter — structural lookup, no semantic matching. + +Then `read` the relevant hits by `path` (optionally `start_line`/`end_line`; prefer `digest/` paths +for durable knowledge) to pull the content behind a hit. Cite the workspace-relative paths you used. +If nothing useful comes back, say so plainly rather than guessing. + +## Server status + +To check ReMe is up: call `version` and `health_check`, then summarize the version and the health +snapshot (components, workspace). If the `mcp__reme__…` tools are not available at all, the server +is not running — tell the user to start it with the command above. The plugin connects at +`http://127.0.0.1:2333/mcp`; a different host/port must match the `url` in `plugins/codex/reme/.mcp.json`. + +## Workspace model + +``` +daily/ lightly-processed memory: daily facts, conversation summaries +digest/ long-term consolidated knowledge (what recall mainly surfaces) +resource/ external raw materials +``` + +Consolidation of `daily/` into `digest/` and proactive interest extraction run **server-side** in +the ReMe process (background watchers + dream cron). The plugin does not drive them. diff --git a/reme/components/service/mcp_service.py b/reme/components/service/mcp_service.py index 4de5b94d..08f69e5f 100644 --- a/reme/components/service/mcp_service.py +++ b/reme/components/service/mcp_service.py @@ -1,4 +1,28 @@ -"""MCP service: expose jobs as MCP tools.""" +"""MCP service: expose jobs as MCP tools. + +Channel binding (the `` +push from background steps to a specific Claude Code window) is uniform +across transports: a single `ChannelSink` lives on +`ApplicationContext.metadata["channel_sink"]`, unbound at startup, and any +client calling the `claim_channel` MCP tool binds itself as the recipient +via `fastmcp.server.dependencies.get_context().session`. Last-claim-wins. + +Under stdio (one client per server process) the client should claim once +after init; until then channel events drop silently. Under shared +streamable-http / sse the human picks which window receives events. + +``ChannelSink`` is colocated here because it is the runtime mechanism +behind this service's channel feature — pushes ``notifications/claude/channel`` +frames to the bound MCP session. Lossy by design: not bound → no-op; +``send_message`` raises → log warning, swallow (failed notifications must +not surface as ingest failures). Uses ``ServerSession.send_message`` +(low-level raw frame) instead of ``send_notification`` because the latter +validates against a closed ``ServerNotification`` RootModel union that +does not include ``notifications/claude/channel`` — Pydantic rejects +custom methods. Meta keys are filtered to ``[A-Za-z0-9_]+``: Claude Code +silently drops keys with hyphens / other chars when projecting onto +```` attrs. +""" from typing import TYPE_CHECKING, Any @@ -55,7 +79,9 @@ class MCPService(BaseService): conflicts = sorted(self.injected_job_kwargs.keys() & kwargs.keys()) if conflicts: names = ", ".join(conflicts) - raise ToolError(f"{names} injected by the MCP server and cannot be provided by the caller") + raise ToolError( + f"{names} injected by the MCP server and cannot be provided by the caller", + ) kwargs.update(self.injected_job_kwargs) response = await job(**kwargs) if self.tool_error_on_failure and not response.success: diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 98650c55..86cb049b 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -161,6 +161,27 @@ jobs: steps: - backend: auto_memory_cc_step + auto_memory_codex: + backend: base + description: "Auto-memory (Codex): record a Codex session into a daily note, resolved from its transcript path" + parameters: + type: object + properties: + transcript_path: + type: string + description: "Absolute path to the Codex transcript JSONL file" + session_id: + type: string + description: "Codex session id; used for session linkage" + default: "" + memory_hint: + type: string + description: "optional hint" + required: + - transcript_path + steps: + - backend: auto_memory_codex_step + auto_resource: backend: base description: "Auto-resource: interpret resource files into daily notes" diff --git a/reme/steps/evolve/__init__.py b/reme/steps/evolve/__init__.py index e2ab5fd6..7851c8bf 100644 --- a/reme/steps/evolve/__init__.py +++ b/reme/steps/evolve/__init__.py @@ -3,6 +3,7 @@ from ._evolve import now from .auto_memory import AutoMemoryStep from .auto_memory_cc import AutoMemoryCCStep +from .auto_memory_codex import AutoMemoryCodexStep from .auto_resource import AutoResourceStep from .dream import DreamExtractStep, DreamFinishStep, DreamIntegrateStep, DreamTopicsStep, ProactiveStep @@ -10,6 +11,7 @@ __all__ = [ "now", "AutoMemoryStep", "AutoMemoryCCStep", + "AutoMemoryCodexStep", "AutoResourceStep", "DreamExtractStep", "DreamFinishStep", diff --git a/reme/steps/evolve/auto_memory_codex.py b/reme/steps/evolve/auto_memory_codex.py new file mode 100644 index 00000000..85a0ddeb --- /dev/null +++ b/reme/steps/evolve/auto_memory_codex.py @@ -0,0 +1,365 @@ +"""auto_memory_codex — record a Codex session from its transcript path. + +The Codex plugin's Stop hook provides ``session_id`` and ``transcript_path``. +This step validates the path, loads the JSONL transcript, deduplicates by +payload id (or call_id), renders messages and tool calls, and delegates to +AutoMemoryStep. + +The persisted Codex rollout schema (per the ``toolpath-codex`` parser and +OpenAI's upstream types): + +* Top-level envelope: ``{"timestamp", "type", "payload"}`` +* ``response_item`` is the conversational row; ``payload.type`` discriminates: + + * ``message`` — user/assistant text with ``role`` and ``content`` blocks + (``input_text`` / ``output_text``) + * ``function_call`` — model-invoked tool call: ``name``, ``arguments`` + (JSON string), ``call_id`` + * ``function_call_output`` — tool result: ``call_id``, ``output`` + * ``custom_tool_call`` — custom tool: ``name``, ``input``, ``call_id`` + * ``custom_tool_call_output`` — custom tool result: ``call_id``, ``output`` + * ``reasoning`` — intentionally skipped (private model reasoning) + +* Other top-level types (``session_meta``, ``event_msg``, ``turn_context``, + ``session_state``, ``compacted``) are not conversational — they are skipped. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +from .auto_memory import AutoMemoryStep +from ...components import R +from ...components.agent_wrapper import CcFileSessionStore + +_TOOL_EXCERPT = 200 +_SESSIONS_SUBDIR = "sessions" + + +@R.register("auto_memory_codex_step") +class AutoMemoryCodexStep(AutoMemoryStep): + """Step that reads a Codex JSONL transcript and records new turns.""" + + _STORE_SUBDIR = "codex" + _REME_PROJECT_KEY = "codex" + + async def execute(self): + assert self.context is not None + session_id: str = self.context.get("session_id", "") + transcript_path: str = self.context.get("transcript_path", "") + + if not session_id and not transcript_path: + self.context.response.success = False + self.context.response.answer = "Error: session_id or transcript_path is required" + self.logger.warning( + f"[{self.name}] missing both session_id and transcript_path", + ) + return + + # Resolve and validate the transcript path. + resolved = await self._resolve_transcript_path(transcript_path, session_id) + if resolved is None: + self.context.response.success = False + self.context.response.answer = "Error: could not resolve transcript path" + self.logger.warning( + f"[{self.name}] unresolved transcript session_id={session_id!r} " f"given={transcript_path!r}", + ) + return + transcript_path = resolved + + entries = await self._load_codex_transcript(transcript_path) + new_entries = await self._save_codex_session(session_id, entries) + messages = self._codex_entries_to_messages(new_entries) + self.logger.info( + f"[{self.name}] resolved Codex session session_id={session_id!r} " + f"transcript_path={transcript_path!r} " + f"transcript={len(entries)} new_entries={len(new_entries)} messages={len(messages)}", + ) + self.context["messages"] = messages + await super().execute() + + # Codex owns the transcript; AutoMemoryStep's Msg-history dialog store does not apply. + async def _save_session_messages(self, session_id: str, messages) -> None: # noqa: D401 + return + + def _session_link(self, session_id: str) -> str: + return f"[[{self._session_dir()}/{self._STORE_SUBDIR}/{self._REME_PROJECT_KEY}/{session_id}.jsonl]]" + + # ----- path resolution & validation ------------------------------------- + + @staticmethod + def _codex_sessions_dir() -> Path: + """Codex session storage root (``$CODEX_HOME/sessions``).""" + codex_home = os.environ.get("CODEX_HOME", "~/.codex") + return Path(codex_home).expanduser() / _SESSIONS_SUBDIR + + def _validate_transcript_path(self, path: Path) -> bool: + """Check *path* is under ``$CODEX_HOME/sessions``.""" + try: + path.resolve().relative_to(self._codex_sessions_dir().resolve()) + return True + except ValueError: + return False + + async def _resolve_transcript_path( + self, + transcript_path: str, + session_id: str, + ) -> str | None: + """Resolve transcript path, with fallback search when the path doesn't exist.""" + if not transcript_path: + return None + path = Path(os.path.expanduser(transcript_path)) + if path.is_file(): + if not self._validate_transcript_path(path): + self.logger.warning( + f"[{self.name}] transcript_path outside sessions dir: {path}", + ) + return None + return transcript_path + + self.logger.info( + f"[{self.name}] transcript_path not found, searching for session_id={session_id!r}", + ) + sessions_dir = self._codex_sessions_dir() + if not sessions_dir.is_dir(): + return None + + # Codex names files rollout--.jsonl. + pattern = f"rollout-*-{session_id}.jsonl" if session_id else "*.jsonl" + candidates = sorted( + sessions_dir.glob(f"**/{pattern}"), + key=lambda p: p.stat().st_mtime if p.is_file() else 0, + reverse=True, + ) + for candidate in candidates: + if candidate.is_file(): + self.logger.info( + f"[{self.name}] fallback transcript found at {candidate}", + ) + return str(candidate) + + return None + + # ----- transcript loading ----------------------------------------------- + + async def _load_codex_transcript(self, transcript_path: str) -> list[dict]: + """Read JSONL transcript entries from disk.""" + if not transcript_path: + return [] + path = Path(os.path.expanduser(transcript_path)) + if not path.is_file(): + self.logger.warning( + f"[{self.name}] transcript not found at {transcript_path!r}", + ) + return [] + entries = [] + for line in path.read_text(encoding="utf-8").splitlines(): + if line.strip(): + entries.append(json.loads(line)) + return entries + + # ----- session dedup / store -------------------------------------------- + + @staticmethod + def _entry_dedup_key(entry: dict) -> str | None: + """Return a stable dedup key for a transcript entry. + + Prefers ``payload.id`` or ``payload.call_id``. Falls back to a + content hash so that id-less entries (valid per the Codex schema + where ``Message.id`` is optional) are not silently discarded. + """ + payload = entry.get("payload") if isinstance(entry, dict) else None + if isinstance(payload, dict): + if payload.get("id"): + return f"id:{payload['id']}" + if payload.get("call_id"): + return f"call:{payload['call_id']}" + # Fallback: stable content fingerprint. + return f"hash:{hash(json.dumps(entry, sort_keys=True, ensure_ascii=False))}" + + async def _save_codex_session(self, session_id: str, entries: list[dict]) -> list[dict]: + """Dedup entries and store the increment. + + Dedup keys: ``payload.id`` > ``payload.call_id`` > content hash. + """ + if not session_id: + return [] + store = self._codex_store() + key = {"project_key": self._REME_PROJECT_KEY, "session_id": session_id} + entries = [e for e in entries if isinstance(e, dict)] + existing = await store.load(key) or [] + seen = {self._entry_dedup_key(e) for e in existing if isinstance(e, dict)} + seen.discard(None) + increment = [e for e in entries if self._entry_dedup_key(e) not in seen] + await store.append(key, increment) + return increment + + def _codex_store(self) -> CcFileSessionStore: + root = self.file_store.workspace_path / self._session_dir() / self._STORE_SUBDIR + return CcFileSessionStore(root) + + # ----- rendering: raw Codex entries -> plain agent messages ------------- + + @classmethod + def _codex_entries_to_messages(cls, entries: list[dict]) -> list[dict[str, str]]: + """Render ``response_item`` rows into ``{role, name, content}`` dicts. + + Handles all conversational payload types: ``message`` (user / assistant + text), ``function_call`` / ``custom_tool_call`` (agent tool invocations), + and ``function_call_output`` / ``custom_tool_call_output`` (tool results). + ``reasoning`` payloads are intentionally skipped. + """ + messages: list[dict[str, str]] = [] + for record in entries: + if not isinstance(record, dict): + continue + if record.get("type") != "response_item": + continue + payload = record.get("payload") + if not isinstance(payload, dict): + continue + msg = cls._render_codex_payload(payload) + if msg: + messages.append(msg) + return messages + + @classmethod + def _render_codex_payload(cls, payload: dict) -> dict[str, str] | None: + """Dispatch a single ``response_item`` payload to the correct renderer.""" + ptype = payload.get("type", "") + + if ptype == "message": + return cls._render_message_payload(payload) + if ptype == "function_call": + return cls._render_function_call_payload(payload) + if ptype == "function_call_output": + return cls._render_function_call_output_payload(payload) + if ptype == "custom_tool_call": + return cls._render_custom_tool_call_payload(payload) + if ptype == "custom_tool_call_output": + return cls._render_custom_tool_call_output_payload(payload) + # reasoning, etc. — intentionally skipped + return None + + # -- message ----------------------------------------------------------- + + @classmethod + def _render_message_payload(cls, payload: dict) -> dict[str, str] | None: + """Render a ``message`` payload (user or assistant turn).""" + role = payload.get("role", "") + if role not in ("user", "assistant"): + return None + + content = payload.get("content", "") + text = cls._render_codex_content(content) + if not text: + return None + return {"role": role, "name": role, "content": text} + + # -- function_call ----------------------------------------------------- + + @classmethod + def _render_function_call_payload(cls, payload: dict) -> dict[str, str] | None: + """Render a ``function_call`` payload as a virtual assistant tool-use message. + + ``arguments`` is a JSON string (per the upstream schema), not a dict. + We round-trip through :func:`json.loads` + :func:`json.dumps` for a + compact single-line representation. + """ + name = payload.get("name", "?") + arguments = payload.get("arguments", "") + try: + args = json.dumps(json.loads(arguments), ensure_ascii=False) + except (json.JSONDecodeError, TypeError, ValueError): + args = str(arguments) + if len(args) > _TOOL_EXCERPT: + args = args[:_TOOL_EXCERPT] + "..." + return { + "role": "assistant", + "name": "assistant", + "content": f"[tool {name}({args})]", + } + + # -- function_call_output ---------------------------------------------- + + @classmethod + def _render_function_call_output_payload(cls, payload: dict) -> dict[str, str] | None: + """Render a ``function_call_output`` payload as a virtual tool-result message.""" + output = payload.get("output", "") + excerpt = str(output).strip() + if len(excerpt) > _TOOL_EXCERPT: + excerpt = excerpt[:_TOOL_EXCERPT] + "..." + if not excerpt: + return None + return { + "role": "user", + "name": "user", + "content": f"[tool_result {excerpt}]", + } + + # -- custom_tool_call -------------------------------------------------- + + @classmethod + def _render_custom_tool_call_payload(cls, payload: dict) -> dict[str, str] | None: + """Render a ``custom_tool_call`` payload as a virtual assistant tool-use message. + + ``input`` is a JSON string (per the upstream schema). + """ + name = payload.get("name", "?") + raw_input = payload.get("input", "") + try: + inp = json.dumps(json.loads(raw_input), ensure_ascii=False) + except (json.JSONDecodeError, TypeError, ValueError): + inp = str(raw_input) + if len(inp) > _TOOL_EXCERPT: + inp = inp[:_TOOL_EXCERPT] + "..." + return { + "role": "assistant", + "name": "assistant", + "content": f"[tool {name}({inp})]", + } + + # -- custom_tool_call_output ------------------------------------------- + + @classmethod + def _render_custom_tool_call_output_payload(cls, payload: dict) -> dict[str, str] | None: + """Render a ``custom_tool_call_output`` payload as a virtual tool-result message.""" + output = payload.get("output", "") + excerpt = str(output).strip() + if len(excerpt) > _TOOL_EXCERPT: + excerpt = excerpt[:_TOOL_EXCERPT] + "..." + if not excerpt: + return None + return { + "role": "user", + "name": "user", + "content": f"[tool_result {excerpt}]", + } + + # -- content blocks ---------------------------------------------------- + + @classmethod + def _render_codex_content(cls, content: Any) -> str: + """Render Codex message content blocks to plain text. + + Codex ``ContentPart`` only has ``input_text`` and ``output_text`` + (both carry a ``text`` field). Everything else is non-text and + intentionally skipped. + """ + if isinstance(content, str): + return content.strip() + if not isinstance(content, list): + return "" + + parts: list[str] = [] + for block in content: + if not isinstance(block, dict): + continue + if block.get("type") in ("input_text", "output_text"): + if t := (block.get("text") or "").strip(): + parts.append(t) + return "\n".join(p for p in parts if p).strip() diff --git a/tests/unit/test_auto_memory_codex.py b/tests/unit/test_auto_memory_codex.py new file mode 100644 index 00000000..00a3dc22 --- /dev/null +++ b/tests/unit/test_auto_memory_codex.py @@ -0,0 +1,402 @@ +"""Unit tests for the auto_memory_codex step.""" + +# pylint: disable=missing-class-docstring,missing-function-docstring,protected-access + +import json +from pathlib import Path + +import pytest + +from reme.steps.evolve.auto_memory_codex import AutoMemoryCodexStep + +# ---- helpers --------------------------------------------------------------- + + +def _msg(role: str, text: str | list | None = None, *, pid: str = "m1") -> dict: + """Codex response_item with a ``message`` payload.""" + if text is None: + text = "hello" + content = text if isinstance(text, list) else [{"type": "input_text", "text": text}] + return { + "type": "response_item", + "payload": {"type": "message", "id": pid, "role": role, "content": content}, + } + + +def _fc(name: str, arguments: str = "{}", *, cid: str = "c1") -> dict: + """Codex response_item with a ``function_call`` payload.""" + return { + "type": "response_item", + "payload": { + "type": "function_call", + "id": cid, + "name": name, + "arguments": arguments, + "call_id": cid, + }, + } + + +def _fco(cid: str, output: str) -> dict: + """Codex response_item with a ``function_call_output`` payload.""" + return { + "type": "response_item", + "payload": {"type": "function_call_output", "call_id": cid, "output": output}, + } + + +# ---- _codex_entries_to_messages -------------------------------------------- + + +class TestEntriesToMessages: + def test_user_and_assistant(self): + msgs = AutoMemoryCodexStep._codex_entries_to_messages( + [_msg("user", "hi", pid="1"), _msg("assistant", "hey", pid="2")], + ) + assert [m["role"] for m in msgs] == ["user", "assistant"] + assert msgs[0]["content"] == "hi" + + def test_filters_system_and_developer_roles(self): + msgs = AutoMemoryCodexStep._codex_entries_to_messages( + [_msg("system", pid="1"), _msg("developer", pid="2"), _msg("user", "real", pid="3")], + ) + assert len(msgs) == 1 + assert msgs[0]["content"] == "real" + + def test_skips_non_response_item_rows(self): + msgs = AutoMemoryCodexStep._codex_entries_to_messages( + [{"type": "session_meta", "payload": {"id": "s1"}}, _msg("user", "real", pid="1")], + ) + assert len(msgs) == 1 + + def test_function_call_and_output(self): + msgs = AutoMemoryCodexStep._codex_entries_to_messages( + [_fc("shell", json.dumps({"cmd": "ls"}), cid="c1"), _fco("c1", "file1\nfile2")], + ) + assert len(msgs) == 2 + assert msgs[0]["role"] == "assistant" + assert "[tool shell" in msgs[0]["content"] + assert msgs[1]["role"] == "user" + assert "[tool_result" in msgs[1]["content"] + assert "file1" in msgs[1]["content"] + + def test_custom_tool_call(self): + entry = { + "type": "response_item", + "payload": { + "type": "custom_tool_call", + "id": "c1", + "name": "mcp_search", + "input": '{"q":"x"}', + "call_id": "c1", + }, + } + msgs = AutoMemoryCodexStep._codex_entries_to_messages([entry]) + assert len(msgs) == 1 + assert "mcp_search" in msgs[0]["content"] + + def test_empty_output_skipped(self): + assert not AutoMemoryCodexStep._codex_entries_to_messages([_fco("c1", "")]) + + def test_long_arguments_truncated(self): + entry = _fc("t", json.dumps({"d": "x" * 500}), cid="c1") + msgs = AutoMemoryCodexStep._codex_entries_to_messages([entry]) + assert msgs[0]["content"].endswith("...)]") + + def test_reasoning_skipped(self): + msgs = AutoMemoryCodexStep._codex_entries_to_messages( + [{"type": "response_item", "payload": {"type": "reasoning", "id": "r1"}}, _msg("user", "real", pid="1")], + ) + assert len(msgs) == 1 + assert msgs[0]["content"] == "real" + + def test_realistic_multi_turn_session(self): + """Parse a realistic Codex transcript with mixed entry types.""" + transcript = [ + # session_meta — skipped + { + "timestamp": "2026-07-20T10:00:00.000Z", + "type": "session_meta", + "payload": { + "id": "sess-abc", + "cwd": "/project", + "originator": "codex_exec", + "cli_version": "0.144.0", + "source": "exec", + "timestamp": "2026-07-20T10:00:00.000Z", + }, + }, + # turn_context — skipped + { + "timestamp": "2026-07-20T10:00:01.000Z", + "type": "turn_context", + "payload": { + "turn_id": "turn-1", + "cwd": "/project", + "model": "gpt-5.1", + }, + }, + # user message + { + "timestamp": "2026-07-20T10:00:02.000Z", + "type": "response_item", + "payload": { + "type": "message", + "id": "msg-1", + "role": "user", + "content": [{"type": "input_text", "text": "帮我看看入口文件是什么"}], + }, + }, + # assistant text + { + "timestamp": "2026-07-20T10:00:03.000Z", + "type": "response_item", + "payload": { + "type": "message", + "id": "msg-2", + "role": "assistant", + "content": [{"type": "output_text", "text": "让我查一下。"}], + }, + }, + # shell function_call + { + "timestamp": "2026-07-20T10:00:04.000Z", + "type": "response_item", + "payload": { + "type": "function_call", + "id": "call-1", + "name": "shell", + "arguments": '{"command": "ls *.py"}', + "call_id": "call-1", + }, + }, + # shell output + { + "timestamp": "2026-07-20T10:00:05.000Z", + "type": "response_item", + "payload": { + "type": "function_call_output", + "call_id": "call-1", + "output": "main.py\nutils.py", + }, + }, + # assistant explains result + { + "timestamp": "2026-07-20T10:00:06.000Z", + "type": "response_item", + "payload": { + "type": "message", + "id": "msg-3", + "role": "assistant", + "content": [{"type": "output_text", "text": "入口是 main.py。"}], + }, + }, + # reasoning — skipped + { + "timestamp": "2026-07-20T10:00:07.000Z", + "type": "response_item", + "payload": { + "type": "reasoning", + "id": "reason-1", + "summary": [{"type": "summary_text", "text": "用户想知道入口文件"}], + }, + }, + # mcpToolCall as function_call + { + "timestamp": "2026-07-20T10:00:08.000Z", + "type": "response_item", + "payload": { + "type": "function_call", + "id": "call-2", + "name": "read", + "arguments": '{"path": "daily/2026-07-20.md"}', + "call_id": "call-2", + }, + }, + # MCP tool output + { + "timestamp": "2026-07-20T10:00:09.000Z", + "type": "response_item", + "payload": { + "type": "function_call_output", + "call_id": "call-2", + "output": "# 2026-07-20\n- 讨论了入口文件\n- 上次提到用 Flask", + }, + }, + # final assistant + { + "timestamp": "2026-07-20T10:00:10.000Z", + "type": "response_item", + "payload": { + "type": "message", + "id": "msg-4", + "role": "assistant", + "content": [{"type": "output_text", "text": "根据记忆，上次也讨论过入口文件。"}], + }, + }, + ] + + msgs = AutoMemoryCodexStep._codex_entries_to_messages(transcript) + + # session_meta, turn_context, reasoning are skipped + # 4 messages + 2 function_calls + 2 function_call_outputs = 8 + assert len(msgs) == 8 + + # Verify ordering and types + assert msgs[0] == {"role": "user", "name": "user", "content": "帮我看看入口文件是什么"} + assert msgs[1]["role"] == "assistant" + assert msgs[1]["content"] == "让我查一下。" + assert "[tool shell" in msgs[2]["content"] and "ls *.py" in msgs[2]["content"] + assert msgs[3]["role"] == "user" and "main.py" in msgs[3]["content"] + assert msgs[4]["content"] == "入口是 main.py。" + assert "[tool read" in msgs[5]["content"] + assert "2026-07-20" in msgs[6]["content"] + assert msgs[7]["content"] == "根据记忆，上次也讨论过入口文件。" + + +# ---- _render_codex_content ------------------------------------------------- + + +class TestRenderContent: + def test_input_and_output_text(self): + content = [{"type": "input_text", "text": "hi"}, {"type": "output_text", "text": "there"}] + assert AutoMemoryCodexStep._render_codex_content(content) == "hi\nthere" + + def test_strips_empty_and_skips_non_dict(self): + content = [{"type": "input_text", "text": " "}, "bare", {"type": "input_text", "text": "real"}] + assert AutoMemoryCodexStep._render_codex_content(content) == "real" + + def test_skips_unknown_block_types(self): + content = [{"type": "image", "data": "x"}, {"type": "output_text", "text": "ok"}] + assert AutoMemoryCodexStep._render_codex_content(content) == "ok" + + +# ---- _resolve_transcript_path ---------------------------------------------- + + +class TestResolvePath: + @pytest.mark.asyncio + async def test_valid_path(self, tmp_path, monkeypatch): + sessions = tmp_path / "sessions" + sessions.mkdir() + t = sessions / "rollout-20260720-abc.jsonl" + t.write_text(json.dumps(_msg("user", pid="r1")) + "\n") + monkeypatch.setattr(AutoMemoryCodexStep, "_codex_sessions_dir", staticmethod(lambda: sessions)) + assert await AutoMemoryCodexStep()._resolve_transcript_path(str(t), "abc") == str(t) + + @pytest.mark.asyncio + async def test_outside_sessions_rejected(self, tmp_path, monkeypatch): + sessions = tmp_path / "sessions" + sessions.mkdir() + outside = tmp_path / "outside.jsonl" + outside.write_text("{}") + monkeypatch.setattr(AutoMemoryCodexStep, "_codex_sessions_dir", staticmethod(lambda: sessions)) + assert await AutoMemoryCodexStep()._resolve_transcript_path(str(outside), "abc") is None + + @pytest.mark.asyncio + async def test_fallback_by_session_id(self, tmp_path, monkeypatch): + sessions = tmp_path / "sessions" / "2026" / "07" / "20" + sessions.mkdir(parents=True) + real = sessions / "rollout-1721400000-sess-abc.jsonl" + real.write_text(json.dumps(_msg("user", pid="r1")) + "\n") + monkeypatch.setattr(AutoMemoryCodexStep, "_codex_sessions_dir", staticmethod(lambda: tmp_path / "sessions")) + result = await AutoMemoryCodexStep()._resolve_transcript_path("/gone/stale.jsonl", "sess-abc") + assert result == str(real) + + +# ---- _load_codex_transcript ------------------------------------------------ + + +@pytest.mark.asyncio +async def test_load_jsonl(tmp_path: Path): + t = tmp_path / "r.jsonl" + t.write_text("\n".join(json.dumps(_msg("user", pid=str(i))) for i in range(2))) + result = await AutoMemoryCodexStep()._load_codex_transcript(str(t)) + assert len(result) == 2 + + +# ---- _save_codex_session --------------------------------------------------- + + +@pytest.mark.asyncio +async def test_dedup_by_payload_id(tmp_path, monkeypatch): + step = AutoMemoryCodexStep() + monkeypatch.setattr(step, "_codex_store", lambda: _store(tmp_path / "codex")) + await step._save_codex_session("s1", [_msg("user", pid="1"), _msg("user", pid="2")]) + inc = await step._save_codex_session("s1", [_msg("user", pid="2"), _msg("user", pid="3")]) + assert len(inc) == 1 + assert inc[0]["payload"]["id"] == "3" + + +@pytest.mark.asyncio +async def test_dedup_by_call_id(tmp_path, monkeypatch): + step = AutoMemoryCodexStep() + monkeypatch.setattr(step, "_codex_store", lambda: _store(tmp_path / "codex")) + await step._save_codex_session("s1", [_fco("c1", "o1"), _fco("c2", "o2")]) + inc = await step._save_codex_session("s1", [_fco("c1", "o1-again"), _fco("c3", "o3")]) + assert len(inc) == 1 + assert inc[0]["payload"]["call_id"] == "c3" + + +@pytest.mark.asyncio +async def test_id_less_entries_not_discarded(tmp_path, monkeypatch): + """Entries without payload.id or call_id are kept, using content hash for dedup.""" + step = AutoMemoryCodexStep() + monkeypatch.setattr(step, "_codex_store", lambda: _store(tmp_path / "codex")) + inc = await step._save_codex_session( + "s1", + [ + _msg("user", pid="1"), + {"type": "response_item", "payload": {"type": "message", "role": "user", "content": "no id"}}, + ], + ) + assert len(inc) == 2 # both kept, id-less entry uses content hash + + +@pytest.mark.asyncio +async def test_id_less_message_rendered_and_deduped(tmp_path, monkeypatch): + """An id-less user message survives dedup and is rendered correctly.""" + id_less = { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "remember me"}]}, + } + step = AutoMemoryCodexStep() + monkeypatch.setattr(step, "_codex_store", lambda: _store(tmp_path / "codex")) + + # First save — saved + inc1 = await step._save_codex_session("s1", [id_less]) + assert len(inc1) == 1 + + # Second save with same content — deduped, empty increment + inc2 = await step._save_codex_session("s1", [id_less]) + assert len(inc2) == 0 + + # Rendering still works + msgs = AutoMemoryCodexStep._codex_entries_to_messages([id_less]) + assert len(msgs) == 1 + assert msgs[0]["content"] == "remember me" + + +@pytest.mark.asyncio +async def test_different_id_less_messages_not_collapsed(tmp_path, monkeypatch): + """Two id-less messages with different content get different hash keys, + so the second is not falsely treated as a duplicate.""" + msg_a = { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "I like Go"}]}, + } + msg_b = { + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "I like Rust"}]}, + } + step = AutoMemoryCodexStep() + monkeypatch.setattr(step, "_codex_store", lambda: _store(tmp_path / "codex")) + + inc = await step._save_codex_session("s1", [msg_a, msg_b]) + assert len(inc) == 2 # both kept, different hash keys + + +def _store(root: Path): + from reme.components.agent_wrapper import CcFileSessionStore + + return CcFileSessionStore(root) diff --git a/tests/unit/test_codex_hook_mcp.py b/tests/unit/test_codex_hook_mcp.py new file mode 100644 index 00000000..a7495626 --- /dev/null +++ b/tests/unit/test_codex_hook_mcp.py @@ -0,0 +1,211 @@ +"""Round-trip tests for the hook's MCP HTTP client against a loopback server. + +These exercise the full chain: hook → JSON-RPC over HTTP → response parsing, +without requiring a running ReMe service. +""" + +# pylint: disable=protected-access,missing-function-docstring + +from __future__ import annotations + +import json +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +import pytest + +# Import the hook module from the Codex plugin. +_HOOK_DIR = Path(__file__).parents[2] / "plugins" / "codex" / "reme" / "hooks" +sys.path.insert(0, str(_HOOK_DIR)) +# pylint: disable=wrong-import-position +import auto_memory # noqa: E402 + +# ---- mock MCP server ------------------------------------------------------- + + +class _McpHandler(BaseHTTPRequestHandler): + """Minimal JSON-RPC handler that captures calls and returns canned responses.""" + + calls: list[dict[str, Any]] + session_header: str + tool_result: dict[str, Any] | None + + def _read_body(self) -> dict[str, Any]: + length = int(self.headers.get("Content-Length", "0")) + return json.loads(self.rfile.read(length) or b"{}") if length else {} + + def _send_json(self, status: int, body: dict[str, Any], extra_headers: dict[str, str] | None = None) -> None: + encoded = json.dumps(body).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + if extra_headers: + for k, v in extra_headers.items(): + self.send_header(k, v) + self.end_headers() + self.wfile.write(encoded) + + def do_POST(self) -> None: # noqa: N802 + body = self._read_body() + self.calls.append(body) + method = body.get("method", "") + + if method == "initialize": + self._send_json( + 200, + {"jsonrpc": "2.0", "id": body.get("id"), "result": {}}, + extra_headers={"mcp-session-id": self.session_header}, + ) + elif method == "notifications/initialized": + self._send_json(202, {}) + elif method == "tools/call": + if self.tool_result is not None: + self._send_json(200, self.tool_result) + else: + self._send_json(200, {"jsonrpc": "2.0", "id": body.get("id"), "result": {"content": []}}) + else: + self._send_json( + 400, + {"jsonrpc": "2.0", "id": body.get("id"), "error": {"code": -32601, "message": "Method not found"}}, + ) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +class McpServer: + """Context manager that runs a JSON-RPC MCP simulator on a random port.""" + + def __init__(self, tool_result: dict[str, Any] | None = None, session_id: str = "test-mcp-session"): + self._tool_result = tool_result + self._session_id = session_id + self._server: ThreadingHTTPServer | None = None + self._thread: threading.Thread | None = None + self.calls: list[dict[str, Any]] = [] + + @property + def url(self) -> str: + assert self._server is not None + host, port = self._server.server_address + return f"http://{host}:{port}" + + def start(self) -> McpServer: + handler = type( + "Handler", + (_McpHandler,), + { + "calls": self.calls, + "session_header": self._session_id, + "tool_result": self._tool_result, + }, + ) + self._server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + return self + + def stop(self) -> None: + if self._server is not None: + self._server.shutdown() + self._server.server_close() + if self._thread is not None: + self._thread.join(timeout=2) + + def __enter__(self) -> McpServer: + return self.start() + + def __exit__(self, *args: object) -> None: + self.stop() + + +# ---- helpers --------------------------------------------------------------- + + +def _tool_result(answer: str) -> dict[str, Any]: + """Build a JSON-RPC tools/call response with ReMe's answer format.""" + return { + "jsonrpc": "2.0", + "id": 2, + "result": {"content": [{"type": "text", "text": answer}]}, + } + + +def _tool_error(code: int, message: str) -> dict[str, Any]: + """Build a JSON-RPC error response.""" + return { + "jsonrpc": "2.0", + "id": 2, + "error": {"code": code, "message": message}, + } + + +# ---- tests ----------------------------------------------------------------- + + +def test_mcp_call_returns_result_on_success(): + """_mcp_call should complete the full handshake and return the tool result.""" + with McpServer(tool_result=_tool_result("Recorded facts about auth rewrite.")) as server: + result = auto_memory._mcp_call( + server.url, + "auto_memory_codex", + {"transcript_path": "/tmp/t.jsonl", "session_id": "sess-1"}, + ) + + assert result is not None + assert "error" not in result + content = result["result"]["content"] + assert content[0]["text"] == "Recorded facts about auth rewrite." + + +def test_mcp_call_handshake_sequence(): + """The MCP handshake must send initialize → initialized → tools/call in order.""" + with McpServer(tool_result=_tool_result("ok")) as server: + auto_memory._mcp_call( + server.url, + "auto_memory_codex", + {"transcript_path": "/tmp/t.jsonl", "session_id": "sess-1"}, + ) + + assert len(server.calls) == 3 + assert server.calls[0]["method"] == "initialize" + assert server.calls[1]["method"] == "notifications/initialized" + assert server.calls[2]["method"] == "tools/call" + assert server.calls[2]["params"]["name"] == "auto_memory_codex" + assert server.calls[2]["params"]["arguments"]["transcript_path"] == "/tmp/t.jsonl" + + +def test_mcp_call_raises_on_unreachable(): + """When the server is not running, _mcp_call raises URLError (caught by main()).""" + import urllib.error + + try: + auto_memory._mcp_call("http://127.0.0.1:1", "auto_memory_codex", {"session_id": "x"}) + except urllib.error.URLError: + pass # expected — connection refused + except OSError: + pass # also possible on some platforms + else: + pytest.fail("expected URLError or OSError for unreachable server") + + +def test_result_status_skipped_from_real_mcp_response(): + """_result_status must detect 'Skipped' in MCP content text.""" + mcp_result = _tool_result("Skipped: no messages") + assert auto_memory._result_status(mcp_result) == "skipped" + + +def test_result_status_error_from_real_mcp_response(): + mcp_result = _tool_error(-32000, "Internal error") + assert auto_memory._result_status(mcp_result) == "error" + + +def test_result_status_ok_from_real_mcp_response(): + mcp_result = _tool_result("Recorded 3 facts into daily/2026-07-20/auth.md") + assert auto_memory._result_status(mcp_result) == "ok" + + +def test_result_status_no_response_from_none(): + assert auto_memory._result_status(None) == "no-response" diff --git a/tests/unit/test_codex_plugin.py b/tests/unit/test_codex_plugin.py new file mode 100644 index 00000000..1b2d9960 --- /dev/null +++ b/tests/unit/test_codex_plugin.py @@ -0,0 +1,94 @@ +"""Tests for the Codex plugin packaging and structure.""" + +# pylint: disable=missing-function-docstring + +import json +from pathlib import Path + +PLUGIN_ROOT = Path(__file__).parents[2] / "plugins" / "codex" / "reme" + + +def _read_json(path: Path) -> dict: + return json.loads(path.read_text(encoding="utf-8")) + + +class TestPluginJson: + """Validate .codex-plugin/plugin.json structure.""" + + def test_plugin_json_exists(self): + assert (PLUGIN_ROOT / ".codex-plugin" / "plugin.json").is_file() + + def test_plugin_json_valid(self): + data = _read_json(PLUGIN_ROOT / ".codex-plugin" / "plugin.json") + assert data["name"] == "reme" + assert "version" in data + assert "description" in data + assert "skills" in data + assert "mcpServers" in data + assert "hooks" in data + + def test_plugin_references_existing_paths(self): + data = _read_json(PLUGIN_ROOT / ".codex-plugin" / "plugin.json") + skills_dir = data.get("skills", "") + mcp_file = data.get("mcpServers", "") + hooks_file = data.get("hooks", "") + assert (PLUGIN_ROOT / skills_dir).is_dir(), f"skills dir not found: {skills_dir}" + assert (PLUGIN_ROOT / mcp_file).is_file(), f"mcp config not found: {mcp_file}" + assert (PLUGIN_ROOT / hooks_file).is_file(), f"hooks config not found: {hooks_file}" + + def test_plugin_interface_fields(self): + data = _read_json(PLUGIN_ROOT / ".codex-plugin" / "plugin.json") + iface = data.get("interface", {}) + assert iface.get("displayName") == "ReMe Memory" + assert iface.get("category") == "Productivity" + + +class TestMcpJson: + """Validate .mcp.json structure and port.""" + + def test_mcp_json_valid(self): + data = _read_json(PLUGIN_ROOT / ".mcp.json") + servers = data.get("mcpServers", {}) + assert "reme" in servers + assert servers["reme"]["type"] == "http" + + def test_mcp_url_uses_default_port_2333(self): + data = _read_json(PLUGIN_ROOT / ".mcp.json") + url = data["mcpServers"]["reme"]["url"] + assert ":2333" in url, f"expected port 2333, got {url}" + + +class TestHooksJson: + """Validate hooks.json structure.""" + + def test_hooks_json_valid(self): + data = _read_json(PLUGIN_ROOT / "hooks" / "hooks.json") + hooks = data.get("hooks", {}) + assert "Stop" in hooks + + def test_hook_command_has_windows_path(self): + data = _read_json(PLUGIN_ROOT / "hooks" / "hooks.json") + stop_hooks = data["hooks"]["Stop"] + hook_block = stop_hooks[0]["hooks"][0] + assert "commandWindows" in hook_block, "hook must declare commandWindows" + assert hook_block.get("timeout") == 30 + + def test_hook_command_points_to_existing_script(self): + """The hook command path is relative to PLUGIN_ROOT — verify the script exists.""" + assert (PLUGIN_ROOT / "hooks" / "auto_memory.py").is_file() + + +class TestSkillMarkdown: + """Validate the reme-memory skill.""" + + def test_skill_md_exists_and_has_frontmatter(self): + skill_path = PLUGIN_ROOT / "skills" / "reme-memory" / "SKILL.md" + assert skill_path.is_file() + content = skill_path.read_text(encoding="utf-8") + assert content.startswith("---") + assert "name: reme-memory" in content + assert "description:" in content + + def test_skill_references_correct_port(self): + content = (PLUGIN_ROOT / "skills" / "reme-memory" / "SKILL.md").read_text(encoding="utf-8") + assert "2333" in content, "skill should reference default port 2333" diff --git a/tests/unit/test_hook_auto_memory.py b/tests/unit/test_hook_auto_memory.py new file mode 100644 index 00000000..68ed906d --- /dev/null +++ b/tests/unit/test_hook_auto_memory.py @@ -0,0 +1,154 @@ +"""Tests for the Codex Stop hook auto_memory.py logic. + +These test the hook's pure functions without requiring a running ReMe server. +""" + +# pylint: disable=missing-class-docstring,missing-function-docstring,protected-access + +import json +import sys +from pathlib import Path + +# Import the hook module from the Codex plugin (not a regular package, so the +# sys.path manipulation is intentional). +_HOOK_DIR = Path(__file__).parents[2] / "plugins" / "codex" / "reme" / "hooks" +sys.path.insert(0, str(_HOOK_DIR)) +# pylint: disable=wrong-import-position +import auto_memory # noqa: E402 + +# ---- _result_status ------------------------------------------------------- + + +class TestResultStatus: + """Tests for the _result_status function (checks MCP answer text).""" + + def test_no_response_on_none(self): + assert auto_memory._result_status(None) == "no-response" + + def test_error_on_jsonrpc_error(self): + result = {"error": {"code": -32600, "message": "Invalid Request"}} + assert auto_memory._result_status(result) == "error" + + def test_skipped_when_answer_starts_with_skipped(self): + """ReMe MCP returns only answer text, metadata is stripped.""" + result = { + "result": { + "content": [{"type": "text", "text": "Skipped: no messages"}], + }, + } + assert auto_memory._result_status(result) == "skipped" + + def test_error_on_mcp_is_error(self): + """ToolError → CallToolResult(isError=True) inside the result key.""" + result = {"result": {"content": [{"type": "text", "text": "Error: boom"}], "isError": True}} + assert auto_memory._result_status(result) == "error" + + def test_ok_when_answer_has_content(self): + result = { + "result": { + "content": [{"type": "text", "text": "Recorded facts about the auth rewrite."}], + }, + } + assert auto_memory._result_status(result) == "ok" + + def test_ok_with_empty_content(self): + """Empty content array should default to 'ok'.""" + result = {"result": {"content": []}} + assert auto_memory._result_status(result) == "ok" + + def test_ok_with_non_dict_result(self): + """Non-dict result content should not crash.""" + assert auto_memory._result_status({"result": "plain string"}) == "ok" + + +# ---- Payload parsing ------------------------------------------------------ + + +class TestPayloadParsing: + def test_session_id_and_transcript_path_extracted(self, monkeypatch): + """The hook reads session_id and transcript_path from the payload.""" + payload = { + "session_id": "sess-abc", + "transcript_path": "/tmp/codex-transcript.jsonl", + } + monkeypatch.setattr("sys.stdin", _fake_stdin(json.dumps(payload))) + + data = json.loads(sys.stdin.read() or "{}") + assert data["session_id"] == "sess-abc" + assert data["transcript_path"] == "/tmp/codex-transcript.jsonl" + + def test_empty_payload_returns_empty_dict(self, monkeypatch): + monkeypatch.setattr("sys.stdin", _fake_stdin("")) + data = json.loads(sys.stdin.read() or "{}") + assert data == {} + + def test_malformed_payload_returns_empty_dict(self, monkeypatch): + monkeypatch.setattr("sys.stdin", _fake_stdin("not valid json {{{")) + try: + data = json.loads(sys.stdin.read() or "{}") + except Exception: + data = {} + assert data == {} + + +# ---- MCP call tool selection ---------------------------------------------- + + +def test_hook_calls_auto_memory_codex(): + """The Codex hook uses auto_memory_codex, not auto_memory_cc.""" + # The tool name is hardcoded in main() — verify it directly. + with open(auto_memory.__file__, encoding="utf-8") as fh: + source = fh.read() + assert '"auto_memory_codex"' in source + assert "transcript_path" in source + + +# ---- Windows detach path -------------------------------------------------- + + +def test_windows_spawn_detached_writes_temp_file(tmp_path, monkeypatch): + """_spawn_detached writes the payload to a temp file and spawns a subprocess.""" + monkeypatch.setattr(auto_memory.sys, "executable", "python") + monkeypatch.setattr(auto_memory.sys, "argv", ["auto_memory.py"]) + + # Replace mkstemp at the tempfile module level (it's a local import in _spawn_detached) + import tempfile as _tempfile + + _real_mkstemp = _tempfile.mkstemp + monkeypatch.setattr( + _tempfile, + "mkstemp", + lambda prefix=None, suffix=None, dir=None: _real_mkstemp( + prefix=prefix, + suffix=suffix, + dir=str(tmp_path), + ), + ) + + calls = [] + + class FakePopen: + def __init__(self, args, **kwargs): + calls.append({"args": args, "kwargs": kwargs}) + + monkeypatch.setattr("subprocess.Popen", FakePopen) + + payload = {"session_id": "sess-1", "transcript_path": "/tmp/transcript.jsonl"} + auto_memory._spawn_detached(payload) + + assert len(calls) == 1 + args = calls[0]["args"] + assert args[0] == "python" + assert "--payload-file" in args + # Verify the temp file was written with the payload + written_file = args[args.index("--payload-file") + 1] + assert json.loads(Path(written_file).read_text(encoding="utf-8")) == payload + + +# ---- Helpers -------------------------------------------------------------- + + +def _fake_stdin(content: str): + import io + + return io.StringIO(content) diff --git a/tests/unit/test_mcp_error_e2e.py b/tests/unit/test_mcp_error_e2e.py new file mode 100644 index 00000000..c24084e9 --- /dev/null +++ b/tests/unit/test_mcp_error_e2e.py @@ -0,0 +1,202 @@ +"""E2E test: verify MCPService signals ReMe job failures as MCP errors. + +Tests the real ``MCPService.add_job`` → ``execute_tool`` → ``Tool.run()`` +path. When a BaseJob returns ``Response(success=False, ...)`` the +``execute_tool`` closure must raise ``ToolError`` so FastMCP produces an +MCP response with ``isError: true``. +""" + +# pylint: disable=missing-class-docstring,missing-function-docstring,protected-access + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest +from fastmcp.exceptions import ToolError + +from reme.components.job import BaseJob +from reme.components.service import MCPService +from reme.schema import Response + +# --------------------------------------------------------------------------- +# Real ReMe jobs that mirror the auto_memory_codex error pattern +# --------------------------------------------------------------------------- + + +class _SucceedJob(BaseJob): + """A job that returns success=True (the normal path).""" + + def _build_steps(self): + return [] + + async def __call__(self, **kwargs) -> Response: + return Response(success=True, answer="Recorded 3 facts into daily/2026-07-25/notes.md") + + +class _FailJob(BaseJob): + """A job that returns success=False — exactly what auto_memory_codex does + when it cannot resolve the transcript path.""" + + def _build_steps(self): + return [] + + async def __call__(self, **kwargs) -> Response: + return Response(success=False, answer="Error: could not resolve transcript path") + + +class _RealisticFailJob(BaseJob): + """A job that mimics auto_memory_codex more closely: sets response in the + same way the real step does.""" + + def _build_steps(self): + return [] + + async def __call__(self, **kwargs) -> Response: + # Same pattern as auto_memory_codex.py:64-66 + resp = Response() + resp.success = False + resp.answer = "Error: could not resolve transcript path" + return resp + + +# --------------------------------------------------------------------------- +# Minimal app object needed by MCPService.build_service +# --------------------------------------------------------------------------- + + +def _dummy_app(name: str = "test") -> SimpleNamespace: + """Minimal object needed by MCPService.build_service.""" + return SimpleNamespace( + config=SimpleNamespace(app_name=name), + context=SimpleNamespace(metadata={}), + ) + + +# --------------------------------------------------------------------------- +# tests +# --------------------------------------------------------------------------- + + +class TestMCPServiceToolError: + """Tests that exercise the real MCPService.add_job + execute_tool path.""" + + @pytest.mark.asyncio + async def test_success_job_returns_answer(self): + """A successful ReMe job → execute_tool returns the answer text.""" + service = MCPService(tool_error_on_failure=True) + service.build_service(_dummy_app("success-test")) + + job = _SucceedJob(name="test_succeed") + assert service.add_job(job) is True + + # Call the tool through FastMCP's tool registry — same as a real MCP call. + tool = await service.service.get_tool("test_succeed") + result = await tool.run({}) + assert result.is_error is False, f"Success job should have is_error=False, got {result}" + content_text = result.content[0].text + assert "Recorded 3 facts" in content_text, f"Wrong content: {content_text}" + + @pytest.mark.asyncio + async def test_fail_job_raises_tool_error(self): + """A failed ReMe job (success=False) → execute_tool raises ToolError. + + This is THE test for the P2 bug. Before the fix, MCPService returned + the error text as a normal result and _result_status logged it as 'ok'. + """ + service = MCPService(tool_error_on_failure=True) + service.build_service(_dummy_app("fail-test")) + + job = _FailJob(name="test_fail") + assert service.add_job(job) is True + + tool = await service.service.get_tool("test_fail") + + # run() should raise ToolError (which FastMCP converts to isError=True) + with pytest.raises(ToolError) as exc_info: + await tool.run({}) + assert "could not resolve transcript path" in str(exc_info.value), f"Wrong error message: {exc_info.value}" + + @pytest.mark.asyncio + async def test_realistic_fail_job_raises_tool_error(self): + """Even when the step sets response.success/answer manually (the + auto_memory_codex pattern), execute_tool must raise ToolError.""" + service = MCPService(tool_error_on_failure=True) + service.build_service(_dummy_app("realistic-test")) + + job = _RealisticFailJob(name="test_realistic_fail") + assert service.add_job(job) is True + + tool = await service.service.get_tool("test_realistic_fail") + + with pytest.raises(ToolError) as exc_info: + await tool.run({}) + assert "could not resolve transcript path" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_both_jobs_in_same_service(self): + """Success and failure coexist, no cross-talk.""" + service = MCPService(tool_error_on_failure=True) + service.build_service(_dummy_app("both-test")) + + assert service.add_job(_SucceedJob(name="test_ok")) is True + assert service.add_job(_FailJob(name="test_err")) is True + + ok_tool = await service.service.get_tool("test_ok") + err_tool = await service.service.get_tool("test_err") + + # Success + ok_result = await ok_tool.run({}) + assert ok_result.is_error is False + + # Failure + with pytest.raises(ToolError) as exc_info: + await err_tool.run({}) + assert "could not resolve transcript path" in str(exc_info.value) + + +# --------------------------------------------------------------------------- +# Hook classification — verifies _result_status handles the real MCP shape +# --------------------------------------------------------------------------- + + +def test_hook_result_status_on_real_response_shape(): + """The hook's _result_status correctly classifies real MCP response dicts.""" + # pylint: disable=import-outside-toplevel + _HOOK_DIR = Path(__file__).parents[2] / "plugins" / "codex" / "reme" / "hooks" + import sys + + sys.path.insert(0, str(_HOOK_DIR)) + import auto_memory # noqa: E402 # pylint: disable=wrong-import-position + + # Real shape from a failed job: ToolError → CallToolResult(isError=True) + failed = { + "jsonrpc": "2.0", + "id": 2, + "result": { + "content": [{"type": "text", "text": "Error: could not resolve transcript path"}], + "isError": True, + }, + } + assert auto_memory._result_status(failed) == "error" + + ok = { + "jsonrpc": "2.0", + "id": 2, + "result": { + "content": [{"type": "text", "text": "Recorded 3 facts."}], + "isError": False, + }, + } + assert auto_memory._result_status(ok) == "ok" + + skipped = { + "jsonrpc": "2.0", + "id": 2, + "result": { + "content": [{"type": "text", "text": "Skipped: no messages"}], + "isError": False, + }, + } + assert auto_memory._result_status(skipped) == "skipped"