refactor(plugins): move BaseAgenticAnswerStep into lme/beam plugins

Move the shared benchmark answer base class from the core
reme.steps.benchmark package into each plugin's own src tree
(reme_lme.base_agentic_answer / reme_beam.base_agentic_answer) with
absolute imports, drop the core benchmark steps package, and update
plugin READMEs accordingly.

plugins/beam/README.md

@@ -31,8 +31,8 @@ The existing `auto_memory`, `agentic_answer`, `answer_judge`, `bench` and `judge
 names and model environment variables are unchanged. Explicit application/CLI overrides
 still take precedence. Installing this plugin does not start an evaluation.
 
-The shared answer base class lives in `reme.steps.benchmark.base_agentic_answer`.
-The old core-owned `reme.steps.benchmark.beam` Python import path is removed.
+The answer base class lives in `reme_beam.base_agentic_answer`.
+The old core-owned `reme.steps.benchmark` Python import path is removed.
 Custom Python callers should import memory, search and answer Steps from `reme_beam`, and install
 `beam-judge` before importing the judge Step from `judge_beam`. After uninstalling,
 Applications and CLI services must omit the plugin until it is installed again.

plugins/beam/README_ZH.md

@@ -28,8 +28,8 @@ editable 安装会注册 `beam` entry point，并让源码修改立即生效。r
 均保持关闭。原有 `auto_memory`、`agentic_answer`、`answer_judge`、`bench`、`judge` 名称及模型环境变量
 保持不变，显式应用参数和 CLI 覆盖仍优先。安装或启用插件不会自动开始评测。
 
-共享回答基类位于 `reme.steps.benchmark.base_agentic_answer`。
-原 `reme.steps.benchmark.beam` Python 导入路径已移除。自定义 Python 调用应从 `reme_beam`
+回答基类位于 `reme_beam.base_agentic_answer`。
+原 `reme.steps.benchmark` Python 导入路径已移除。自定义 Python 调用应从 `reme_beam`
 导入记忆、搜索和回答 Step；安装 `beam-judge` 后再从 `judge_beam` 导入评判 Step。
 卸载插件后，Application 和 CLI 服务必须移除插件选择，直到再次安装。
 卸载不会删除数据集、工作区或结果。修改插件后需重启已有服务。

plugins/beam/src/reme_beam/agentic_answer.py

@@ -1,6 +1,6 @@
 """BEAM agentic answer step – ReAct agent that answers questions using the search tool."""
 
-from reme.steps.benchmark import BaseAgenticAnswerStep
+from .base_agentic_answer import BaseAgenticAnswerStep
 
 
 class BeamAgenticAnswerStep(BaseAgenticAnswerStep):

plugins/beam/src/reme_beam/base_agentic_answer.py

@@ -1,15 +1,15 @@
-"""Shared base class for benchmark agentic-answer steps."""
+"""Base class for the BEAM agentic-answer step."""
 
 import os
 
-from ..base_step import BaseStep
-from ..index._dedup import _ToolContextDedupMixin
-from ...enumeration import ChunkEnum
-from ...utils.counter import global_counter_inc
+from reme.enumeration import ChunkEnum
+from reme.steps.base_step import BaseStep
+from reme.steps.index._dedup import _ToolContextDedupMixin
+from reme.utils.counter import global_counter_inc
 
 
 class BaseAgenticAnswerStep(BaseStep):
-    """ReAct-agent answer implementation shared by benchmark plugins.
+    """ReAct-agent answer implementation for the BEAM benchmark.
 
     Subclasses only need to set:
         TOOL_CONTEXT_PREFIX (str): prefix used to build the unique tool_context_id.
@@ -43,7 +43,7 @@ class BaseAgenticAnswerStep(BaseStep):
         ``INJECTED_JOB_KWARGS`` with per-request values derived from ``query``.
 
         When the runtime context carries a truthy ``compress_session`` flag,
-        session-transcript compression is enabled in the benchmark plugin's search Step by
+        session-transcript compression is enabled in this plugin's search Step by
         injecting a ``_search._compress.session`` marker plus the current
         ``query`` as the query-aware relevance filter. Default (falsy) leaves
         session chunks uncompressed.

plugins/lme/README.md

@@ -31,8 +31,8 @@ The existing `auto_memory`, `agentic_answer`, `answer_judge`, `bench` and `judge
 names and model environment variables are unchanged. Explicit application/CLI overrides
 still take precedence. Installing this plugin does not start an evaluation.
 
-The shared answer base class lives in `reme.steps.benchmark.base_agentic_answer`.
-The old core-owned `reme.steps.benchmark.lme` Python import path is removed.
+The answer base class lives in `reme_lme.base_agentic_answer`.
+The old core-owned `reme.steps.benchmark` Python import path is removed.
 Custom Python callers should import memory, search and answer Steps from `reme_lme`, and install
 `lme-judge` before importing the judge Step from `judge_lme`. After uninstalling,
 Applications and CLI services must omit the plugin until it is installed again.

plugins/lme/README_ZH.md

@@ -28,8 +28,8 @@ editable 安装会注册 `lme` entry point，并让源码修改立即生效。ru
 均保持关闭。原有 `auto_memory`、`agentic_answer`、`answer_judge`、`bench`、`judge` 名称及模型环境变量
 保持不变，显式应用参数和 CLI 覆盖仍优先。安装或启用插件不会自动开始评测。
 
-共享回答基类位于 `reme.steps.benchmark.base_agentic_answer`。
-原 `reme.steps.benchmark.lme` Python 导入路径已移除。自定义 Python 调用应从 `reme_lme`
+回答基类位于 `reme_lme.base_agentic_answer`。
+原 `reme.steps.benchmark` Python 导入路径已移除。自定义 Python 调用应从 `reme_lme`
 导入记忆、搜索和回答 Step；安装 `lme-judge` 后再从 `judge_lme` 导入评判 Step。
 卸载插件后，Application 和 CLI 服务必须移除插件选择，直到再次安装。
 卸载不会删除数据集、工作区或结果。修改插件后需重启已有服务。

plugins/lme/src/reme_lme/agentic_answer.py

@@ -1,6 +1,6 @@
 """LongMemEval agentic answer step – ReAct agent that answers questions using the search tool."""
 
-from reme.steps.benchmark import BaseAgenticAnswerStep
+from .base_agentic_answer import BaseAgenticAnswerStep
 
 
 class LmeAgenticAnswerStep(BaseAgenticAnswerStep):

plugins/lme/src/reme_lme/base_agentic_answer.py

@@ -0,0 +1,128 @@
+"""Base class for the LongMemEval agentic-answer step."""
+
+import os
+
+from reme.enumeration import ChunkEnum
+from reme.steps.base_step import BaseStep
+from reme.steps.index._dedup import _ToolContextDedupMixin
+from reme.utils.counter import global_counter_inc
+
+
+class BaseAgenticAnswerStep(BaseStep):
+    """ReAct-agent answer implementation for the LongMemEval benchmark.
+
+    Subclasses only need to set:
+        TOOL_CONTEXT_PREFIX (str): prefix used to build the unique tool_context_id.
+        JOB_TOOLS (list[str]): tools exposed to the ReAct agent; override to customize.
+        INJECTED_JOB_KWARGS (dict): server-owned kwargs injected into every job
+            tool call via ``injected_job_kwargs``; override the attribute or the
+            ``_injected_job_kwargs`` hook to customize.
+
+    Concrete subclasses are registered by the plugin manifest.
+
+    Inputs (from RuntimeContext):
+        query       (str, required): The question to answer.
+        query_time  (str, optional): ISO timestamp representing the query time,
+                    used to ground the agent's temporal context.
+
+    Output (written to context.response.answer):
+        The agent's final answer text.
+    """
+
+    # Reasoning-round budget. The AgentScope wrapper converts it to the
+    # backend's iteration-counting semantics; other backends ignore it.
+    MAX_ITERATION = 10
+    TOOL_CONTEXT_PREFIX: str = "content_agentic_answer"
+    JOB_TOOLS: list[str] = ["search", "add_draft", "read_all_draft", "read"]
+    INJECTED_JOB_KWARGS: dict = {"read_step_format_session": True}
+
+    def _injected_job_kwargs(self, query: str) -> dict:  # pylint: disable=unused-argument
+        """Server-owned kwargs injected into every job tool call.
+
+        Overridable hook: subclasses can extend the static
+        ``INJECTED_JOB_KWARGS`` with per-request values derived from ``query``.
+
+        When the runtime context carries a truthy ``compress_session`` flag,
+        session-transcript compression is enabled in this plugin's search Step by
+        injecting a ``_search._compress.session`` marker plus the current
+        ``query`` as the query-aware relevance filter. Default (falsy) leaves
+        session chunks uncompressed.
+        """
+        injected = dict(self.INJECTED_JOB_KWARGS)
+        if self.context is not None and self.context.get("compress_session"):
+            injected["_search"] = {
+                "_compress": {"session": "true"},
+                "queries": [query],
+                "type": "query-aware",
+            }
+        return injected
+
+    async def execute(self):
+        assert self.context is not None
+        query: str = self.context.get("query", "")
+        query_time: str | None = self.context.get("query_time")
+
+        if not query:
+            self.context.response.success = False
+            self.context.response.answer = "Skipped: empty query"
+            return self.context.response
+
+        # Build system prompt with optional temporal context
+        sys_prompt = self.get_prompt("system_prompt")
+        if query_time:
+            sys_prompt += "\n" + self.prompt_format("temporal_hint", query_time=query_time)
+
+        if self.app_context is not None:
+            tool_context_id = (
+                f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_"
+                f"{global_counter_inc(self.app_context.metadata, [self.TOOL_CONTEXT_PREFIX])}"
+            )
+        else:
+            tool_context_id = f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_local"
+        wrapper_kwargs = {
+            "system_prompt": sys_prompt,
+            "job_tools": list(self.JOB_TOOLS),
+            "react_config": {"max_iters": self.MAX_ITERATION},
+            "tool_context_id": tool_context_id,
+        }
+        if injected_job_kwargs := self._injected_job_kwargs(f"{query}(query time: {query_time})"):
+            wrapper_kwargs["injected_job_kwargs"] = injected_job_kwargs
+
+        if self.context.stream:
+            text = await self._stream_reply(query, **wrapper_kwargs)
+        else:
+            result = await self.agent_wrapper.reply(query, **wrapper_kwargs)
+            text = (result.get("result") or "").strip()
+
+        self.logger.debug(f"[{self.name}] response: {text!r}")
+
+        self.context.response.success = True
+        self.context.response.answer = text
+        self.context.response.metadata.update(
+            {
+                "query": query,
+                "query_time": query_time,
+                "sys_prompt": sys_prompt,
+                "response": text,
+            },
+        )
+
+        if self.app_context is not None:
+            self.app_context.metadata.get(_ToolContextDedupMixin.TOOL_CONTEXTS_KEY, {}).pop(tool_context_id, None)
+        return self.context.response
+
+    async def _stream_reply(self, query: str, **wrapper_kwargs) -> str:
+        """Stream unified chunks to the context stream queue."""
+        assert self.context is not None
+        text_parts: list[str] = []
+
+        async for chunk in self.agent_wrapper.reply_stream(query, **wrapper_kwargs):
+            await self.context.add_stream_string(chunk.chunk, chunk.chunk_type)
+
+            if chunk.chunk_type == ChunkEnum.CONTENT and isinstance(chunk.chunk, str):
+                text_parts.append(chunk.chunk)
+
+            if chunk.session_id:
+                self.context.response.metadata["session_id"] = chunk.session_id
+
+        return "".join(text_parts).strip()

reme/steps/__init__.py

@@ -1,11 +1,10 @@
 """steps"""
 
-from . import benchmark, common, evolve, file_io, index, transfer
+from . import common, evolve, file_io, index, transfer
 from .base_step import BaseStep
 
 __all__ = [
     "BaseStep",
-    "benchmark",
     "common",
     "evolve",
     "file_io",

reme/steps/benchmark/__init__.py

@@ -1,5 +0,0 @@
-"""Shared benchmark steps; concrete implementations live in plugins."""
-
-from .base_agentic_answer import BaseAgenticAnswerStep
-
-__all__ = ["BaseAgenticAnswerStep"]