Memory Evaluation

A memory system fails quietly. It returns five records for every query, plausible-looking, and the model writes a confident answer around whichever ones it got. Nothing throws, nothing logs, and the first signal is a user saying "it used to be better at this."

This skill builds the instrument that makes those failures visible: a probe suite, five metrics, a negative control, and a longitudinal simulation that catches the decline no single-point measurement can.

It is the only skill in this repo that is never the wrong choice, and the one people skip.

  probes ──► write `given` into a clean store ──► run the query
                                                      │
                                                      ▼
                             compare against `expect` and `forbid`
                                                      │
   ┌──────────────────────────────────────────────────┤
   ▼                    ▼                  ▼          ▼
 recall@k             MRR            staleness    LEAK GATE
   │                                                  │
   └──► negative control: does the suite FAIL         └── any leak = fail,
        against a store that recalls nothing?             not a threshold

Use this when

• You have built any of the other skills in this repo. All of them end with "report recall@k on a probe suite", and this is the thing that produces it.
• You are about to tune a ranker, change a chunking strategy, or add a retrieval stage — none of which can be justified by the one query you happened to try.
• Recall feels like it is degrading as the store grows, and you need to know whether that is real.

There is no "do not use this when". The only bad version is a suite so thin it measures nothing, and the negative control below is how you find out whether yours is.

Workflow

1. Collect real probes. Not generated ones — generated probes test the generator. Ask the user for 20-30 things their agent should remember and be asked about, from actual usage. If they have a transcript log, mine it.
2. Ask what "correct" means. For each probe: which specific record must be retrieved? Probes with fuzzy expectations produce fuzzy metrics.
3. Scaffold the project into the directory the user names. Kebab-case folder, snake_case package, hatchling + pyproject.toml, src/ layout. This one can also be dropped into an existing memory project as a evals/ package — ask which the user wants.
4. Write the five modules: probe.py (the fixture), metrics.py, runner.py, simulate.py (longitudinal), report.py. Then cli.py and a CI workflow.
5. Run it against a deliberately broken store first. If the suite passes, the suite is the thing that is broken.

The decisions that matter

1. A probe is a fixture, not a question

The mistake is writing probes as questions and grading the agent's answer. That measures the model and the memory system together, with no way to attribute a failure to either.

A probe fixes the store contents, runs the retrieval, and grades the retrieval:

@dataclass
class Probe:
    name: str
    given: list[Record]       # written into a CLEAN store before the query
    query: str
    expect: list[str]         # record ids that MUST appear in the top k
    forbid: list[str] = ()    # ids that must NOT appear: stale, other-scope, superseded
    k: int = 5
    scope: str = "default"
    tags: list[str] = ()      # "multi-hop", "temporal", "negative", "paraphrase"

given is what makes probes reproducible. The store is rebuilt from scratch for each probe (or each probe group), so results do not depend on the order tests ran in or on what a previous probe left behind.

forbid is the half people omit and the half that catches the serious bugs. Cross-scope leaks and superseded facts are both silent, and neither shows up in recall@k — a suite with no forbid entries cannot see either.

2. Five metrics, and one of them is a gate

The distinction in that last row matters. Four of these are numbers you trade off against each other. The leak rate is not tunable: a single cross-scope hit is a failed run, regardless of how good recall was.

distinct@k is the one that catches near-duplicate pile-up — five phrasings of one fact filling all five slots while the second fact the answer needed never surfaces. Recall@k can look fine while this is happening, because the one fact you probed for is there.

3. The negative control is the most important test

def test_suite_detects_a_broken_store():
    """If this passes, the probes are being answered by the model's priors."""
    results = run(PROBES, store=NullStore())     # returns [] for every query
    assert results.recall_at_k < 0.2

Run it with three broken stores, because each catches a different flavour of worthless suite:

• NullStore — returns nothing. Recall must collapse.
• RandomStore — returns k random records. Recall must collapse.
• ShuffledStore — returns the right records in scrambled order. Recall@k holds (correctly — they are in the top k) but MRR must drop. If it does not, your probes have k so large that ranking is not being measured at all.

A suite that survives any of these is measuring nothing, and every number it produces afterwards is noise.

4. Longitudinal, not point-in-time

Most memory failures are gradual: recall holds at 100 records and degrades at 10,000 as duplicates accumulate and decay misfires. A single run cannot see it.

for session in range(200):
    write_synthetic_session(store, session)
    if session % 20 == 0:
        record(session, run(PROBES, store))

Plot three curves and read them together:

• recall@5 over store size — should be flat. A downward slope means the ranker is being outvoted by volume; you need MMR, decay, or consolidation.
• store size over sessions — linear early is fine. Still linear after consolidation runs means consolidation is not working.
• p95 query latency over store size — the point where a brute-force scan stops being acceptable, which is the only honest signal that it is time for a real index.

Seed the synthetic sessions deterministically. A drift measurement you cannot reproduce is an anecdote.

5. Results are a file in the repo, and CI fails on a regression

{"commit": "a3f91c2", "at": "2026-09-14T10:22:00Z", "probes": 34,
 "recall_at_5": 0.88, "mrr": 0.74, "distinct_at_5": 0.83,
 "staleness": 0.02, "leaks": 0, "p95_ms": 41}

Commit eval-results.json. Fail CI when recall drops by more than a tolerance (0.03 is reasonable) or when leaks > 0, exactly as you would for a performance budget.

Without this, the numbers get produced, admired once, and never compared. The diff across commits is the entire value; a single run tells you almost nothing.

Keep a per-probe breakdown too, not just aggregates. "recall fell from 0.88 to 0.84" is not actionable; "these three probes started failing" is.

Build it

<project>/
├── pyproject.toml
├── README.md
├── probes/
│   ├── core.yaml          # the hand-written probes, in a format humans can edit
│   └── negative.yaml      # probes where returning nothing is the right answer
└── src/<package>/
    ├── __init__.py
    ├── probe.py           # Probe + loading from yaml
    ├── stores.py          # NullStore, RandomStore, ShuffledStore - the controls
    ├── metrics.py         # the five metrics
    ├── runner.py          # rebuild store, run probes, collect
    ├── simulate.py        # longitudinal drift
    ├── report.py          # json + a readable table + the regression check
    └── cli.py             # run / control / simulate / compare / add

probe.py

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class Probe:
    name: str
    query: str
    given: list = field(default_factory=list)
    expect: list[str] = field(default_factory=list)
    forbid: list[str] = field(default_factory=list)
    k: int = 5
    scope: str = "default"
    tags: list[str] = field(default_factory=list)

    @property
    def is_negative(self) -> bool:
        """Retrieving nothing is the correct answer. These are easy to forget
        and they are what keep a system from retrieving on every single turn."""
        return not self.expect


def load(path: str | Path) -> list[Probe]:
    import yaml                                   # the one non-stdlib dep, for editability
    raw = yaml.safe_load(Path(path).read_text(encoding="utf-8")) or []
    probes = [Probe(**item) for item in raw]
    names = [p.name for p in probes]
    duplicates = {n for n in names if names.count(n) > 1}
    if duplicates:
        raise ValueError(f"duplicate probe names: {sorted(duplicates)}")
    return probes

stores.py

"""The controls. A suite that passes against these is measuring nothing."""

from __future__ import annotations

import random


class NullStore:
    def put(self, record): return record.id
    def search(self, query, *, k=5, **kwargs): return []


class RandomStore:
    def __init__(self, seed: int = 0):
        self.records, self.rng = [], random.Random(seed)

    def put(self, record):
        self.records.append(record)
        return record.id

    def search(self, query, *, k=5, **kwargs):
        sample = self.rng.sample(self.records, min(k, len(self.records)))
        return [Hit(record=r, score=0.5, why="random control") for r in sample]


class ShuffledStore:
    """Wraps a real store and scrambles the order. Recall@k survives; MRR must not."""

    def __init__(self, inner, seed: int = 0):
        self.inner, self.rng = inner, random.Random(seed)

    def put(self, record):
        return self.inner.put(record)

    def search(self, query, *, k=5, **kwargs):
        hits = self.inner.search(query, k=k, **kwargs)
        self.rng.shuffle(hits)
        return hits

metrics.py

from __future__ import annotations

from dataclasses import dataclass


@dataclass
class ProbeResult:
    probe: object
    returned: list[str]
    hit_all: bool
    first_rank: int | None
    leaked: list[str]
    stale: list[str]
    distinct: float
    latency_ms: float


def evaluate(probe, hits, *, is_stale) -> ProbeResult:
    returned = [h.record.id for h in hits[: probe.k]]
    expected = set(probe.expect)

    hit_all = expected.issubset(set(returned)) if expected else not returned
    first_rank = next((i + 1 for i, rid in enumerate(returned) if rid in expected), None)
    leaked = [rid for rid in returned if rid in set(probe.forbid)]
    stale = [h.record.id for h in hits[: probe.k] if is_stale(h.record)]

    texts = {h.record.text.strip().lower() for h in hits[: probe.k]}
    distinct = len(texts) / len(returned) if returned else 1.0

    return ProbeResult(probe, returned, hit_all, first_rank, leaked, stale, distinct, 0.0)


@dataclass
class Summary:
    probes: int
    recall_at_k: float
    mrr: float
    distinct_at_k: float
    staleness: float
    leaks: int
    negative_correct: float          # share of negative probes that correctly returned nothing
    p95_ms: float

    @property
    def passed(self) -> bool:
        return self.leaks == 0       # the gate. Everything else is a threshold.


def summarize(results: list[ProbeResult]) -> Summary:
    total = len(results) or 1
    positives = [r for r in results if r.probe.expect]
    negatives = [r for r in results if not r.probe.expect]
    latencies = sorted(r.latency_ms for r in results)

    return Summary(
        probes=len(results),
        recall_at_k=sum(r.hit_all for r in positives) / (len(positives) or 1),
        mrr=sum(1.0 / r.first_rank for r in positives if r.first_rank) / (len(positives) or 1),
        distinct_at_k=sum(r.distinct for r in results) / total,
        staleness=sum(len(r.stale) for r in results) / max(1, sum(len(r.returned) for r in results)),
        leaks=sum(len(r.leaked) for r in results),
        negative_correct=sum(not r.returned for r in negatives) / (len(negatives) or 1),
        p95_ms=latencies[int(len(latencies) * 0.95)] if latencies else 0.0,
    )

runner.py

"""Rebuild the store per probe group, run, time, collect."""

from __future__ import annotations

import time

from .metrics import evaluate, summarize


def run(probes, *, make_store, is_stale=lambda r: False, retrieve=None):
    """`make_store` returns a FRESH store. `retrieve(store, probe) -> list[Hit]`."""
    retrieve = retrieve or (lambda store, probe: store.search(
        probe.query, k=probe.k, scope=probe.scope))

    results = []
    for probe in probes:
        store = make_store()                       # clean slate: probes cannot interfere
        for record in probe.given:
            store.put(record)
        started = time.perf_counter()
        hits = retrieve(store, probe)
        elapsed = (time.perf_counter() - started) * 1000
        result = evaluate(probe, hits, is_stale=is_stale)
        result.latency_ms = elapsed
        results.append(result)
    return summarize(results), results


def run_controls(probes, *, make_store, **kwargs) -> dict:
    """The negative control. Every one of these must FAIL to recall."""
    from .stores import NullStore, RandomStore, ShuffledStore

    null_summary, _ = run(probes, make_store=NullStore, **kwargs)
    random_summary, _ = run(probes, make_store=lambda: RandomStore(seed=7), **kwargs)
    shuffled_summary, _ = run(probes, make_store=lambda: ShuffledStore(make_store()), **kwargs)

    return {
        "null_recall": null_summary.recall_at_k,             # must be < 0.2
        "random_recall": random_summary.recall_at_k,         # must be < 0.2
        "shuffled_mrr": shuffled_summary.mrr,                # must be well below the real mrr
        "valid": (null_summary.recall_at_k < 0.2
                  and random_summary.recall_at_k < 0.2),
    }

report.py

"""JSON for CI, a table for humans, and the regression check."""

from __future__ import annotations

import json
from dataclasses import asdict
from pathlib import Path

TOLERANCE = 0.03


def write(summary, results, path: Path, *, commit: str = "") -> None:
    payload = {
        "commit": commit,
        **{k: round(v, 4) if isinstance(v, float) else v for k, v in asdict(summary).items()},
        "per_probe": {r.probe.name: {"hit": r.hit_all, "rank": r.first_rank,
                                     "leaked": r.leaked} for r in results},
    }
    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")


def check_regression(current: dict, baseline_path: Path) -> tuple[bool, list[str]]:
    if not baseline_path.exists():
        return True, ["no baseline; recording this run as the baseline"]
    baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
    problems = []

    if current["leaks"] > 0:
        problems.append(f"LEAK GATE: {current['leaks']} forbidden record(s) returned")
    for metric, direction in (("recall_at_k", 1), ("mrr", 1), ("distinct_at_k", 1)):
        delta = current[metric] - baseline.get(metric, 0)
        if delta * direction < -TOLERANCE:
            problems.append(f"{metric}: {baseline[metric]:.3f} -> {current[metric]:.3f}")
    if current["staleness"] - baseline.get("staleness", 0) > TOLERANCE:
        problems.append(f"staleness rose to {current['staleness']:.3f}")

    # Per-probe: the actionable part.
    newly_failing = [
        name for name, entry in current["per_probe"].items()
        if not entry["hit"] and baseline.get("per_probe", {}).get(name, {}).get("hit")
    ]
    if newly_failing:
        problems.append("newly failing probes: " + ", ".join(sorted(newly_failing)))
    return not problems, problems

cli.py

• <command> run [--probes probes/core.yaml] — the table plus the JSON.
• <command> control — the three broken stores. Run this first, on a new suite, before trusting any number it produces.
• <command> simulate --sessions 200 --seed 7 — the drift curves.
• <command> compare <baseline.json> — the regression check, exit 1 on failure. This is the CI entry point.
• <command> add — an interactive helper to append a probe from a real query that just went wrong. The suite grows from failures; make that one command.

CI

- run: python -m <package>.cli control     # the suite is valid
- run: python -m <package>.cli run --out eval-results.json
- run: python -m <package>.cli compare eval-results.baseline.json

control running first is deliberate. A suite that has silently rotted into uselessness would otherwise pass every subsequent step.

Failure modes

• The suite passes against a null store. The probes are being answered by the model's priors, or expect is so loose that anything satisfies it. Every number the suite has ever produced is noise. Run control first, always.
• Generated probes. An LLM asked to write probes writes probes an LLM can answer, from the same distribution as the system under test. Hand-write them from real failures — thirty real probes beat three hundred generated ones.
• No forbid entries. Cross-scope leaks and stale facts are invisible to recall@k. Without forbid, the suite cannot see the two failure modes that actually hurt users.
• No negative probes. Every probe expecting a hit trains you toward a system that retrieves on every turn, injecting noise into turns that needed none. At least 15% of probes should expect nothing.
• Probes that share a store. Probe 7 passes because probe 3 left something behind. Rebuild per probe; the cost is milliseconds.
• k too large. At k=20, recall@k is trivially high and ranking is not being measured. The ShuffledStore control catches this — if MRR barely moves when order is scrambled, k is doing the work.
• Point-in-time only. Everything looks fine at 100 records. The failure is at 10,000, and only the longitudinal run sees it coming.
• Metrics with no baseline. Numbers produced, admired once, never compared. The diff is the value; commit the results file.
• Tuning against the suite until it passes. At that point the suite is a training set, not a test set. Keep a holdout: 20% of probes that you look at only when you think you are done.

Required tests

The harness needs its own tests — a broken measuring instrument is worse than none. All offline:

def test_null_store_fails_the_suite():
    summary, _ = run(PROBES, make_store=NullStore)
    assert summary.recall_at_k < 0.2


def test_random_store_fails_the_suite():
    summary, _ = run(PROBES, make_store=lambda: RandomStore(seed=1))
    assert summary.recall_at_k < 0.2


def test_shuffled_store_keeps_recall_but_drops_mrr(real_store_factory):
    real, _ = run(PROBES, make_store=real_store_factory)
    shuffled, _ = run(PROBES, make_store=lambda: ShuffledStore(real_store_factory()))
    assert shuffled.recall_at_k >= real.recall_at_k - 0.05
    assert shuffled.mrr < real.mrr - 0.1


def test_a_leak_fails_the_run_regardless_of_recall():
    summary, _ = run([leaking_probe()], make_store=leaky_store_factory)
    assert summary.leaks > 0
    assert summary.passed is False


def test_negative_probe_scores_correctly_on_empty_result():
    probe = Probe(name="n1", query="something nobody said", expect=[], given=[])
    summary, _ = run([probe], make_store=NullStore)
    assert summary.negative_correct == 1.0


def test_probes_do_not_share_state():
    first = Probe(name="a", query="x", given=[record("r1", "x")], expect=["r1"])
    second = Probe(name="b", query="x", given=[], expect=[])
    _, results = run([first, second], make_store=fresh_store)
    assert results[1].returned == []


def test_duplicate_probe_names_are_rejected(tmp_path):
    (tmp_path / "p.yaml").write_text("- {name: a, query: q}\n- {name: a, query: r}\n")
    with pytest.raises(ValueError, match="duplicate probe names"):
        load(tmp_path / "p.yaml")


def test_distinct_at_k_penalizes_near_duplicates():
    result = evaluate(probe_k3(), three_hits_same_text(), is_stale=lambda r: False)
    assert result.distinct < 0.4


def test_regression_check_fails_on_a_drop(tmp_path):
    baseline = {"recall_at_k": 0.90, "mrr": 0.75, "distinct_at_k": 0.8,
                "staleness": 0.01, "per_probe": {"a": {"hit": True}}}
    (tmp_path / "b.json").write_text(json.dumps(baseline))
    ok, problems = check_regression(
        {"recall_at_k": 0.80, "mrr": 0.75, "distinct_at_k": 0.8, "staleness": 0.01,
         "leaks": 0, "per_probe": {"a": {"hit": False, "leaked": []}}},
        tmp_path / "b.json")
    assert not ok
    assert any("recall_at_k" in p for p in problems)
    assert any("newly failing probes: a" in p for p in problems)


def test_regression_check_fails_on_any_leak(tmp_path):
    ok, problems = check_regression({"leaks": 1, "recall_at_k": 0.99, "mrr": 0.9,
                                     "distinct_at_k": 0.9, "staleness": 0.0,
                                     "per_probe": {}}, baseline_at(tmp_path))
    assert not ok and "LEAK GATE" in problems[0]

Verify

Follow references/evaluation.md — this skill implements that contract, so verification here is mostly self-referential and unusually strict:

1. Tier 0 — install, import, --help, run against the bundled example probes.
2. Tier 1 — pytest -q, offline. Every test above.
3. Tier 2 — <command> control against the user's real memory system. All three controls must fail to recall. If they do not, stop: report that the probe suite is invalid and do not present any other numbers from it. A number from an invalid suite is worse than no number, because it will be believed.
4. Tier 3 — <command> simulate --sessions 200 and read the three curves. Report the recall slope explicitly: flat is the pass condition, and a downward slope is a finding that should change what the user builds next.

Then tell the user what actually ran: the probe count, the split between positive and negative probes, the control results, the five metrics, and the drift slope. Lead with the control result — every other number depends on it.