Fix #2526: PrefEval drops a sample when search returns no text memories, so the judge gets

evaluation/scripts/PrefEval/pref_memos.py

@@ -4,6 +4,7 @@
 import os
 import sys
 import time
+import traceback
 
 import tiktoken
 
@@ -80,6 +81,10 @@ def search_memory_for_line(line_data, mem_client, top_k_value):
     """
 
     i, line = line_data
+    # Pre-declare so the except handler can reuse the already-parsed dict
+    # rather than re-parsing ``line`` (which would discard any intermediate
+    # state already set on ``original_data``).
+    original_data = None
     try:
         original_data = json.loads(line)
 
@@ -99,13 +104,27 @@ def search_memory_for_line(line_data, mem_client, top_k_value):
         start_time_search = time.monotonic()
         relevant_memories = mem_client.search(query=question, user_id=user_id, top_k=top_k_value)
         search_memories_duration = time.monotonic() - start_time_search
-        memories_str = (
-            "\n".join(
-                f"- {entry.get('memory', '')}"
-                for entry in relevant_memories["text_mem"][0]["memories"]
-            )
-            + f"\n{relevant_memories.get('pref_string', '')}"
+
+        # Defensively extract the first text_mem bucket's memories. An empty
+        # ``text_mem`` list (or a bucket without a ``memories`` key) is a
+        # valid "no retrieved text memories" state — not an error — and the
+        # row must still flow to the response/judge stages.
+        text_mem_entries: list = []
+        if isinstance(relevant_memories, dict):
+            text_mem_list = relevant_memories.get("text_mem") or []
+            if text_mem_list and isinstance(text_mem_list[0], dict):
+                text_mem_entries = text_mem_list[0].get("memories") or []
+        pref_string = (
+            relevant_memories.get("pref_string", "") if isinstance(relevant_memories, dict) else ""
         )
+        # Build the memories blob so an empty ``text_mem`` + empty
+        # ``pref_string`` yields ``""`` instead of a bare ``"\n"``. Downstream
+        # (and plain ``if not memories_str`` truthiness checks) then correctly
+        # treat the "no memories retrieved" case.
+        memory_lines = [f"- {entry.get('memory', '')}" for entry in text_mem_entries]
+        if pref_string:
+            memory_lines.append(pref_string)
+        memories_str = "\n".join(memory_lines).strip()
 
         memory_tokens_used = len(tokenizer.encode(memories_str))
 
@@ -121,9 +140,38 @@ def search_memory_for_line(line_data, mem_client, top_k_value):
         return original_data
 
     except Exception as e:
-        user_id_from_data = json.loads(line).get("user_id", "N/A")
-        print(f"Error searching memory for line {i + 1} (user_id: {user_id_from_data}): {e}")
-        return None
+        # Preserve the sample instead of returning ``None`` and letting the
+        # writer gate silently drop it. The row carries an ``error`` field so
+        # the response stage passes it through, and ``retrieved_memories_text``
+        # is set to an empty string so the row's downstream contract matches
+        # the "no memories retrieved" case the response stage already handles.
+        #
+        # Reuse ``original_data`` whenever the top-level ``json.loads`` already
+        # succeeded (the common case: a ``mem_client.search`` / tokenizer /
+        # metrics-update failure). Only fall back to re-parsing when the parse
+        # itself was the raising statement.
+        if original_data is not None:
+            fallback_data = original_data
+        else:
+            try:
+                fallback_data = json.loads(line)
+            except Exception:
+                fallback_data = None
+        user_id_from_data = fallback_data.get("user_id", "N/A") if fallback_data else "N/A"
+        # Log the full traceback so the originating frame (e.g. ``mem_client``,
+        # ``tokenizer.encode``, ``metrics_dict.update``) is diagnosable even
+        # after the exception has been swallowed into the row's ``error`` field.
+        print(
+            f"Error searching memory for line {i + 1} "
+            f"(user_id: {user_id_from_data}): {e}\n{traceback.format_exc()}"
+        )
+        if fallback_data is None:
+            return None
+        fallback_data.setdefault("metrics", {})["retrieved_memories_text"] = ""
+        new_error = f"search_error: {e}"
+        prev_error = fallback_data.get("error")
+        fallback_data["error"] = f"{prev_error}; {new_error}" if prev_error else new_error
+        return fallback_data
 
 
 def generate_response_for_line(line_data, openai_client, lib):
@@ -133,19 +181,27 @@ def generate_response_for_line(line_data, openai_client, lib):
     from utils.prompts import PREFEVAL_ANSWER_PROMPT
 
     i, line = line_data
+    # Pre-declare so the except handler can reuse the already-parsed dict
+    # rather than re-parsing ``line`` and losing any intermediate state
+    # (e.g. ``metrics`` fields already populated earlier).
+    original_data = None
     try:
         original_data = json.loads(line)
 
         question = original_data.get("question")
         metrics_dict = original_data.get("metrics", {})
         memories_str = metrics_dict.get("retrieved_memories_text")
 
-        # If an error occurred in 'add' or 'search' mode, just pass the line through
+        # If an error occurred in 'add' or 'search' mode, pass the line
+        # through. pref_eval.py reads ``data["response"]`` on every row, so
+        # we guarantee the key exists even for already-flagged failures.
         if original_data.get("error"):
+            original_data.setdefault("response", "")
             return original_data
 
         if not question:
             original_data["error"] = "Question not found in this line."
+            original_data.setdefault("response", "")
             return original_data
 
         # Check for None, as an empty string (no memories found) is a valid result
@@ -154,6 +210,7 @@ def generate_response_for_line(line_data, openai_client, lib):
                 "Error: retrieved_memories_text not found in metrics. "
                 "Please run 'search' mode first."
             )
+            original_data.setdefault("response", "")
             return original_data
 
         system_prompt = PREFEVAL_ANSWER_PROMPT.format(context=memories_str)
@@ -169,9 +226,38 @@ def generate_response_for_line(line_data, openai_client, lib):
         return original_data
 
     except Exception as e:
-        user_id_from_data = json.loads(line).get("user_id", "N/A")
-        print(f"Error generating response for line {i + 1} (user_id: {user_id_from_data}): {e}")
-        return None
+        # Preserve the sample instead of returning ``None`` and letting the
+        # writer gate silently drop it. The row keeps a ``response`` placeholder
+        # so ``pref_eval.py``'s judge still sees it (and classifies the empty
+        # response as an Unhelpful Response, the correct accounting for an
+        # upstream API failure).
+        #
+        # Reuse ``original_data`` whenever the top-level ``json.loads`` already
+        # succeeded; re-parse ``line`` only if the parse itself was the raising
+        # statement. Reusing the parsed dict preserves intermediate state (for
+        # instance, ``metrics`` that was already populated by the search stage
+        # and read from ``line``).
+        if original_data is not None:
+            fallback_data = original_data
+        else:
+            try:
+                fallback_data = json.loads(line)
+            except Exception:
+                fallback_data = None
+        user_id_from_data = fallback_data.get("user_id", "N/A") if fallback_data else "N/A"
+        print(
+            f"Error generating response for line {i + 1} "
+            f"(user_id: {user_id_from_data}): {e}\n{traceback.format_exc()}"
+        )
+        if fallback_data is None:
+            return None
+        new_error = f"response_error: {e}"
+        prev_error = fallback_data.get("error")
+        # Preserve any pre-existing ``error`` (e.g. flagged by the search
+        # stage) instead of silently discarding it.
+        fallback_data["error"] = f"{prev_error}; {new_error}" if prev_error else new_error
+        fallback_data.setdefault("response", "")
+        return fallback_data
 
 
 def main():

tests/evaluation/test_pref_memos.py

@@ -0,0 +1,194 @@
+"""Tests for evaluation/scripts/PrefEval/pref_memos.py.
+
+These tests pin the row-preservation contract described in
+openspec/changes/2026-10-21-2526-prefeval-drops-a-sample-when-search-returns-no-text-memories/spec.md:
+
+- An empty ``text_mem`` list is a valid "no retrieved text memories" state,
+  not an error, and the row must still be emitted downstream.
+- A search exception annotates the row with an ``error`` field (and empty
+  ``retrieved_memories_text``) instead of dropping it.
+- A response-generation exception annotates the row with an ``error`` field
+  and sets ``response = ""`` so ``pref_eval.py`` can run the judge.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+
+from unittest.mock import MagicMock
+
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+EVAL_SCRIPTS_DIR = os.path.join(REPO_ROOT, "evaluation", "scripts")
+EVAL_PREFEVAL_DIR = os.path.join(EVAL_SCRIPTS_DIR, "PrefEval")
+
+# The script relies on sibling-import semantics (``from irrelevant_conv import ...``
+# and ``from utils.prompts import ...``). Mirror the same sys.path plumbing the
+# script does at runtime so the module is importable in the test process.
+for _path in (EVAL_PREFEVAL_DIR, EVAL_SCRIPTS_DIR, REPO_ROOT):
+    if _path not in sys.path:
+        sys.path.insert(0, _path)
+
+import pref_memos  # noqa: E402  (sys.path mutation above is intentional)
+
+
+def _line_data(payload: dict) -> tuple[int, str]:
+    return (0, json.dumps(payload))
+
+
+def test_search_memory_for_line_returns_sample_when_text_mem_empty():
+    """Empty ``text_mem`` must not crash and must not drop the row."""
+    mem_client = MagicMock()
+    mem_client.search.return_value = {"text_mem": [], "pref_string": "pref-A"}
+    payload = {
+        "user_id": "u1",
+        "question": "What should I study?",
+        "metrics": {},
+        "preference": "likes math",
+    }
+
+    result = pref_memos.search_memory_for_line(_line_data(payload), mem_client, top_k_value=10)
+
+    assert result is not None, "sample must not be dropped when text_mem is empty"
+    assert "error" not in result, "empty text_mem is a valid state, not an error"
+    # Empty text_mem list + pref_string -> just the pref_string (no bare leading
+    # newline). Earlier the join produced "\npref-A"; after whitespace
+    # normalization the leading newline is stripped so downstream truthiness
+    # checks keep working.
+    assert result["metrics"]["retrieved_memories_text"] == "pref-A"
+    # Timing / token metrics should still be present.
+    assert "search_memories_duration_seconds" in result["metrics"]
+    assert result["metrics"]["memory_tokens_used"] >= 0
+    # Top-level fields are preserved.
+    assert result["user_id"] == "u1"
+    assert result["question"] == "What should I study?"
+    assert result["preference"] == "likes math"
+
+
+def test_search_memory_for_line_empty_text_mem_and_empty_pref_string():
+    """Both empty must yield ``""`` so ``if not memories_str`` detects the case."""
+    mem_client = MagicMock()
+    mem_client.search.return_value = {"text_mem": [], "pref_string": ""}
+    payload = {"user_id": "u1", "question": "Q", "metrics": {}}
+
+    result = pref_memos.search_memory_for_line(_line_data(payload), mem_client, top_k_value=10)
+
+    assert result is not None
+    assert "error" not in result
+    assert result["metrics"]["retrieved_memories_text"] == "", (
+        "no-text-mem + no-pref-string must produce empty string, not bare newline"
+    )
+
+
+def test_search_memory_for_line_returns_sample_with_error_on_exception():
+    """Any exception from mem_client.search must keep the row (with error)."""
+    mem_client = MagicMock()
+    mem_client.search.side_effect = RuntimeError("boom")
+    payload = {
+        "user_id": "u1",
+        "question": "Q",
+        "metrics": {},
+    }
+
+    result = pref_memos.search_memory_for_line(_line_data(payload), mem_client, top_k_value=10)
+
+    assert result is not None, "search exception must not cause row to be dropped"
+    assert "error" in result
+    assert "boom" in result["error"]
+    # Response stage looks for retrieved_memories_text to decide whether to proceed.
+    assert result["metrics"]["retrieved_memories_text"] == ""
+
+
+def test_search_memory_for_line_preserves_existing_error_on_exception():
+    """A row that already carries an ``error`` must not have it overwritten.
+
+    The exception-path error is appended, so multi-stage failure history is
+    preserved for the operator reading the JSONL after the fact.
+    """
+    mem_client = MagicMock()
+    mem_client.search.side_effect = RuntimeError("boom")
+    payload = {
+        "user_id": "u1",
+        "question": "Q",
+        "error": "add_error: earlier failure",
+        "metrics": {},
+    }
+
+    result = pref_memos.search_memory_for_line(_line_data(payload), mem_client, top_k_value=10)
+
+    assert result is not None
+    assert "add_error: earlier failure" in result["error"], (
+        "previous error must survive — appending preserves failure history"
+    )
+    assert "search_error: boom" in result["error"]
+
+
+def test_generate_response_for_line_returns_sample_with_error_on_api_failure():
+    """OpenAI exception must keep the row with response="" and error annotated."""
+    openai_client = MagicMock()
+    openai_client.chat.completions.create.side_effect = RuntimeError("HTTP 503")
+    payload = {
+        "user_id": "u1",
+        "question": "Q",
+        "preference": "p",
+        "metrics": {"retrieved_memories_text": ""},
+    }
+
+    result = pref_memos.generate_response_for_line(_line_data(payload), openai_client, "memos-api")
+
+    assert result is not None, "API error must not cause row to be dropped"
+    assert "error" in result
+    assert "503" in result["error"]
+    # pref_eval.py reads data["response"] - the key must exist.
+    assert "response" in result, "pref_eval.py requires 'response' key on every row"
+    assert result["response"] == ""
+
+
+def test_generate_response_for_line_passes_through_search_error_with_empty_response():
+    """A row that already carries an ``error`` from search must still get a ``response`` key.
+
+    Without this, ``pref_eval.py`` fails ``data["response"]`` lookup on
+    pass-through rows and the sample is silently dropped at the judge stage.
+    """
+    openai_client = MagicMock()
+    payload = {
+        "user_id": "u1",
+        "question": "Q",
+        "preference": "p",
+        "error": "search_error: something bad",
+        "metrics": {},
+    }
+
+    result = pref_memos.generate_response_for_line(_line_data(payload), openai_client, "memos-api")
+
+    assert result is not None
+    assert result["error"] == "search_error: something bad"
+    assert result.get("response", None) == ""
+    # Must not have tried to call OpenAI for an already-flagged row.
+    openai_client.chat.completions.create.assert_not_called()
+
+
+def test_search_memory_for_line_happy_path_unchanged():
+    """Populated text_mem produces the same output format as before."""
+    mem_client = MagicMock()
+    mem_client.search.return_value = {
+        "text_mem": [
+            {
+                "memories": [
+                    {"memory": "likes math"},
+                    {"memory": "dislikes seafood"},
+                ]
+            }
+        ],
+        "pref_string": "pref-string-X",
+    }
+    payload = {"user_id": "u1", "question": "Q", "metrics": {}}
+
+    result = pref_memos.search_memory_for_line(_line_data(payload), mem_client, top_k_value=10)
+
+    assert result is not None
+    assert result["metrics"]["retrieved_memories_text"] == (
+        "- likes math\n- dislikes seafood\npref-string-X"
+    )