From 5972eeddb960cc3cea2dc46b8f21afb217f18cc3 Mon Sep 17 00:00:00 2001 From: Osher Elhadad Date: Thu, 30 Jul 2026 02:22:49 +0300 Subject: [PATCH 1/2] GEPA eval-cache hits now carry output/trace (no more hollow reflective dataset) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The eval cache stored only {reward, feedback}, so `_eval_minibatch` rebuilt a cache-hit `Score` with `raw={"cached": True}` and no output/trace. GEPA re-samples parents constantly, so the parent minibatch that FEEDS reflection frequently hit the cache — and `_write_reflection` then emitted `- Agent output:` (empty) for those failing tasks. GEPA's headline advantage (rich reflection over failures) was being degraded by GEPA's own cache: a blind hill-climb on cached parents. Fix (issue #111 Option A, pointer flavour): each cache entry gains `rollout_file`, the name of the rollout json under `rollouts//` that produced the score. On a hit, `_replay_cached` re-reads that json for the real output/trace and hardlinks it under the current eval tag, so `rollouts//*____t0.json` is complete either way and the tag-pinned `trajectories/` dir exists for a cached minibatch too. A pointer flavour, not a payload flavour, because the traces are ALREADY persisted in the run dir — duplicating them into eval_cache.json would grow the cache with the trace size while the reflection stayed truncated. A pointer costs ~42 bytes/entry, keeps reflection untruncated, and the hardlink makes a hit free on disk. 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 serving an empty "Agent output:" as if it were a trace. Replayed output/trace go through the same `_short` truncation a fresh eval uses, so prompt size stays bounded. --- core/cap_evolve/cache.py | 44 +++++-- core/cap_evolve/gepa.py | 72 ++++++++++-- core/tests/test_gepa_cache_traces.py | 166 +++++++++++++++++++++++++++ 3 files changed, 262 insertions(+), 20 deletions(-) create mode 100644 core/tests/test_gepa_cache_traces.py diff --git a/core/cap_evolve/cache.py b/core/cap_evolve/cache.py index 62596afb..374320e8 100644 --- a/core/cap_evolve/cache.py +++ b/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//``, 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,11 @@ 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 ``{ "::": {...} }`` rewritten on each put — fine - for the run sizes here (a few thousand entries) and trivially portable. + ``get(candidate_hash, task_id)`` -> ``{"reward", "feedback", "rollout_file"}`` or + ``None``; ``put(candidate_hash, task_id, reward, feedback, rollout_file=...)`` + persists. Persistence is a single JSON object ``{ "::": {...} }`` + rewritten on each put — fine for the run sizes here (a few thousand entries) and + trivially portable. """ def __init__(self, path: Path): @@ -86,9 +100,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//`` 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: diff --git a/core/cap_evolve/gepa.py b/core/cap_evolve/gepa.py index 15445940..02620621 100644 --- a/core/cap_evolve/gepa.py +++ b/core/cap_evolve/gepa.py @@ -44,6 +44,7 @@ from __future__ import annotations import json +import os import random import shutil import time @@ -130,6 +131,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/*____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 +159,9 @@ 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) 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 +178,15 @@ 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( + rfile = f"{task.id}__{tag}__t0.json" + (out_dir / rfile).write_text( json.dumps({"input": task.input, "rollout": rollout.to_dict(), "score": sc.to_dict()}, default=str), encoding="utf-8", ) 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 +200,51 @@ def _eval_minibatch( return result +def _replay_cached(cached: dict, out_dir: Path, task_id: str, tag: str) -> 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//`` 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. + """ + rfile = str(cached.get("rollout_file") or "") + if not rfile: + return None + src = out_dir / rfile + try: + rec = json.loads(src.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, ValueError): + return None + rollout = rec.get("rollout") or {} + reward = float(cached.get("reward", 0.0)) + # 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. + dst = out_dir / f"{task_id}__{tag}__t0.json" + if dst != src and not dst.exists(): + try: + os.link(src, dst) + except OSError: + try: + dst.write_text(json.dumps(rec, default=str), encoding="utf-8") + except OSError: + pass # ponytail: trajectories/ then falls back to REFLECTION.md excerpts + 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 "" diff --git a/core/tests/test_gepa_cache_traces.py b/core/tests/test_gepa_cache_traces.py new file mode 100644 index 00000000..c0b704b7 --- /dev/null +++ b/core/tests/test_gepa_cache_traces.py @@ -0,0 +1,166 @@ +"""A GEPA eval-cache HIT must yield the same reflective signal as a fresh eval (#111). + +The cache used to store only ``{reward, feedback}``, so ``_eval_minibatch`` rebuilt a +``Score`` with ``raw={"cached": True}`` and no ``output``/``trace``. ``_write_reflection`` +then emitted ``- Agent output:`` (empty) for cached failing tasks — GEPA's whole learning +signal, blank, on exactly the parents it re-samples most. + +These tests pin the fix at both levels: the cache entry carries a pointer to the rollout +json that produced the score, and a hit re-reads it (and re-materializes it under the new +eval tag so the tag-pinned ``trajectories/`` dir still exists). +""" + +import json +import sys +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO / "core")) + + +def _adapter(): + """A deterministic adapter whose rollouts carry a distinctive output AND trace.""" + from cap_evolve import CapabilityAdapter, Rollout, Score, Task + + class _A(CapabilityAdapter): + def tasks(self, split): # noqa: ARG002 + return [Task(id=f"t{i}", input=f"in-{i}", target="ok") for i in range(4)] + + def run_target(self, task, ctx, *, seed=0): # noqa: ARG002 + return Rollout(task_id=task.id, + output=f"WRONG-ANSWER-for-{task.id}", + trace=f"STEP1 read {task.input}; STEP2 guessed") + + def score(self, task, rollout): # noqa: ARG002 + return Score(task_id=task.id, reward=0.0, + feedback=f"expected ok, got {rollout.output}", + trial_rewards=[0.0]) + + def materialize(self, candidate_dir, edits=None): # noqa: ARG002 + return None + + return _A() + + +@pytest.fixture +def setup(tmp_path): + from cap_evolve import Budget, RunDir + from cap_evolve.cache import EvalCache + cand = tmp_path / "cand" + cand.mkdir() + (cand / "cap.md").write_text("seed capability\n", encoding="utf-8") + run_dir = RunDir.create(tmp_path / ".capevolve", ts="c1", budget=Budget(max_iterations=4)) + return _adapter(), run_dir, cand, EvalCache(run_dir.root / "eval_cache.json") + + +def test_cache_hit_reflective_dataset_has_output_and_trace(setup): + """Fail-before / pass-after: a fully-cached minibatch must still produce a + REFLECTION.md with the agent's real output AND trajectory.""" + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + ids = ["t0", "t1", "t2"] + + first = gepa._eval_minibatch(adapter, cand, ids, run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + # Second eval of the same candidate + tasks: every task is a cache hit (0 rollouts). + before = run_dir.spent.metric_calls + second = gepa._eval_minibatch(adapter, cand, ids, run_dir=run_dir, cache=cache, + tag="mb_p_0001", seed=0) + assert run_dir.spent.metric_calls == before, "expected a full cache hit" + assert second.reward == first.reward + + for pt in second.per_task: + raw = pt.get("raw") or {} + assert raw.get("cached") is True + assert f"WRONG-ANSWER-for-{pt['task_id']}" in str(raw.get("output")) + assert "STEP1 read" in str(raw.get("trace")) + + wd = run_dir.root / "work" / "x" + wd.mkdir(parents=True) + gepa._write_reflection(wd, second) + refl = (wd / "REFLECTION.md").read_text(encoding="utf-8") + assert "- Agent output: \n" not in refl, "hollow reflective dataset" + assert "WRONG-ANSWER-for-t0" in refl + assert "- Trajectory: STEP1 read in-0" in refl + + +def test_cache_hit_rematerializes_rollouts_under_new_tag(setup): + """The cached hit must leave ``rollouts/train/*____t0.json`` for the NEW tag, so + ``harness._copy_step_trajectories(tag=...)`` still finds this minibatch verbatim.""" + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + gepa._eval_minibatch(adapter, cand, ["t0", "t1"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + gepa._eval_minibatch(adapter, cand, ["t0", "t1"], run_dir=run_dir, cache=cache, + tag="mb_p_0001", seed=0) + got = sorted(p.name for p in (run_dir.rollouts / "train").glob("*__mb_p_0001__t0.json")) + assert got == ["t0__mb_p_0001__t0.json", "t1__mb_p_0001__t0.json"] + rec = json.loads((run_dir.rollouts / "train" / "t0__mb_p_0001__t0.json") + .read_text(encoding="utf-8")) + assert rec["rollout"]["output"] == "WRONG-ANSWER-for-t0" + assert "STEP1" in rec["rollout"]["trace"] + + +def test_cache_entry_stores_rollout_pointer_not_payload(setup): + """Cache format: reward + feedback + a POINTER. The trace is NOT copied into the + cache, so the cache stays tiny however large the traces get.""" + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + raw = json.loads((run_dir.root / "eval_cache.json").read_text(encoding="utf-8")) + (entry,) = raw.values() + assert set(entry) == {"reward", "feedback", "rollout_file"} + assert entry["rollout_file"] == "t0__mb_p_0000__t0.json" + assert "STEP1" not in json.dumps(entry), "trace must not be duplicated into the cache" + + +def test_pointerless_or_missing_rollout_is_treated_as_a_miss(setup): + """A pre-#111 (score-only) entry, or one whose rollout json was pruned, must be + RE-RUN rather than served as an empty reflective row.""" + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + + # (a) legacy score-only entry + from cap_evolve.cache import hash_candidate_dir + chash = hash_candidate_dir(cand) + cache._data[f"{chash}::t0"] = {"reward": 0.0, "feedback": "legacy"} + cache._flush() + before = run_dir.spent.metric_calls + res = gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + assert run_dir.spent.metric_calls == before + 1, "legacy entry must miss" + assert "WRONG-ANSWER-for-t0" in str((res.per_task[0]["raw"] or {}).get("output")) + + # (b) pointer present but the rollout json is gone + (run_dir.rollouts / "train" / "t0__mb_p_0000__t0.json").unlink() + before = run_dir.spent.metric_calls + res = gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0002", seed=0) + assert run_dir.spent.metric_calls == before + 1, "dangling pointer must miss" + assert "WRONG-ANSWER-for-t0" in str((res.per_task[0]["raw"] or {}).get("output")) + + +def test_trace_is_bounded_in_the_reflective_signal(setup): + """Cache size + prompt size stay bounded: the replayed output/trace go through the + same ``_short`` truncation a fresh eval uses.""" + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + huge = "X" * 50_000 + orig = adapter.run_target + + def _big(task, ctx, *, seed=0): + r = orig(task, ctx, seed=seed) + r.trace = huge + return r + adapter.run_target = _big + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + res = gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0001", seed=0) + trace = str((res.per_task[0]["raw"] or {}).get("trace")) + assert len(trace) < 2000 and trace.endswith("…[truncated]") + # and the cache itself never grew by the trace + assert (run_dir.root / "eval_cache.json").stat().st_size < 1000 From 55b6041f2b0ecfd6b54dd8d6061d6ebb96eb1600 Mon Sep 17 00:00:00 2001 From: Osher Elhadad Date: Thu, 30 Jul 2026 03:05:30 +0300 Subject: [PATCH 2/2] Harden the GEPA cache-replay mechanism: confine the rollout pointer, stop archived rollouts being rewritten MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review fixes for #210. The reflective-dataset content was correct; the re-materialization *mechanism* had two exploitable defects. BLOCKING 1 — `rollout_file` was unvalidated. `eval_cache.json` is plain JSON in the optimizer-writable run dir, so every field it carries is untrusted input — exactly the threat #142 models (whose guard hashes `.capevolve/project` and excludes the run dir, so it does not cover this file). A crafted `../../../../secrets.json` or absolute pointer read any json-shaped file on disk straight into REFLECTION.md — i.e. into the optimizer LLM's prompt — and hardlinked the target's inode into the run dir. `_replay_cached` now requires a bare filename (`Path(rfile).name != rfile` rejects absolute, `../` and `sub/` in one comparison — an allowlist of the only legitimate shape, not a denylist of bad ones) and confines the resolved path to the rollouts dir as a second layer. A malformed `reward` now degrades to an honest MISS instead of raising `ValueError` mid-loop. BLOCKING 2 — a cache hit hardlinks a rollout under the new tag, and the fresh-eval writer used `write_text`, which truncates the *shared inode*. On a torn iteration + `--resume` (tags repeat, since `_save()` only runs at step end) a fresh write silently rewrote a PREVIOUS iteration's archived rollout with another candidate's output — no event, no warning, falsifying the evidence trail a sealed number rests on. Both writers into `rollouts/` now use `rundir._atomic_write` (tmp + `os.replace`), which replaces the directory entry instead of truncating, so an archived record is immutable once written. Hardlink sharing is retained, so a cache hit is still free on disk. Also: drop the stale `not dst.exists()` short-circuit, which let `trajectories/` serve a PRIOR parent's rollout while REFLECTION.md showed the current one, under #199's unconditional "SAME minibatch VERBATIM" claim; log `rollout_rematerialize_failed` instead of a silent `pass` when both link and copy fail; and correct the EvalCache docstring, which promised a 3-key entry shape while `rollout_file` is optional. 9 regression tests: path traversal (3 shapes), malformed reward, hardlink write-through, tag reuse, EXDEV copy fallback, loud re-materialization failure, and the `_copy_step_trajectories` no-write-through invariant. --- core/cap_evolve/cache.py | 6 +- core/cap_evolve/gepa.py | 59 ++++++-- core/cap_evolve/harness.py | 13 +- core/tests/test_gepa_cache_traces.py | 198 +++++++++++++++++++++++++++ 4 files changed, 254 insertions(+), 22 deletions(-) diff --git a/core/cap_evolve/cache.py b/core/cap_evolve/cache.py index 374320e8..98c2310f 100644 --- a/core/cap_evolve/cache.py +++ b/core/cap_evolve/cache.py @@ -77,8 +77,10 @@ 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", "rollout_file"}`` or - ``None``; ``put(candidate_hash, task_id, reward, feedback, rollout_file=...)`` + ``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 ``{ "::": {...} }`` rewritten on each put — fine for the run sizes here (a few thousand entries) and trivially portable. diff --git a/core/cap_evolve/gepa.py b/core/cap_evolve/gepa.py index 02620621..abd37301 100644 --- a/core/cap_evolve/gepa.py +++ b/core/cap_evolve/gepa.py @@ -64,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] @@ -159,7 +160,8 @@ 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 - replay = _replay_cached(cached, out_dir, task.id, tag) if cached else None + replay = (_replay_cached(cached, out_dir, task.id, tag, run_dir) + if cached else None) if replay is not None: scores.append(replay) continue @@ -179,11 +181,15 @@ def _eval_minibatch( "trace": _short(getattr(rollout, "trace", None))}, )) rfile = f"{task.id}__{tag}__t0.json" - (out_dir / rfile).write_text( - json.dumps({"input": task.input, "rollout": rollout.to_dict(), - "score": sc.to_dict()}, default=str), - encoding="utf-8", - ) + # ``_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 "", rollout_file=rfile) @@ -200,7 +206,8 @@ def _eval_minibatch( return result -def _replay_cached(cached: dict, out_dir: Path, task_id: str, tag: str) -> Score | None: +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 @@ -212,30 +219,52 @@ def _replay_cached(cached: dict, out_dir: Path, task_id: str, tag: str) -> Score 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: + 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")) - except (OSError, json.JSONDecodeError, ValueError): - return None + 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 {} - reward = float(cached.get("reward", 0.0)) # 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 and not dst.exists(): + if dst != src: try: + dst.unlink(missing_ok=True) os.link(src, dst) except OSError: try: - dst.write_text(json.dumps(rec, default=str), encoding="utf-8") - except OSError: - pass # ponytail: trajectories/ then falls back to REFLECTION.md excerpts + _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], diff --git a/core/cap_evolve/harness.py b/core/cap_evolve/harness.py index 1b604cb0..1ec3dba6 100644 --- a/core/cap_evolve/harness.py +++ b/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 diff --git a/core/tests/test_gepa_cache_traces.py b/core/tests/test_gepa_cache_traces.py index c0b704b7..948d270c 100644 --- a/core/tests/test_gepa_cache_traces.py +++ b/core/tests/test_gepa_cache_traces.py @@ -164,3 +164,201 @@ def _big(task, ctx, *, seed=0): assert len(trace) < 2000 and trace.endswith("…[truncated]") # and the cache itself never grew by the trace assert (run_dir.root / "eval_cache.json").stat().st_size < 1000 + + +# --- security / integrity of the re-materialization mechanism (review of #210) ------- +# +# ``eval_cache.json`` is plain JSON in the OPTIMIZER-WRITABLE run dir, so every field it +# carries is untrusted input — exactly #142's threat model. #197's tamper guard hashes +# ``.capevolve/project`` and excludes the run dir, so it does not cover this file. + + +@pytest.mark.parametrize("pointer", [ + "../../../../secrets.json", # relative traversal escapes out_dir + "sub/secrets.json", # subdir + "/etc/passwd", # absolute path REPLACES out_dir entirely +]) +def test_crafted_rollout_pointer_cannot_read_outside_the_rollouts_dir(setup, tmp_path, + pointer): + """A tampered ``rollout_file`` must not read any file outside ``rollouts//`` + into the reflective signal (i.e. into the optimizer LLM's prompt), and must not + hardlink an out-of-tree inode into the run dir. It degrades to an honest MISS.""" + from cap_evolve import gepa + from cap_evolve.cache import hash_candidate_dir + adapter, run_dir, cand, cache = setup + + secrets = tmp_path / "secrets.json" + secrets.write_text(json.dumps({ + "input": "x", "score": {"reward": 0.0}, + "rollout": {"task_id": "t0", "output": "sk-SUPER-SECRET-KEY", + "trace": "AWS_SECRET=xyz", "error": None}, + }), encoding="utf-8") + ptr = str(secrets) if pointer.startswith("/etc") else pointer + cache._data[f"{hash_candidate_dir(cand)}::t0"] = { + "reward": 0.0, "feedback": "fb", "rollout_file": ptr} + cache._flush() + + before = run_dir.spent.metric_calls + res = gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + assert run_dir.spent.metric_calls == before + 1, "crafted pointer must MISS, not serve" + raw = res.per_task[0].get("raw") or {} + assert "SUPER-SECRET" not in json.dumps(raw), "exfiltrated into the reflective signal" + assert "WRONG-ANSWER-for-t0" in str(raw.get("output")), "must be the real re-run" + + wd = run_dir.root / "work" / "x" + wd.mkdir(parents=True) + gepa._write_reflection(wd, res) + refl = (wd / "REFLECTION.md").read_text(encoding="utf-8") + assert "SUPER-SECRET" not in refl and "AWS_SECRET" not in refl + assert secrets.stat().st_nlink == 1, "out-of-tree inode hardlinked into the run dir" + + +def test_malformed_cached_reward_degrades_to_a_miss_instead_of_crashing(setup): + """A non-numeric ``reward`` in the cache must not raise mid-loop.""" + from cap_evolve import gepa + from cap_evolve.cache import hash_candidate_dir + adapter, run_dir, cand, cache = setup + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + cache._data[f"{hash_candidate_dir(cand)}::t0"]["reward"] = "not-a-number" + cache._flush() + + before = run_dir.spent.metric_calls + res = gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0001", seed=0) + assert run_dir.spent.metric_calls == before + 1 + assert res.reward == 0.0 + + +def test_fresh_write_never_rewrites_a_hardlinked_archived_rollout(setup): + """A cache hit hardlinks a prior rollout under the new tag. A later FRESH eval on + that same tag (a torn iteration + ``--resume`` reuses tags) must replace the + directory entry, not truncate the shared inode — otherwise it silently rewrites a + PREVIOUS iteration's archived evidence with another candidate's output.""" + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + train = run_dir.rollouts / "train" + + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + archived = train / "t0__mb_p_0000__t0.json" + original = archived.read_text(encoding="utf-8") + # cache hit under a second tag -> hardlink, shared inode + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_c_0000", seed=0) + linked = train / "t0__mb_c_0000__t0.json" + assert linked.stat().st_ino == archived.stat().st_ino, "expected a hardlink" + + # a REAL edit -> new hash -> MISS -> fresh write onto the reused tag + (cand / "cap.md").write_text("REAL-EDIT\n", encoding="utf-8") + adapter.run_target = lambda task, ctx, *, seed=0: __import__( + "cap_evolve").Rollout(task_id=task.id, output="CHILD-REAL-EDIT", trace="tr") + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_c_0000", seed=0) + + assert archived.read_text(encoding="utf-8") == original, \ + "fresh write rewrote a PREVIOUS iteration's archived rollout via the shared inode" + assert "CHILD-REAL-EDIT" in linked.read_text(encoding="utf-8") + assert linked.stat().st_ino != archived.stat().st_ino, "link must be broken, not reused" + + +def test_tag_reuse_replaces_a_prior_parents_trajectory(setup): + """#199's prompt claims ``trajectories/`` holds this minibatch VERBATIM. On a resumed + run a tag can already hold a DIFFERENT parent's rollout, so the replay must overwrite + it — serving parent A on disk while REFLECTION.md shows parent B is dishonest.""" + from cap_evolve import Rollout, gepa + adapter, run_dir, cand_a, cache = setup + cand_b = cand_a.parent / "cand_b" + cand_b.mkdir() + (cand_b / "cap.md").write_text("candidate B\n", encoding="utf-8") + train = run_dir.rollouts / "train" + + gepa._eval_minibatch(adapter, cand_a, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) # parent A owns tag mb_p_0000 + adapter.run_target = lambda task, ctx, *, seed=0: Rollout( + task_id=task.id, output="OUT-B", trace="tr-B") + gepa._eval_minibatch(adapter, cand_b, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0001", seed=0) # parent B cached under its own tag + # resumed run: frontier picks B, but tag mb_p_0000 exists from A -> cache HIT for B + res = gepa._eval_minibatch(adapter, cand_b, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + + on_disk = json.loads((train / "t0__mb_p_0000__t0.json") + .read_text(encoding="utf-8"))["rollout"]["output"] + in_prompt = str((res.per_task[0]["raw"] or {}).get("output")) + assert on_disk == "OUT-B" == in_prompt, \ + f"trajectories/ serves {on_disk!r} while REFLECTION.md shows {in_prompt!r}" + + +def test_cross_filesystem_link_failure_falls_back_to_a_byte_identical_copy(setup, + monkeypatch): + """``EXDEV`` / a filesystem without hardlinks must fall back to a real copy.""" + import errno + import os as _os + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + train = run_dir.rollouts / "train" + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + src = train / "t0__mb_p_0000__t0.json" + + def _no_link(*a, **k): + raise OSError(errno.EXDEV, "Cross-device link") + monkeypatch.setattr(_os, "link", _no_link) + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0001", seed=0) + + dst = train / "t0__mb_p_0001__t0.json" + assert dst.is_file() and dst.stat().st_ino != src.stat().st_ino, "expected a copy" + a, b = (json.loads(p.read_text(encoding="utf-8")) for p in (src, dst)) + assert a == b, "copy fallback is not equivalent to the link source" + + +def test_rematerialize_failure_is_logged_not_swallowed(setup, monkeypatch): + """When BOTH link and copy fail, trajectories/ silently lacks this task while the + prompt still claims VERBATIM completeness. Make it loud.""" + import errno + import os as _os + from cap_evolve import gepa + adapter, run_dir, cand, cache = setup + gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0000", seed=0) + + def _boom(*a, **k): + raise OSError(errno.EROFS, "Read-only file system") + monkeypatch.setattr(_os, "link", _boom) + monkeypatch.setattr(gepa, "_atomic_write", _boom) # only gepa's writer, not rundir's + res = gepa._eval_minibatch(adapter, cand, ["t0"], run_dir=run_dir, cache=cache, + tag="mb_p_0001", seed=0) + + assert res.reward == 0.0, "the Score itself is still complete" + kinds = [json.loads(ln)["kind"] for ln in + (run_dir.root / "events.jsonl").read_text(encoding="utf-8").splitlines() if ln] + assert "rollout_rematerialize_failed" in kinds, "silent failure looks like success" + + +def test_copy_step_trajectories_never_writes_through_a_hardlink(setup): + """``harness._copy_step_trajectories`` uses ``shutil.copyfile``, which DOES write + through a pre-existing hardlinked destination. It is safe only because ``_copy_tag`` + rmtree's ``trajectories/`` first — pin that invariant: no copied trajectory may share + an inode with the ``rollouts/`` record it came from, or copying would mutate the + archive.""" + from cap_evolve import gepa, harness + adapter, run_dir, cand, cache = setup + gepa._eval_minibatch(adapter, cand, ["t0", "t1"], run_dir=run_dir, cache=cache, + tag="seed", seed=0) + workdir = run_dir.root / "work" / "step" + workdir.mkdir(parents=True) + harness._copy_step_trajectories(adapter, run_dir, workdir, "train") + + copied = sorted((workdir / "trajectories").glob("*.json")) + assert copied, "expected the seed tag's rollouts to be copied" + rollout_inodes = {p.stat().st_ino for p in (run_dir.rollouts / "train").glob("*.json")} + assert not [p for p in copied if p.stat().st_ino in rollout_inodes], \ + "a trajectories/ copy shares a rollouts/ inode — a later copyfile would mutate it" + # and copying twice (a second step) must not corrupt the archive either + archived = (run_dir.rollouts / "train" / "t0__seed__t0.json") + original = archived.read_text(encoding="utf-8") + harness._copy_step_trajectories(adapter, run_dir, workdir, "train") + assert archived.read_text(encoding="utf-8") == original