feat(store): add per-run keying and config-drift guard to EvaluationDataStore (#347)

src/strands_evals/evaluation_data_store.py

@@ -9,24 +9,46 @@ class EvaluationDataStore(Protocol):
 
     Implementations can use any storage backend (local files, S3, databases, etc.)
     as long as they implement the load and save methods.
+
+    A case may be evaluated more than once — for statistical confidence you run the same case
+    N times and keep every result. `run_index` selects which run to read or write; it defaults
+    to `0`, so single-run callers can ignore it entirely and see the same behavior as before.
+    `completed_run_count` reports how many runs a case already has, which lets a caller resume an
+    interrupted N-run sweep by running only the runs that are missing.
     """
 
-    def load(self, case_name: str) -> EvaluationData | None:
-        """Load cached evaluation data by case name.
+    def load(self, case_name: str, run_index: int = 0) -> EvaluationData | None:
+        """Load cached evaluation data by case name and run index.
 
         Args:
             case_name: The name of the case to load results for.
+            run_index: Which run of the case to load. Defaults to `0` (the first/only run).
 
         Returns:
             The cached EvaluationData if found, None otherwise.
         """
         ...
 
-    def save(self, case_name: str, result: EvaluationData) -> None:
-        """Save evaluation data for a case.
+    def save(self, case_name: str, result: EvaluationData, run_index: int = 0) -> None:
+        """Save evaluation data for one run of a case.
 
         Args:
             case_name: The name of the case to save results for.
             result: The EvaluationData to save.
+            run_index: Which run of the case this result is. Defaults to `0`.
+        """
+        ...
+
+    def completed_run_count(self, case_name: str) -> int:
+        """Return how many consecutive runs of a case are already stored.
+
+        Counts stored runs starting at `run_index=0` and stopping at the first gap, so the
+        result is the next `run_index` a caller should write when topping up to N runs.
+
+        Args:
+            case_name: The name of the case to count runs for.
+
+        Returns:
+            The number of consecutive stored runs, or `0` if none exist.
         """
         ...

src/strands_evals/local_file_task_result_store.py

@@ -1,39 +1,142 @@
+import json
 from pathlib import Path
 
 from .types.evaluation import EvaluationData
 
+# A case's runs live in a per-case subdirectory as run_0.json, run_1.json, .... A pre-existing
+# flat `{case_name}.json` from the single-run layout is treated as that case's run 0, so old
+# stores keep loading without a migration step.
+_RUN_PREFIX = "run_"
+_MANIFEST_NAME = "_manifest.json"
+
+
+class ConfigDriftError(RuntimeError):
+    """Raised when a store's recorded config no longer matches the current experiment.
+
+    Mixing results produced under different evaluators, cases, or model config into one report
+    silently compares apples to oranges. When the store already holds results written under a
+    different `config_hash`, resuming raises this instead of appending, unless the caller passes
+    `overwrite=True` to discard the stale results and start fresh.
+    """
+
 
 class LocalFileTaskResultStore:
     """Task result store backed by local JSON files.
 
-    Saves one JSON file per case in the specified directory,
-    using the case name as the filename.
+    Each case gets its own subdirectory holding one JSON file per run (`run_0.json`,
+    `run_1.json`, ...), so a case can be evaluated N times and every result is kept. A legacy
+    flat `{case_name}.json` file from the original single-run layout is read as that case's
+    run 0, so existing stores keep working unchanged.
+
+    An optional `config_hash` guards against config drift: pass the hash of the current
+    experiment's evaluators/cases/model config and the store records it in a `_manifest.json`.
+    A later run whose hash differs raises `ConfigDriftError` rather than mixing incompatible
+    results, unless the store is opened with `overwrite=True`.
     """
 
-    def __init__(self, directory: str | Path):
+    def __init__(self, directory: str | Path, config_hash: str | None = None, overwrite: bool = False):
+        """Open (or create) a store rooted at `directory`.
+
+        Args:
+            directory: Directory that holds the per-case result subdirectories.
+            config_hash: Optional hash of the current experiment config. When given, it is
+                checked against the stored manifest (drift raises unless `overwrite=True`) and
+                then recorded.
+            overwrite: When True, discard any existing results and manifest before use. Use this
+                to start fresh after an intentional config change.
+
+        Raises:
+            ConfigDriftError: If `config_hash` is given, differs from the stored manifest, and
+                `overwrite` is False.
+        """
         self._directory = Path(directory)
         self._directory.mkdir(parents=True, exist_ok=True)
 
-    def load(self, case_name: str) -> EvaluationData | None:
-        """Load a cached task result from a JSON file.
+        if overwrite:
+            self._clear()
+
+        if config_hash is not None:
+            self._guard_config(config_hash, overwrite=overwrite)
+
+    def _clear(self) -> None:
+        """Remove all stored case results and the manifest, leaving an empty store directory."""
+        for child in self._directory.iterdir():
+            if child.is_dir():
+                for f in child.glob(f"{_RUN_PREFIX}*.json"):
+                    f.unlink()
+                # Drop the now-empty case directory; ignore if other files remain.
+                if not any(child.iterdir()):
+                    child.rmdir()
+            elif child.suffix == ".json":
+                child.unlink()
+
+    def _manifest_path(self) -> Path:
+        return self._directory / _MANIFEST_NAME
+
+    def _guard_config(self, config_hash: str, overwrite: bool) -> None:
+        """Check the stored config hash against `config_hash`, then record the current one."""
+        manifest_path = self._manifest_path()
+        if manifest_path.exists() and not overwrite:
+            stored = json.loads(manifest_path.read_text()).get("config_hash")
+            if stored is not None and stored != config_hash:
+                raise ConfigDriftError(
+                    "Experiment configuration changed since results were last written "
+                    f"(stored config_hash={stored}, current={config_hash}). "
+                    "Open the store with overwrite=True to discard cached results and start fresh."
+                )
+
+        manifest_path.write_text(json.dumps({"config_hash": config_hash}, indent=2))
+
+    def _case_dir(self, case_name: str) -> Path:
+        return self._directory / case_name
+
+    def _run_path(self, case_name: str, run_index: int) -> Path:
+        return self._case_dir(case_name) / f"{_RUN_PREFIX}{run_index}.json"
+
+    def _legacy_path(self, case_name: str) -> Path:
+        return self._directory / f"{case_name}.json"
+
+    def load(self, case_name: str, run_index: int = 0) -> EvaluationData | None:
+        """Load a cached task result for one run of a case.
 
         Args:
             case_name: The name of the case to load results for.
+            run_index: Which run to load. Defaults to `0`.
 
         Returns:
-            The cached EvaluationData if the file exists, None otherwise.
+            The cached EvaluationData if the run exists, None otherwise. Run 0 also reads a
+            legacy flat `{case_name}.json` file when no per-run file is present.
         """
-        path = self._directory / f"{case_name}.json"
-        if not path.exists():
-            return None
-        return EvaluationData.model_validate_json(path.read_text())
+        path = self._run_path(case_name, run_index)
+        if path.exists():
+            return EvaluationData.model_validate_json(path.read_text())
+
+        if run_index == 0:
+            legacy = self._legacy_path(case_name)
+            if legacy.exists():
+                return EvaluationData.model_validate_json(legacy.read_text())
+
+        return None
 
-    def save(self, case_name: str, result: EvaluationData) -> None:
-        """Save a task result to a JSON file.
+    def save(self, case_name: str, result: EvaluationData, run_index: int = 0) -> None:
+        """Save a task result for one run of a case.
 
         Args:
             case_name: The name of the case to save results for.
             result: The EvaluationData to save.
+            run_index: Which run this result is. Defaults to `0`.
         """
-        path = self._directory / f"{case_name}.json"
+        path = self._run_path(case_name, run_index)
+        path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(result.model_dump_json(indent=2))
+
+    def completed_run_count(self, case_name: str) -> int:
+        """Return the number of consecutive stored runs for a case.
+
+        Counts from `run_index=0` upward and stops at the first missing run, so the return value
+        is the next index to write. A legacy flat file counts as run 0.
+        """
+        count = 0
+        while self.load(case_name, count) is not None:
+            count += 1
+        return count

tests/strands_evals/cli/test_run.py

@@ -313,8 +313,8 @@ def test_run_data_store_creates_directory(experiment_file: Path, tmp_path: Path)
     )
     assert exit_code == 0
     assert store_dir.is_dir()
-    # One file per case (case 'c1' in the fixture).
-    assert (store_dir / "c1.json").exists()
+    # One directory per case and one file per run (case 'c1' in the fixture).
+    assert (store_dir / "c1" / "run_0.json").exists()
 
 
 def test_run_exit_zero_overrides_fail_on(experiment_file: Path, tmp_path: Path):

tests/strands_evals/test_experiment.py

@@ -1889,13 +1889,19 @@ class DictEvaluationDataStore:
     """Simple in-memory store for testing."""
 
     def __init__(self):
-        self._data: dict[str, EvaluationData] = {}
+        self._data: dict[tuple[str, int], EvaluationData] = {}
 
-    def load(self, case_name: str) -> EvaluationData | None:
-        return self._data.get(case_name)
+    def load(self, case_name: str, run_index: int = 0) -> EvaluationData | None:
+        return self._data.get((case_name, run_index))
 
-    def save(self, case_name: str, result: EvaluationData) -> None:
-        self._data[case_name] = result
+    def save(self, case_name: str, result: EvaluationData, run_index: int = 0) -> None:
+        self._data[(case_name, run_index)] = result
+
+    def completed_run_count(self, case_name: str) -> int:
+        count = 0
+        while (case_name, count) in self._data:
+            count += 1
+        return count
 
 
 class TestEvaluationDataStore:

tests/strands_evals/test_local_file_task_result_store.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from strands_evals.local_file_task_result_store import LocalFileTaskResultStore
+from strands_evals.local_file_task_result_store import ConfigDriftError, LocalFileTaskResultStore
 from strands_evals.types import EvaluationData
 
 
@@ -44,10 +44,95 @@ def test_creates_directory(self, tmp_path):
         LocalFileTaskResultStore(directory=new_dir)
         assert new_dir.exists()
 
-    def test_save_writes_json_file(self, store, evaluation_data, tmp_path):
+    def test_save_writes_run_file(self, store, evaluation_data, tmp_path):
         store.save("math_case", evaluation_data)
-        file_path = tmp_path / "results" / "math_case.json"
+        file_path = tmp_path / "results" / "math_case" / "run_0.json"
         assert file_path.exists()
         data = json.loads(file_path.read_text())
         assert data["input"] == "What is 2+2?"
         assert data["actual_output"] == "4"
+
+
+class TestPerRunKeying:
+    """A case can be evaluated N times; each run is stored and read independently."""
+
+    def _data(self, name, output):
+        return EvaluationData(input="q", actual_output=output, name=name, expected_output="q")
+
+    def test_runs_are_stored_and_loaded_independently(self, store):
+        store.save("c", self._data("c", "run-a"), run_index=0)
+        store.save("c", self._data("c", "run-b"), run_index=1)
+
+        assert store.load("c", 0).actual_output == "run-a"
+        assert store.load("c", 1).actual_output == "run-b"
+        assert store.load("c", 2) is None
+
+    def test_completed_run_count_counts_consecutive_runs(self, store):
+        assert store.completed_run_count("c") == 0
+        store.save("c", self._data("c", "0"), run_index=0)
+        store.save("c", self._data("c", "1"), run_index=1)
+        assert store.completed_run_count("c") == 2
+
+    def test_completed_run_count_stops_at_first_gap(self, store):
+        store.save("c", self._data("c", "0"), run_index=0)
+        store.save("c", self._data("c", "2"), run_index=2)  # gap at 1
+        assert store.completed_run_count("c") == 1
+
+    def test_run_files_live_in_per_case_subdir(self, store, tmp_path):
+        store.save("c", self._data("c", "x"), run_index=3)
+        assert (tmp_path / "results" / "c" / "run_3.json").exists()
+
+    def test_legacy_flat_file_is_read_as_run_zero(self, tmp_path):
+        # A store written by the old single-run layout: flat {case_name}.json.
+        results_dir = tmp_path / "results"
+        results_dir.mkdir()
+        legacy = self._data("legacy", "old")
+        (results_dir / "legacy.json").write_text(legacy.model_dump_json())
+
+        store = LocalFileTaskResultStore(directory=results_dir)
+        assert store.load("legacy", 0).actual_output == "old"
+        assert store.completed_run_count("legacy") == 1
+
+
+class TestConfigDrift:
+    """An optional config hash prevents mixing results from incompatible experiment configs."""
+
+    def _data(self, name):
+        return EvaluationData(input="q", actual_output="a", name=name, expected_output="q")
+
+    def test_first_open_records_hash(self, tmp_path):
+        d = tmp_path / "results"
+        LocalFileTaskResultStore(directory=d, config_hash="hash-1")
+        manifest = json.loads((d / "_manifest.json").read_text())
+        assert manifest["config_hash"] == "hash-1"
+
+    def test_reopen_with_same_hash_is_allowed(self, tmp_path):
+        d = tmp_path / "results"
+        s1 = LocalFileTaskResultStore(directory=d, config_hash="hash-1")
+        s1.save("c", self._data("c"))
+        s2 = LocalFileTaskResultStore(directory=d, config_hash="hash-1")
+        assert s2.load("c") is not None
+
+    def test_reopen_with_different_hash_raises(self, tmp_path):
+        d = tmp_path / "results"
+        LocalFileTaskResultStore(directory=d, config_hash="hash-1")
+        with pytest.raises(ConfigDriftError, match="configuration changed"):
+            LocalFileTaskResultStore(directory=d, config_hash="hash-2")
+
+    def test_overwrite_discards_results_and_updates_hash(self, tmp_path):
+        d = tmp_path / "results"
+        s1 = LocalFileTaskResultStore(directory=d, config_hash="hash-1")
+        s1.save("c", self._data("c"))
+        assert s1.load("c") is not None
+
+        s2 = LocalFileTaskResultStore(directory=d, config_hash="hash-2", overwrite=True)
+        assert s2.load("c") is None
+        manifest = json.loads((d / "_manifest.json").read_text())
+        assert manifest["config_hash"] == "hash-2"
+
+    def test_no_hash_skips_the_guard(self, tmp_path):
+        d = tmp_path / "results"
+        LocalFileTaskResultStore(directory=d, config_hash="hash-1")
+        # Opening without a hash must not raise even though a manifest exists.
+        store = LocalFileTaskResultStore(directory=d)
+        assert store is not None