Time every modality and ask which memory came back first

One stride over file order, truncated to the probe count, dropped the
tail of the suite: a 300-probe sample held 128 code, 86 pdf, 36
screenshot, 26 image and 24 video queries and no conversation query at
all, so every report built from it measured five of six modalities.
The budget is now split by suite weight with a floor of one probe per
modality, and a test pins that the frozen sample covers all six.

The set's question type rides along on each query, and a second metric
asks the question recall@10 cannot: when the answer lives in several
dated memories, did the newest come back first? A knowledge-update
question scores 1.000 on recall while the ranking still leads with the
value the speaker has since replaced, and that is the failure the next
task is meant to fix.

bench/leg_check.py

@@ -21,8 +21,8 @@ import time
 from pathlib import Path
 from typing import Any
 
-from bench.metrics import aggregate, score_query
-from bench.query_set import load_queries, scale_probes
+from bench.metrics import aggregate, newest_gold_first, score_query
+from bench.query_set import load_queries, scale_probes, source_times
 from bench.report import git_commit
 from bench.run import latency
 from bench.scratch import copy_store
@@ -189,6 +189,32 @@ def rank_all(service: Any, ids: dict[str, str], probes: list[Any], top_k: int) -
     return rankings, distinct, timings
 
 
+def _groups(
+    answerable: list[Any], scores: list[Any], rankings: list[list[str]], times: dict[str, str]
+) -> dict[str, Any]:
+    """recall@10 by modality and by the question type the set recorded, plus who came back first."""
+    by_modality: dict[str, list[float]] = {}
+    by_category: dict[str, list[float]] = {}
+    newest: dict[str, list[bool]] = {}
+    every_newest: list[bool] = []
+    for query, score, ranking in zip(answerable, scores, rankings, strict=True):
+        by_modality.setdefault(query.modality, []).append(score.recall_at_10)
+        by_category.setdefault(query.category or "unlabelled", []).append(score.recall_at_10)
+        flag = newest_gold_first(ranking, set(query.gold), times)
+        if flag is not None:
+            newest.setdefault(query.category or "unlabelled", []).append(flag)
+            every_newest.append(flag)
+    return {
+        "by_modality": {name: sum(rows) / len(rows) for name, rows in sorted(by_modality.items())},
+        "modality_counts": {name: len(rows) for name, rows in sorted(by_modality.items())},
+        "by_category": {name: sum(rows) / len(rows) for name, rows in sorted(by_category.items())},
+        "category_counts": {name: len(rows) for name, rows in sorted(by_category.items())},
+        "newest_gold_first": sum(every_newest) / len(every_newest) if every_newest else None,
+        "newest_gold_first_by_category": {name: sum(rows) / len(rows) for name, rows in sorted(newest.items())},
+        "newest_gold_first_counts": {name: len(rows) for name, rows in sorted(newest.items())},
+    }
+
+
 def measure(home: Path, scratch: Path, objects: Path, clip_index: Path, probes: int, models: dict[str, Path]) -> dict[str, Any]:
     """Score every configuration on the same probes, one warm service per configuration."""
     from ely_eye.config import Settings
@@ -207,6 +233,7 @@ def measure(home: Path, scratch: Path, objects: Path, clip_index: Path, probes:
     if not answerable:
         raise SystemExit("the frozen suite carries no answerable query; nothing to score")
     print(f"{len(answerable)} answerable probes", flush=True)
+    times = source_times()
 
     vectors = _dense_vectors(home / "data" / "ely_eye.sqlite")
     if vectors == 0 and any(config["dense"] for config in CONFIGS):
@@ -253,9 +280,6 @@ def measure(home: Path, scratch: Path, objects: Path, clip_index: Path, probes:
         ]
         summary = aggregate(scores)
         timing = latency(timings)
-        by_modality: dict[str, list[float]] = {}
-        for query, score in zip(answerable, scores, strict=True):
-            by_modality.setdefault(query.modality, []).append(score.recall_at_10)
         arms.append(
             {
                 "label": config["label"],
@@ -269,10 +293,7 @@ def measure(home: Path, scratch: Path, objects: Path, clip_index: Path, probes:
                 "p50_seconds": timing["p50"],
                 "p95_seconds": timing["p95"],
                 "median_distinct_sources": statistics.median(distinct),
-                "by_modality": {
-                    modality: sum(rows) / len(rows) for modality, rows in sorted(by_modality.items())
-                },
-                "modality_counts": {modality: len(rows) for modality, rows in sorted(by_modality.items())},
+                **_groups(answerable, scores, rankings, times),
             }
         )
         print(
@@ -326,6 +347,37 @@ def write_markdown(report: dict[str, Any], path: Path) -> None:
             + " |"
         )
     counts = report["arms"][0]["modality_counts"]
+
+    def category_table(title: str, field: str, sizes_field: str) -> list[str]:
+        """A table over the question types the sets label; unlabelled sets stay out of it."""
+        names = sorted({name for arm in report["arms"] for name in arm[field] if name != "unlabelled"})
+        if not names:
+            return []
+        lines = [
+            f"## {title}",
+            "",
+            "| leg configuration | " + " | ".join(names) + " |",
+            "| --- |" + " --- |" * len(names),
+        ]
+        for arm in report["arms"]:
+            values, sizes = arm[field], arm[sizes_field]
+            cells = []
+            for name in names:
+                if name not in values:
+                    cells.append("—")
+                elif field == "by_category":
+                    cells.append(f"{values[name]:.3f}")
+                else:
+                    cells.append(f"{values[name]:.2f} ({sizes[name]})")
+            lines.append(f"| {arm['label']} | " + " | ".join(cells) + " |")
+        labelled = report["arms"][0][sizes_field]
+        lines += [
+            "",
+            "_Queries per type: "
+            + ", ".join(f"{name} {size}" for name, size in labelled.items() if name != "unlabelled")
+            + "._",
+        ]
+        return lines
     path.write_text(
         "\n".join(
             [
@@ -375,6 +427,22 @@ def write_markdown(report: dict[str, Any], path: Path) -> None:
                 "",
                 f"_Query counts per modality: {', '.join(f'{name} {count}' for name, count in counts.items())}._",
                 "",
+                *category_table("recall@10 by question type", "by_category", "category_counts"),
+                "",
+                *category_table(
+                    "the newest of the gold memories came back first",
+                    "newest_gold_first_by_category",
+                    "newest_gold_first_counts",
+                ),
+                "",
+                (
+                    "recall@10 counts any gold memory, so a knowledge-update question can score 1.000 "
+                    "while the ranking still leads with the value the speaker has since replaced. The "
+                    "table above asks the sharper question: when the answer lives in several dated "
+                    "memories, is the newest one the first one returned? Only questions with at least two "
+                    "dated gold memories and at least one returned are counted."
+                ),
+                "",
                 (
                     "The picture leg is not a switch this harness flips per row: the product runs it only "
                     "when the sparse page comes back thin, so these rows measure the product with the leg "

bench/metrics.py

@@ -8,7 +8,7 @@ behaviour instead of recall, and never mixed into the recall denominator.
 from __future__ import annotations
 
 import math
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 
 Z_TWO_SIDED_95 = 1.959963984540054
@@ -89,6 +89,23 @@ def chance_recall_at_k(gold: int, corpus_size: int, k: int) -> float:
     return min(1.0, k / corpus_size)
 
 
+def newest_gold_first(ranked: Sequence[str], gold: set[str], times: Mapping[str, str]) -> bool | None:
+    """For a question answered by several memories, did the newest one come back first?
+
+    recall@10 counts any gold memory, which hides a knowledge-update failure: the older value is
+    still a memory, it is just no longer the answer. None when the question has fewer than two
+    dated gold memories or the ranking returned none of them.
+    """
+    dated = [source for source in gold if source in times]
+    if len(dated) < 2:
+        return None
+    newest = max(dated, key=lambda source: times[source])
+    found = [source for source in ranked if source in gold]
+    if not found:
+        return None
+    return found[0] == newest
+
+
 def score_query(
     query_id: str,
     ranked: Sequence[str],

bench/query_set.py

@@ -17,6 +17,7 @@ class Query:
     holder: str | None
     answerable: bool
     as_of: str | None
+    category: str | None = None
 
 
 
@@ -44,6 +45,7 @@ def load_queries(modalities: Sequence[str] = MODALITIES) -> list[Query]:
                     holder=record["scope"]["holder"],
                     answerable=answerable,
                     as_of=record.get("as_of"),
+                    category=(record.get("origin") or {}).get("category"),
                 )
             )
     if not queries:
@@ -65,19 +67,64 @@ def holder_index(modalities: Sequence[str] = MODALITIES) -> dict[str, str]:
 
 
 
+def source_times(modalities: Sequence[str] = MODALITIES) -> dict[str, str]:
+    """source_id -> when that source says it happened, from the corpus side of the suite."""
+    times: dict[str, str] = {}
+    for modality in modalities:
+        path = SETS / "corpus" / f"{modality}.jsonl"
+        if not path.is_file():
+            raise SystemExit(f"{path}: missing corpus index, run python -m bench.build_sets first")
+        for record in read_jsonl(path):
+            observed = record.get("observed_at")
+            if observed:
+                times[record["source_id"]] = str(observed)
+    return times
+
+
+
 SCALE_PROBES = 40
 
 
+def _shares(pools: dict[str, list[Query]], count: int) -> dict[str, int]:
+    """How many probes each modality gets: its share of the suite, never zero when there is room."""
+    names = sorted(pools)
+    total = sum(len(pools[name]) for name in names)
+    exact = {name: count * len(pools[name]) / total for name in names}
+    share = {name: int(exact[name]) for name in names}
+    short = count - sum(share.values())
+    for name in sorted(names, key=lambda name: (-(exact[name] - int(exact[name])), name))[:short]:
+        share[name] += 1
+    if count >= len(names):
+        for name in names:
+            while share[name] == 0:
+                donor = max((other for other in names if share[other] > 1), key=lambda other: share[other])
+                share[donor] -= 1
+                share[name] += 1
+    return {name: min(share[name], len(pools[name])) for name in names}
+
+
 def scale_probes(queries: Sequence[Query], count: int = SCALE_PROBES) -> list[Query]:
     """Queries to time at every corpus size.
 
     Latency does not need gold hits, and a truncated corpus keeps almost none of them,
-    so selecting probes by gold would silently drop the small sizes entirely.
+    so selecting probes by gold would silently drop the small sizes entirely. The budget is then
+    split across modalities by suite weight: one stride over file order truncated the tail and
+    timed 300 probes that held no conversation query at all.
     """
     if count <= 0:
         raise SystemExit("the scale curve needs at least one probe query")
-    answerable = [query for query in queries if query.answerable]
-    if not answerable:
+    pools: dict[str, list[Query]] = {}
+    for query in queries:
+        if query.answerable:
+            pools.setdefault(query.modality, []).append(query)
+    if not pools:
         raise SystemExit("the scale curve needs answerable queries")
-    stride = max(1, len(answerable) // count)
-    return answerable[::stride][:count]
+    shares = _shares(pools, count)
+    chosen: list[Query] = []
+    for modality in sorted(pools):
+        if shares[modality] == 0:
+            continue
+        pool = pools[modality]
+        stride = max(1, len(pool) // shares[modality])
+        chosen.extend(pool[::stride][: shares[modality]])
+    return chosen

bench/tests/test_leg_check.py

@@ -32,6 +32,11 @@ def test_the_report_prints_every_arm_with_its_numbers(tmp_path: Path) -> None:
         "median_distinct_sources": 4,
         "by_modality": {"code": 0.953, "pdf": 0.849},
         "modality_counts": {"code": 128, "pdf": 86},
+        "by_category": {"knowledge-update": 1.0, "temporal-reasoning": 0.947, "unlabelled": 0.84},
+        "category_counts": {"knowledge-update": 25, "temporal-reasoning": 19, "unlabelled": 256},
+        "newest_gold_first": 0.44,
+        "newest_gold_first_by_category": {"knowledge-update": 0.4, "temporal-reasoning": 0.526},
+        "newest_gold_first_counts": {"knowledge-update": 25, "temporal-reasoning": 19},
     }
     report: dict[str, Any] = {
         "commit": "abcdef123456",
@@ -63,4 +68,8 @@ def test_the_report_prints_every_arm_with_its_numbers(tmp_path: Path) -> None:
     assert (
         "53,668 atoms at 512 dimensions" in text and "44.7" in text
     ), "the picture index needs its shape in the report, and its build cost derived from a measured rate"
+    assert (
+        "knowledge-update" in text and "1.000" in text and "0.40 (25)" in text
+    ), "the two question types the contract names need their own recall and their own n"
+    assert "unlabelled" not in text, "an unlabelled bucket would hide which set a number came from"
     assert "abcdef123456" in text and "load_average" not in text

bench/tests/test_metrics.py

@@ -9,6 +9,7 @@ from bench.metrics import (
     chance_recall_at_k,
     first_hit_rank,
     ndcg_at_10,
+    newest_gold_first,
     recall_at_k,
     reciprocal_rank,
     score_query,
@@ -101,3 +102,28 @@ def test_aggregate_separates_answerable_from_unanswerable() -> None:
 def test_aggregate_needs_answerable_queries() -> None:
     with pytest.raises(SystemExit):
         aggregate([score_query("none", [], set())])
+
+
+TIMES = {
+    "old": "2023-01-01T00:00:00+00:00",
+    "new": "2024-06-01T00:00:00+00:00",
+    "undated": "",
+}
+
+
+def test_a_recall_hit_on_an_outdated_memory_is_not_reported_as_the_answer() -> None:
+    """recall@10 scores both orders the same; a replaced value is not the answer."""
+    assert newest_gold_first(["old", "new"], {"old", "new"}, TIMES) is False
+    assert newest_gold_first(["new", "old"], {"old", "new"}, TIMES) is True
+
+
+def test_a_question_with_one_dated_memory_has_no_newest_answer() -> None:
+    assert newest_gold_first(["old"], {"old"}, TIMES) is None
+    assert newest_gold_first(["old", "undated"], {"old", "undated"}, {"old": TIMES["old"]}) is None
+
+
+def test_a_ranking_that_returned_none_of_the_dated_memories_is_not_counted() -> None:
+    assert newest_gold_first(["x", "y"], {"old", "new"}, TIMES) is None
+    assert (
+        newest_gold_first(["old", "x"], {"old", "new"}, TIMES) is False
+    ), "a newest memory that never came back did not come back first"

bench/tests/test_query_set.py

@@ -0,0 +1,83 @@
+"""Who gets timed: the probe sample must cover the suite it claims to measure."""
+
+from __future__ import annotations
+
+from collections import Counter
+
+import pytest
+
+from bench.query_set import Query, load_queries, scale_probes
+
+
+def _suite(sizes: dict[str, int]) -> list[Query]:
+    queries: list[Query] = []
+    for modality, size in sizes.items():
+        queries.extend(
+            Query(
+                query_id=f"{modality}-{index}",
+                text="t",
+                modality=modality,
+                gold=frozenset({f"{modality}-s{index}"}),
+                holder="h",
+                answerable=True,
+                as_of=None,
+            )
+            for index in range(size)
+        )
+    return queries
+
+
+SIZES = {"code": 1237, "pdf": 766, "ui_screenshot": 320, "image": 240, "video": 215, "conversation": 150}
+
+
+def test_a_probe_sample_covers_every_modality_it_claims_to_measure() -> None:
+    probes = scale_probes(_suite(SIZES), 300)
+    mix = Counter(query.modality for query in probes)
+    assert len(probes) == 300
+    assert set(mix) == set(SIZES), f"a modality missing from the sample is unmeasured: {dict(mix)}"
+    assert mix["code"] > mix["conversation"], "the sample should still lean toward the big modalities"
+    assert probes == scale_probes(_suite(SIZES), 300), "the curve must be reproducible across runs"
+
+
+def test_the_frozen_sample_holds_a_conversation_query() -> None:
+    """The suite's smallest modality is the one a plain stride over file order used to drop."""
+    mix = Counter(query.modality for query in scale_probes(load_queries(), 300))
+    assert set(mix) == set(SIZES), f"the frozen sample is missing {sorted(set(SIZES) - set(mix))}: {dict(mix)}"
+
+
+def test_a_sample_smaller_than_the_modality_count_drops_the_smallest_pools() -> None:
+    mix = Counter(query.modality for query in scale_probes(_suite(SIZES), 2))
+    assert set(mix) == {"code", "pdf"}
+    assert sum(mix.values()) == 2
+
+
+def test_a_query_carries_the_category_its_set_recorded() -> None:
+    """LongMemEval labels each question's kind; the acceptance measurement groups by it."""
+    assert {query.category for query in load_queries(["conversation"])} == {
+        "knowledge-update",
+        "multi-session",
+        "single-session-assistant",
+        "single-session-preference",
+        "single-session-user",
+        "temporal-reasoning",
+    }
+    assert {query.category for query in load_queries(["pdf"])} == {None}, "a set without types stays ungrouped"
+
+
+def test_an_unanswerable_query_is_never_timed() -> None:
+    queries = [
+        Query(
+            query_id=f"q{index}",
+            text="t",
+            modality="pdf",
+            gold=frozenset() if index % 2 else frozenset({f"s{index}"}),
+            holder="h",
+            answerable=index % 2 == 0,
+            as_of=None,
+        )
+        for index in range(20)
+    ]
+    probes = scale_probes(queries, 4)
+    assert len(probes) == 4 and all(query.answerable for query in probes)
+    with pytest.raises(SystemExit, match="needs answerable"):
+        scale_probes([query for query in queries if not query.answerable], 4)