fix(steps): guard compressor exceptions and fix ReadStep boolean override

1. search_v2: catch per-entry exceptions from run_job('compressor') inside
   compress() so asyncio.gather never propagates a compressor failure (e.g.
   temporary LLM outage). The failing entry keeps its original body while
   remaining entries are still compressed, preserving already-retrieved
   search results.

2. read: replace 'context_value or yaml_value' with an existence check so
   that a runtime-injected False can explicitly disable a YAML-true
   read_step_format_session flag.

Add focused unit tests for both paths.

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)

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(

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

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