diff --git a/reme/steps/file_io/read.py b/reme/steps/file_io/read.py index 4d431313..0df2342a 100644 --- a/reme/steps/file_io/read.py +++ b/reme/steps/file_io/read.py @@ -132,9 +132,12 @@ class ReadStep(BaseStep): start_line, end_line = self.context.get("start_line"), self.context.get("end_line") with_neighbors: bool = bool(self.kwargs.get("with_neighbors", False)) max_neighbors_per_direction: int = int(self.kwargs.get("max_neighbors_per_direction", 10)) - format_session: bool = bool( - self.context.get("read_step_format_session") or self.kwargs.get("read_step_format_session", False), - ) + # Injected value takes precedence over YAML kwargs; check existence + # (not truthiness) so an explicit False can disable a YAML-true flag. + _format_session = self.context.get("read_step_format_session") + if _format_session is None: + _format_session = self.kwargs.get("read_step_format_session", False) + format_session: bool = bool(_format_session) # Validate inputs and target before touching the filesystem twice. target = self._resolve_target(raw) diff --git a/reme/steps/index/search_v2.py b/reme/steps/index/search_v2.py index 02462a76..0ea4b2ed 100644 --- a/reme/steps/index/search_v2.py +++ b/reme/steps/index/search_v2.py @@ -264,6 +264,12 @@ class SearchV2Step(_ToolContextDedupMixin, BaseStep): logged, so search behaves as if compression were disabled. This avoids a hard ``Job compressor not found`` failure when a benchmark config forgets to define the compressor job/component. + + Per-entry exceptions raised by the compressor job (e.g. a temporary + LLM outage) are caught inside ``compress`` so they never propagate + through ``asyncio.gather``: the failing entry keeps its original body + while the remaining entries are still compressed, preserving already + retrieved search results. """ assert self.context is not None # Guard: when the compressor job is missing from the active config @@ -292,7 +298,14 @@ class SearchV2Step(_ToolContextDedupMixin, BaseStep): body = (entry.get("body", "") or "").strip() if not body: return - response = await self.run_job("compressor", text=body, queries=queries) + try: + response = await self.run_job("compressor", text=body, queries=queries) + except Exception as exc: # pylint: disable=broad-except + self.logger.warning( + f"[{self.name}] session body compression raised path={path!r} " + f"error={exc!r}; keeping original", + ) + return compressed = str(response.answer or "").strip() if not response.success or not compressed: self.logger.warning( diff --git a/tests/unit/test_crud_steps.py b/tests/unit/test_crud_steps.py index 38feea75..1132a686 100644 --- a/tests/unit/test_crud_steps.py +++ b/tests/unit/test_crud_steps.py @@ -778,6 +778,39 @@ def test_read_empty_path_rejected(): _run(run()) +def test_read_format_session_injected_false_disables_yaml_true(): + """Runtime-injected ``read_step_format_session=False`` takes precedence + over a YAML-true config — existence check, not truthiness.""" + + async def run(): + from agentscope.message import Msg + + m = Msg(name="user", role="user", content=[{"type": "text", "text": "hello world"}]) + jsonl_body = m.model_dump_json() + "\n" + + with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): + session_path = Path(tmp) / "session" / "dialog" / "s.jsonl" + session_path.parent.mkdir(parents=True, exist_ok=True) + session_path.write_text(jsonl_body, encoding="utf-8") + + store = await _make_store() + + # YAML config read_step_format_session=True, no injection → formatting active. + step = crud_read.ReadStep(file_store=store, read_step_format_session=True) + await step(path="session/dialog/s.jsonl") + assert "[user @" in str(step.context.response.answer) + + # Same YAML config, but runtime injects False → must disable formatting. + step = crud_read.ReadStep(file_store=store, read_step_format_session=True) + await step(path="session/dialog/s.jsonl", read_step_format_session=False) + assert "[user @" not in str(step.context.response.answer) + + await store.close() + print("✓ test_read_format_session_injected_false_disables_yaml_true passed") + + _run(run()) + + # -- write / edit -------------------------------------------------------- diff --git a/tests/unit/test_search_step.py b/tests/unit/test_search_step.py index c5c1388a..84189e5e 100644 --- a/tests/unit/test_search_step.py +++ b/tests/unit/test_search_step.py @@ -1228,3 +1228,34 @@ def test_search_v2_step_keeps_original_body_when_compression_fails(): assert "boom" not in resp.answer asyncio.run(run()) + + +def test_search_v2_step_keeps_original_body_when_compression_raises(): + """A compressor job that raises (e.g. temporary LLM outage) keeps the original + body for that entry while other entries are still compressed; the search itself + never fails.""" + + async def run(): + failing = _chunk("s1", "session/dialog/s1.jsonl", "dialog text one", "vector", 0.9, line=2) + ok = _chunk("s2", "session/dialog/s2.jsonl", "dialog text two", "vector", 0.8, line=1) + store = FakeSearchStore(vector_results=[failing, ok]) + step = SearchV2Step(file_store=store, expand_links=False) + + async def fake_run_job(name, /, **kwargs): + if "one" in kwargs.get("text", ""): + raise RuntimeError("temporary LLM outage") + return Response(success=True, answer="compressed!") + + step.run_job = fake_run_job + ctx = RuntimeContext(query="alpha", _search={"_compress": {"session": "true"}}) + + resp = await step(ctx) + + assert resp.success is True + # The failing entry keeps its original body. + assert "dialog text one" in resp.answer + # The healthy entry is still compressed. + assert "compressed!" in resp.answer + assert "dialog text two" not in resp.answer + + asyncio.run(run())