GEPA eval-cache hits now carry output/trace (no more hollow reflective dataset)

core/cap_evolve/cache.py

@@ -1,12 +1,25 @@
 """Eval cache — skip a rollout when the same candidate was already scored on a task.
 
-Keyed by ``(hash of the candidate's editable files, task_id) -> {reward, feedback}``
-and persisted in the run dir, so re-evaluating an identical candidate (e.g. a parent
-re-sampled in GEPA, or a resumed run) costs nothing. The hash is over file CONTENTS,
-so two byte-identical candidates share cache entries even under different ids.
+Keyed by ``(hash of the candidate's editable files, task_id) ->
+{reward, feedback, rollout_file}`` and persisted in the run dir, so re-evaluating an
+identical candidate (e.g. a parent re-sampled in GEPA, or a resumed run) costs nothing.
+The hash is over file CONTENTS, so two byte-identical candidates share cache entries
+even under different ids.
+
+``rollout_file`` (#111) names the rollout json, under ``rollouts/<split>/``, that
+PRODUCED this cached score. GEPA's reflective dataset — its entire learning signal —
+needs the agent's ``output``/``trace``, not just the number; a score-only hit made
+``REFLECTION.md`` emit ``Agent output:`` (empty) for cached failing tasks and quietly
+degraded GEPA to a blind hill-climb on cached parents. Storing a POINTER rather than
+the payload keeps the cache small (~40 bytes/entry) and the reflection UNTRUNCATED:
+the rollout json is already persisted in the run dir, so a hit re-reads the real
+record instead of a redacted copy of it. A hit whose pointer no longer resolves is
+treated as a MISS by the reflection-bearing caller, so a score-only entry can never
+hollow out a reflective dataset again.
 
 Honesty notes:
-  * The cache stores only the SCORE (reward + feedback), never gold answers.
+  * The cache stores only the SCORE (reward + feedback) plus a pointer to the
+    already-persisted rollout json, never gold answers.
   * It is keyed on candidate-file content, so an edit (even whitespace) busts the
     key — a stale score can never be served for changed files.
   * It is an optimization, not a source of truth: ``events.jsonl`` still records
@@ -64,10 +77,13 @@ def hash_candidate_dir(candidate_dir: Path) -> str:
 class EvalCache:
     """A tiny JSON-file eval cache living in the run dir.
 
-    ``get(candidate_hash, task_id)`` -> ``{"reward", "feedback"}`` or ``None``;
-    ``put(candidate_hash, task_id, reward, feedback)`` persists. Persistence is a
-    single JSON object ``{ "<hash>::<task_id>": {...} }`` rewritten on each put — fine
-    for the run sizes here (a few thousand entries) and trivially portable.
+    ``get(candidate_hash, task_id)`` -> ``{"reward", "feedback"}`` plus an OPTIONAL
+    ``"rollout_file"`` (omitted when unknown — e.g. a pre-#111 entry; readers treat its
+    absence as a miss), or ``None``;
+    ``put(candidate_hash, task_id, reward, feedback, rollout_file=...)``
+    persists. Persistence is a single JSON object ``{ "<hash>::<task_id>": {...} }``
+    rewritten on each put — fine for the run sizes here (a few thousand entries) and
+    trivially portable.
     """
 
     def __init__(self, path: Path):
@@ -86,9 +102,15 @@ def _key(candidate_hash: str, task_id: str) -> str:
     def get(self, candidate_hash: str, task_id: str) -> dict | None:
         return self._data.get(self._key(candidate_hash, task_id))
 
-    def put(self, candidate_hash: str, task_id: str, reward: float, feedback: str = "") -> None:
-        self._data[self._key(candidate_hash, task_id)] = {
-            "reward": float(reward), "feedback": str(feedback or "")}
+    def put(self, candidate_hash: str, task_id: str, reward: float, feedback: str = "",
+            rollout_file: str = "") -> None:
+        """Persist a score. ``rollout_file`` is the ``rollouts/<split>/`` filename whose
+        json holds the rollout behind this score, so a later hit can rebuild the full
+        reflective signal (output + trace) instead of only the number (#111)."""
+        entry = {"reward": float(reward), "feedback": str(feedback or "")}
+        if rollout_file:
+            entry["rollout_file"] = str(rollout_file)
+        self._data[self._key(candidate_hash, task_id)] = entry
         self._flush()
 
     def _flush(self) -> None:

core/cap_evolve/gepa.py

@@ -44,6 +44,7 @@
 from __future__ import annotations
 
 import json
+import os
 import random
 import shutil
 import time
@@ -63,6 +64,7 @@
 )
 from .loop import SplitResult, aggregate_scores
 from .rundir import RunDir
+from .rundir import _atomic_write  # separate line: keeps the #199 merge clean
 from .types import Rollout, Score
 
 OptimizerFn = Callable[[Path, str], None]
@@ -130,6 +132,17 @@ def _eval_minibatch(
       * consults the **eval cache** keyed on ``(candidate-hash, task_id)`` to skip
         a rollout that was already scored for byte-identical candidate files.
 
+    A cache hit yields the SAME reflective signal as a fresh eval (#111). The cache
+    entry points at the rollout json that produced the score; on a hit we re-read that
+    json for the real ``output``/``trace`` and re-materialize it under THIS eval's tag,
+    so ``rollouts/train/*__<tag>__t0.json`` is complete either way and the optimizer's
+    ``trajectories/`` (pinned to the tag by ``_copy_step_trajectories``) holds this
+    minibatch's rollouts verbatim. The candidate hash is over file CONTENTS, so a hit's
+    rollout genuinely IS this minibatch's record, not a lookalike from elsewhere.
+    A hit whose pointer does not resolve (a pre-#111 cache, or pruned rollouts) is
+    treated as a MISS and re-run: paying one rollout beats handing the optimizer an
+    empty "Agent output:" and calling it reflection.
+
     Minibatch tasks are drawn from TRAIN only (test stays sealed; val is for the
     honest gate). One trial per task — the minibatch is a cheap signal, not the
     significance test.
@@ -147,12 +160,10 @@ def _eval_minibatch(
     with _live(adapter, candidate_dir) as ctx:
         for task in tasks:
             cached = cache.get(chash, task.id) if cache is not None else None
-            if cached is not None:
-                reward = float(cached.get("reward", 0.0))
-                fb = str(cached.get("feedback", ""))
-                scores.append(Score(task_id=task.id, reward=reward, feedback=fb,
-                                    n=1, stderr=0.0, trial_rewards=[reward],
-                                    raw={"cached": True}))
+            replay = (_replay_cached(cached, out_dir, task.id, tag, run_dir)
+                      if cached else None)
+            if replay is not None:
+                scores.append(replay)
                 continue
             rollout = adapter.run_target(task, ctx, seed=seed)
             if rollout is None:
@@ -169,13 +180,19 @@ def _eval_minibatch(
                      "output": _short(getattr(rollout, "output", None)),
                      "trace": _short(getattr(rollout, "trace", None))},
             ))
-            (out_dir / f"{task.id}__{tag}__t0.json").write_text(
-                json.dumps({"input": task.input, "rollout": rollout.to_dict(),
-                            "score": sc.to_dict()}, default=str),
-                encoding="utf-8",
-            )
+            rfile = f"{task.id}__{tag}__t0.json"
+            # ``_atomic_write`` (tmp + ``os.replace``) and NOT ``write_text``: a cache
+            # hit may have HARDLINKED a previous iteration's archived rollout to this
+            # name, and ``write_text`` truncates the shared inode — silently rewriting
+            # that earlier iteration's evidence. Replacing the directory entry breaks
+            # the link instead, so the archived record is immutable once written.
+            _atomic_write(out_dir / rfile,
+                          json.dumps({"input": task.input,
+                                      "rollout": rollout.to_dict(),
+                                      "score": sc.to_dict()}, default=str))
             if cache is not None:
-                cache.put(chash, task.id, sc.reward, sc.feedback or "")
+                cache.put(chash, task.id, sc.reward, sc.feedback or "",
+                          rollout_file=rfile)
 
     elapsed = time.time() - t0
     # Count ONLY rollouts actually fired (cache hits cost nothing) toward budget.
@@ -189,6 +206,74 @@ def _eval_minibatch(
     return result
 
 
+def _replay_cached(cached: dict, out_dir: Path, task_id: str, tag: str,
+                   run_dir: RunDir | None = None) -> Score | None:
+    """Rebuild a full ``Score`` (output + trace) from a cache hit, or ``None`` = miss.
+
+    The fix for #111. The cache entry carries ``rollout_file`` — the rollout json in
+    ``rollouts/<split>/`` that produced the cached score. We re-read it for the real
+    ``output``/``trace``, so the reflective dataset built from a cached parent minibatch
+    is identical to one built from a fresh eval, and we copy it under THIS eval's tag so
+    the tag-pinned ``trajectories/`` dir exists too.
+
+    Returning ``None`` (→ the caller re-runs the rollout) is deliberate for a pointerless
+    or unreadable entry: a score-only hit is exactly the hollow-reflection bug, so we pay
+    one rollout instead of serving empty "Agent output:" as if it were a trace.
+
+    ``eval_cache.json`` is plain JSON in the optimizer-writable run dir, so ``rollout_file``
+    is UNTRUSTED input (#142's threat model). It must be a BARE FILENAME under ``out_dir``:
+    ``Path(rfile).name != rfile`` rejects ``/etc/x``, ``../y`` and ``sub/z`` in one
+    comparison — an allowlist of the only legitimate shape, not a denylist of bad ones —
+    and the resolved-parent check confines it to ``out_dir`` even through a symlink.
+    Otherwise a tampered entry reads any json-shaped file on disk straight into the
+    optimizer's prompt (and hardlinks its inode into the run dir).
+    """
+    rfile = str(cached.get("rollout_file") or "")
+    if not rfile or Path(rfile).name != rfile:
+        return None
+    src = out_dir / rfile
+    try:
+        # Second layer: the resolved path must still live directly in out_dir.
+        if src.resolve().parent != out_dir.resolve():
+            return None
+        rec = json.loads(src.read_text(encoding="utf-8"))
+        reward = float(cached.get("reward", 0.0))
+    except (OSError, TypeError, ValueError, json.JSONDecodeError):
+        return None  # unreadable pointer OR malformed reward → honest MISS, never a crash
+    rollout = rec.get("rollout") or {}
+    # Re-materialize under this eval's tag so ``_copy_step_trajectories(tag=...)`` finds
+    # this minibatch's rollouts. Same bytes, new name — no re-run, no fabrication. A
+    # hardlink keeps a cache hit free on disk too (rollout jsons are write-once); the
+    # copy fallback covers filesystems/paths where linking isn't available.
+    #
+    # Overwrite unconditionally (no ``not dst.exists()`` short-circuit): on a resumed run
+    # the tag can already hold a DIFFERENT parent's rollout, and keeping it would let
+    # ``trajectories/`` serve candidate A while REFLECTION.md shows candidate B — under
+    # #199's unconditional "SAME minibatch VERBATIM" claim. ``os.link`` needs a free name,
+    # and the unlink is safe because fresh writes no longer truncate a shared inode.
+    dst = out_dir / f"{task_id}__{tag}__t0.json"
+    if dst != src:
+        try:
+            dst.unlink(missing_ok=True)
+            os.link(src, dst)
+        except OSError:
+            try:
+                _atomic_write(dst, json.dumps(rec, default=str))
+            except OSError as e:
+                # Loud, not silent: the Score is still complete, but trajectories/ now
+                # lacks this task while the prompt claims VERBATIM completeness.
+                if run_dir is not None:
+                    run_dir.log_event("rollout_rematerialize_failed", task_id=task_id,
+                                      tag=tag, file=dst.name, error=str(e)[:300])
+    return Score(
+        task_id=task_id, reward=reward, feedback=str(cached.get("feedback", "")),
+        n=1, stderr=0.0, trial_rewards=[reward],
+        raw={"cached": True, "errored": bool(rollout.get("error")),
+             "output": _short(rollout.get("output")),
+             "trace": _short(rollout.get("trace"))},
+    )
+
+
 def _short(x, n: int = 1500) -> str:
     if x is None:
         return ""

core/cap_evolve/harness.py

@@ -249,11 +249,14 @@ def _persist_trial(k: int, rollouts_for_k: dict) -> None:
             per_task_trials[tid].append(sc.reward)
             per_task_feedback[tid] = sc.feedback or per_task_feedback[tid]
             per_task_metrics[tid].append(sc.metrics)
-            (out_dir / f"{tid}__{tag}__t{k}.json").write_text(
-                json.dumps({"input": task.input, "rollout": rollout.to_dict(),
-                            "score": sc.to_dict()}, default=str),
-                encoding="utf-8",
-            )
+            # tmp + ``os.replace``, never ``write_text``: GEPA's cache replay hardlinks
+            # rollout jsons across tags, and truncating a shared inode would rewrite a
+            # PREVIOUS iteration's archived rollout in place. Replacing the directory
+            # entry breaks the link, keeping every archived record immutable.
+            _atomic_write(out_dir / f"{tid}__{tag}__t{k}.json",
+                          json.dumps({"input": task.input,
+                                      "rollout": rollout.to_dict(),
+                                      "score": sc.to_dict()}, default=str))
 
     # ``live()`` makes the candidate the one the target uses for this evaluation and
     # yields the ``ctx`` the runner consumes (default ctx == candidate_dir). Using a