From dd12c67a9ba4ffa669c6f97fe5dcb7c71274c704 Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 15:09:52 +0530 Subject: [PATCH 1/9] Phase 9 task 1: keep the memory of a run With --keep, the memory arm runs each question over a local database of its own, named by the sample, the extractor variant and a digest of its prompt, how the facts cite their session, the memory ranking, the last commit that changed src/hotmemory, and the question id. The database stays after the question, and a question that raises deletes it. The runner refuses --keep when src/hotmemory has changes that are not committed, when a kept database of the run exists already, and with an arm other than memory. The result file records the name. plan.md task 1 now says one database for each question: BM25 ranks the whole table before the namespace filter, so a shared table changes the block. --- bench/e2e.py | 153 +++++++++++++++++++++++++++++++++------- bench/tests/test_e2e.py | 98 +++++++++++++++++++++++++ docs/internal/plan.md | 9 ++- 3 files changed, 233 insertions(+), 27 deletions(-) diff --git a/bench/e2e.py b/bench/e2e.py index ff8b09f..eb49d51 100644 --- a/bench/e2e.py +++ b/bench/e2e.py @@ -18,6 +18,7 @@ import json import multiprocessing import random +import subprocess import time import uuid from collections.abc import Callable, Collection, Iterator, Mapping, Sequence @@ -62,6 +63,7 @@ RECALL_K = 30 RECALL_BUDGET = 8000 EPISODES = ("none", "cited", "search") +KEPT_RUN = "kept" COST_LIMIT = 50.0 EXTRACT_OVERHEAD = 700 ACTOR = "longmemeval-extractor" @@ -381,12 +383,13 @@ def accuracy(found: Sequence[bool]) -> dict[str, Any]: @contextmanager def local_store( - url: str, run_id: str, embedder: Embedder | None = None + url: str, name: str, embedder: Embedder | None = None, keep: bool = False ) -> Iterator[tuple[dict[str, Store], Callable[[Sequence[str]], None]]]: """Yield a HotdataStore for each ranking, and a call that empties a namespace. - The stores share one new local database, and the database is deleted after. Without - `embedder`, the stores embed with `embed.openai_embedder()`. + The stores share the local database named `name`, which `provision` opens or creates. + The database is deleted after, unless `keep` is true and the block exits without an + error. Without `embedder`, the stores embed with `embed.openai_embedder()`. """ import pyarrow as pa from hotdata_framework import HotdataClient @@ -396,7 +399,7 @@ def local_store( client = HotdataClient("local", "local", host=url) embedder = embedder or embed.openai_embedder() store = HotdataStore.provision( - f"hotmemory-bench-{run_id}", + name, embedder=embedder, model=embed.MODEL, dimensions=embed.DIMENSIONS, @@ -413,23 +416,85 @@ def empty(namespace: Sequence[str]) -> None: store._load_all([(table, rows, "delete"), (CUE_TABLE, rows, "delete")]) stores: dict[str, Store] = { - name: HotdataStore( + ranking: HotdataStore( client, store.database, embedder=embedder, model=embed.MODEL, dimensions=embed.DIMENSIONS, - ranking=name, # type: ignore[arg-type] + ranking=ranking, # type: ignore[arg-type] ) - for name in RANKINGS + for ranking in RANKINGS } + kept = False try: yield stores, empty + kept = keep finally: - client.delete_managed_database(store.database) + if not kept: + client.delete_managed_database(store.database) client.close() +def existing_databases(url: str, names: Collection[str]) -> list[str]: + """Return the names in `names` that a local database has, sorted.""" + from hotdata_framework import HotdataClient + + client = HotdataClient("local", "local", host=url) + try: + found = {database.description for database in client.list_managed_databases()} + finally: + client.close() + return sorted(name for name in found & set(names) if name is not None) + + +def library_commit(root: Path = HERE.parent) -> str: + """Return the last commit that changed `src/hotmemory`. + + Raises RuntimeError if `src/hotmemory` holds a change that is not committed, because + then no commit names the code that wrote the memory. + """ + changed = subprocess.run( + ["git", "status", "--porcelain", "--", "src/hotmemory"], + capture_output=True, + text=True, + cwd=root, + check=True, + ) + if changed.stdout.strip(): + raise RuntimeError("src/hotmemory has changes that are not committed") + last = subprocess.run( + ["git", "log", "-1", "--format=%h", "--", "src/hotmemory"], + capture_output=True, + text=True, + cwd=root, + check=True, + ) + return last.stdout.strip() + + +def kept_name(sample: str, config: Config, commit: str) -> str: + """Return the name of the kept memory of `sample` under `config` and the library `commit`. + + The name holds what decides the memory that `capture` writes: the sample, the extractor + variant with 6 hex digits of the SHA-256 of its model and prompt, how the facts cite + their session (`episodes`, `sources`, or `plain`), the memory ranking, and the commit. + The database of one question appends `-`. + """ + variant = config.variant + digest = hashlib.sha256(f"{variant.model}\n{variant.prompt}".encode()).hexdigest()[:6] + if config.session_episodes: + cites = "episodes" + elif config.fact_sources: + cites = "sources" + else: + cites = "plain" + return ( + f"hotmemory-kept-{sample}-{config.extractor}{digest}-{cites}-" + f"{config.memory_ranking}-{commit}" + ) + + @dataclass(frozen=True) class Job: """What every worker of one run shares.""" @@ -439,6 +504,7 @@ class Job: config: Config replay: bool total: int + kept: str | None = None @dataclass(frozen=True) @@ -496,35 +562,55 @@ def row_of(question: Question, result: dict[str, Arm]) -> dict[str, Any]: def run_shard(job: Job, shard: Shard) -> Done: """Run the questions of `shard` in order, over a new local database of its own. - The worker opens its own model cache, embedder, and store, so it can run in its own - process. It prints one line for each question. + With `job.kept`, each question runs over a new database named `-`, and the database keeps the memory of the question after the run. A question that + raises deletes its database. The worker opens its own model cache, embedder, and + store, so it can run in its own process. It prints one line for each question. """ from bench.models import Offline, Remote models = CachedModels(Offline() if job.replay else Remote(), CACHE_DIR / "replies.sqlite") embedder = embed.openai_embedder() rows: list[dict[str, Any]] = [] + + def finish(question: Question, result: dict[str, Arm]) -> None: + rows.append(row_of(question, result)) + marks = " ".join( + f"{arm}={'yes' if found.correct else 'no'}" for arm, found in result.items() + ) + print( + f"{shard.run_id} {len(rows)}/{len(shard.questions)} of {job.total} " + f"{question.id} {marks} spent ${models.spent:.2f}", + flush=True, + ) + try: - with local_store(job.url, shard.run_id, embedder) as (stores, empty): + if job.kept is not None: for question in shard.questions: - result = run_question( - question, models, stores, empty, shard.run_id, job.arms, job.config - ) - rows.append(row_of(question, result)) - marks = " ".join( - f"{arm}={'yes' if found.correct else 'no'}" for arm, found in result.items() - ) - print( - f"{shard.run_id} {len(rows)}/{len(shard.questions)} of {job.total} " - f"{question.id} {marks} spent ${models.spent:.2f}", - flush=True, - ) + name = f"{job.kept}-{question.id}" + with local_store(job.url, name, embedder, keep=True) as (stores, _): + result = run_question( + question, models, stores, _keep, KEPT_RUN, job.arms, job.config + ) + finish(question, result) + else: + name = f"hotmemory-bench-{shard.run_id}" + with local_store(job.url, name, embedder) as (stores, empty): + for question in shard.questions: + result = run_question( + question, models, stores, empty, shard.run_id, job.arms, job.config + ) + finish(question, result) finally: models.close() embedder.close() return Done(rows, models.spent, models.calls, embedder.misses) +def _keep(namespace: Sequence[str]) -> None: + """Do nothing: a kept run leaves the namespace of each question in its database.""" + + def main(argv: Sequence[str] | None = None) -> Path | None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument( @@ -564,6 +650,11 @@ def main(argv: Sequence[str] | None = None) -> Path | None: default=1, help="processes that run the questions, each over a database of its own", ) + parser.add_argument( + "--keep", + action="store_true", + help="keep the memory of each question in a local database of its own", + ) parser.add_argument("--tag", default="", help="a suffix for the name of the result file") parser.add_argument("--url", default="http://localhost:3000") args = parser.parse_args(argv) @@ -572,6 +663,8 @@ def main(argv: Sequence[str] | None = None) -> Path | None: parser.error(f"arms must be among {', '.join(ARMS)}") if args.workers < 1: parser.error("workers must be at least 1") + if args.keep and arms != ("memory",): + parser.error("--keep needs --arms memory") try: config = Config( retrieval_ranking=args.retrieval_ranking, @@ -606,12 +699,23 @@ def main(argv: Sequence[str] | None = None) -> Path | None: print("no model was called; pass --run to run, or --replay to run from the cache") return None + kept = None + if args.keep: + try: + kept = kept_name(args.sample, config, library_commit()) + except RuntimeError as error: + parser.error(str(error)) + found = existing_databases(args.url, [f"{kept}-{question.id}" for question in questions]) + if found: + parser.error(f"{len(found)} kept databases exist already, for example {found[0]}") + print(f"keeping the memory as {kept}-") + commit = _commit() started = datetime.now(UTC) run_id = f"{started:%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" clock_start = time.perf_counter() shards = shards_of(questions, args.workers, run_id) - job = Job(args.url, arms, config, args.replay, len(questions)) + job = Job(args.url, arms, config, args.replay, len(questions), kept) if len(shards) == 1: done = [run_shard(job, shards[0])] else: @@ -667,6 +771,7 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "fact_sources": config.fact_sources, "session_episodes": config.session_episodes, "episodes": config.episodes, + "kept": kept, }, "arms": { "full": "every session of the question", diff --git a/bench/tests/test_e2e.py b/bench/tests/test_e2e.py index f4231e3..a7fa80a 100644 --- a/bench/tests/test_e2e.py +++ b/bench/tests/test_e2e.py @@ -1,4 +1,7 @@ import json +import os +import subprocess +import uuid from collections.abc import Sequence from dataclasses import replace from pathlib import Path @@ -15,8 +18,13 @@ choose_sample, episode_key, estimate, + existing_databases, extractor, fact_source, + kept_name, + library_commit, + local_store, + main, run_question, sample_ids, shards_of, @@ -350,3 +358,93 @@ def test_capture_session_replays_the_same_extractor_prompts( capture_session(memory, ("bench", run, question.id), session, models, config, []) seen.append([p for m, p in models.prompts if m == EXTRACTOR]) assert seen[0] == seen[1] + + +def test_kept_name_holds_what_decides_the_memory() -> None: + config = Config(extractor="detail", session_episodes=True, episodes="search", budget=64000) + name = kept_name("a", config, "b5443ac") + assert name.startswith("hotmemory-kept-a-detail") + assert name.endswith("-episodes-fused-b5443ac") + after_capture = replace(config, episodes="cited", budget=32000, k=10) + assert kept_name("a", after_capture, "b5443ac") == name + for other in ( + kept_name("b", config, "b5443ac"), + kept_name("a", config, "1234567"), + kept_name("a", replace(config, extractor="baseline"), "b5443ac"), + kept_name("a", replace(config, memory_ranking="vector"), "b5443ac"), + kept_name("a", Config(extractor="detail"), "b5443ac"), + kept_name("a", Config(extractor="detail", fact_sources=False), "b5443ac"), + ): + assert other != name + + +def test_kept_name_changes_with_the_extractor_prompt(monkeypatch: pytest.MonkeyPatch) -> None: + config = Config(extractor="detail") + name = kept_name("a", config, "b5443ac") + variant = prompts.VARIANTS["detail"] + changed = prompts.Variant(variant.prompt + " ", variant.model) + monkeypatch.setitem(prompts.VARIANTS, "detail", changed) + assert kept_name("a", config, "b5443ac") != name + + +def test_library_commit_refuses_a_change_that_is_not_committed(tmp_path: Path) -> None: + def git(*args: str) -> str: + done = subprocess.run( + ["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], + cwd=tmp_path, + capture_output=True, + text=True, + check=True, + ) + return done.stdout.strip() + + git("init", "-q") + code = tmp_path / "src" / "hotmemory" / "a.py" + code.parent.mkdir(parents=True) + code.write_text("x = 1\n") + git("add", ".") + git("commit", "-q", "-m", "code") + first = git("rev-parse", "--short", "HEAD") + (tmp_path / "other.txt").write_text("y\n") + git("add", ".") + git("commit", "-q", "-m", "other") + assert library_commit(tmp_path) == first + code.write_text("x = 2\n") + with pytest.raises(RuntimeError, match="not committed"): + library_commit(tmp_path) + + +def test_keep_needs_the_memory_arm_only(capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit): + main(["--keep", "--arms", "retrieval,memory"]) + assert "--keep needs --arms memory" in capsys.readouterr().err + + +def _embed(texts: Sequence[str]) -> list[list[float]]: + return [[1.0] + [0.0] * 1535 for _ in texts] + + +@pytest.mark.hotdata +def test_local_store_keeps_the_database_only_on_success() -> None: + url = os.environ.get("HOTMEMORY_TEST_URL") + if not url: + pytest.skip("HOTMEMORY_TEST_URL is not set") + kept = f"hotmemory-test-kept-{uuid.uuid4().hex[:8]}" + failed = f"{kept}-failed" + temporary = f"{kept}-temporary" + with local_store(url, kept, _embed, keep=True): + pass + with pytest.raises(ValueError), local_store(url, failed, _embed, keep=True): + raise ValueError("stop") + with local_store(url, temporary, _embed): + pass + try: + assert existing_databases(url, [kept, failed, temporary]) == [kept] + finally: + from hotdata_framework import HotdataClient + + client = HotdataClient("local", "local", host=url) + for database in client.list_managed_databases(): + if database.description == kept: + client.delete_managed_database(database) + client.close() diff --git a/docs/internal/plan.md b/docs/internal/plan.md index 1e49911..843e01f 100644 --- a/docs/internal/plan.md +++ b/docs/internal/plan.md @@ -131,10 +131,13 @@ lost every finished question, because a run writes its result file only at the e Tasks: 1. Keep the memory of a run. With `--keep`, the runner does not empty the namespaces, and - it keeps the database of each worker. The name of the database states what decides + it keeps one database for each question. The name of the database states what decides the memory: the sample, the extractor variant, the episode setting, the fact sources, - the ranking, the worker count, and the last commit that changed `src/hotmemory`. The - runner records the name in the result file. + the ranking, the last commit that changed `src/hotmemory`, and the question id. The + runner records the name in the result file. One database holds one question, because + BM25 ranks the whole table before the namespace filter applies. In a shared table, the + memory of other questions changes the block and the known facts of `capture`. The + owner agreed to this layout on 2026-10-16. 2. Reuse the memory. With `--reuse`, the runner opens the kept databases, skips `capture`, and runs `recall`, the reader, and the judge. It refuses to run if a database with the expected name does not exist. From 7394376dfdf798e69bc94fa7d63fbdfbff4bec7c Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 15:32:18 +0530 Subject: [PATCH 2/9] Phase 9 task 2: reuse the kept memory With --reuse, the memory arm opens the kept database of each question, skips capture, and runs recall, the reader, and the judge. The runner refuses when a kept database of the run does not exist, and never deletes a kept database in a reused run. local_store takes `delete` (always, on_error, never) in place of `keep`. A reused row has None for facts written and sessions at the limit, and each memory row now records the SHA-256 of its block. Also fixes the name of the database of bench.diagnose, which the task 1 commit passed without its prefix. --- bench/diagnose.py | 2 +- bench/e2e.py | 104 +++++++++++++++++++++++++++++----------- bench/tests/test_e2e.py | 58 +++++++++++++++++----- 3 files changed, 122 insertions(+), 42 deletions(-) diff --git a/bench/diagnose.py b/bench/diagnose.py index 9dd0f9c..99a38d1 100644 --- a/bench/diagnose.py +++ b/bench/diagnose.py @@ -189,7 +189,7 @@ def main(argv: Sequence[str] | None = None) -> Path: marker = CachedModels(Remote(), CACHE_DIR / "replies.sqlite") rows: list[dict[str, Any]] = [] run_id = f"diagnose-{datetime.now(UTC):%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" - with local_store(args.url, run_id) as (stores, empty): + with local_store(args.url, f"hotmemory-bench-{run_id}") as (stores, empty): memory = Memory(stores[config.memory_ranking]) for count, question_id in enumerate(missed, start=1): question = questions[question_id] diff --git a/bench/e2e.py b/bench/e2e.py index eb49d51..a199afc 100644 --- a/bench/e2e.py +++ b/bench/e2e.py @@ -27,7 +27,7 @@ from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path -from typing import Any, cast +from typing import Any, Literal, cast from bench import embed, prompts from bench.data import ( @@ -227,9 +227,10 @@ class Arm: context_chars: int answer: Reply verdict: Reply - extracted: int = 0 - at_limit: int = 0 + extracted: int | None = 0 + at_limit: int | None = 0 episodes: tuple[str, ...] = () + block_sha256: str | None = None @property def correct(self) -> bool: @@ -241,14 +242,15 @@ def answer( question: Question, prompt: str, context_chars: int, - extracted: int = 0, - at_limit: int = 0, + extracted: int | None = 0, + at_limit: int | None = 0, episodes: tuple[str, ...] = (), + block_sha256: str | None = None, ) -> Arm: """Read `prompt` with READER, judge the answer with JUDGE, and return the arm.""" reply = models(READER, prompt, READER_MAX_TOKENS) verdict = models(JUDGE, prompts.judge(question, reply.text), JUDGE_MAX_TOKENS) - return Arm(context_chars, reply, verdict, extracted, at_limit, episodes) + return Arm(context_chars, reply, verdict, extracted, at_limit, episodes, block_sha256) def run_question( @@ -259,10 +261,13 @@ def run_question( run_id: str, arms: Sequence[str] = ARMS, config: Config = BASELINE, + capture: bool = True, ) -> dict[str, Arm]: """Run each arm on `question`, and empty its namespaces after. - `stores` maps each ranking to a store, and every store holds the same records. + `stores` maps each ranking to a store, and every store holds the same records. Without + `capture`, the memory arm recalls from the records that its namespace holds already, + and its count of facts written and of sessions at the limit is None. """ result: dict[str, Arm] = {} if "full" in arms: @@ -281,10 +286,14 @@ def run_question( if "memory" in arms: namespace = ("bench", "lme_s", run_id, "memory", question.id) memory = Memory(stores[config.memory_ranking]) - written = 0 + written: int | None = None at_limit: list[str] = [] - for session in question.sessions: - written += len(capture_session(memory, namespace, session, models, config, at_limit)) + if capture: + written = 0 + for session in question.sessions: + written += len( + capture_session(memory, namespace, session, models, config, at_limit) + ) records, block = memory.recall( question.question, [namespace], @@ -297,7 +306,14 @@ def run_question( shown = tuple(record.key for record in records if record.kind == "episode") prompt = prompts.read_facts(question, block) result["memory"] = answer( - models, question, prompt, len(block), written, len(at_limit), shown + models, + question, + prompt, + len(block), + written, + len(at_limit) if capture else None, + shown, + hashlib.sha256(block.encode()).hexdigest(), ) return result @@ -381,15 +397,19 @@ def accuracy(found: Sequence[bool]) -> dict[str, Any]: } +Delete = Literal["always", "on_error", "never"] + + @contextmanager def local_store( - url: str, name: str, embedder: Embedder | None = None, keep: bool = False + url: str, name: str, embedder: Embedder | None = None, delete: Delete = "always" ) -> Iterator[tuple[dict[str, Store], Callable[[Sequence[str]], None]]]: """Yield a HotdataStore for each ranking, and a call that empties a namespace. The stores share the local database named `name`, which `provision` opens or creates. - The database is deleted after, unless `keep` is true and the block exits without an - error. Without `embedder`, the stores embed with `embed.openai_embedder()`. + `delete` says when the database is deleted after the block: `always`, `on_error` when + the block raises, or `never`. Without `embedder`, the stores embed with + `embed.openai_embedder()`. """ import pyarrow as pa from hotdata_framework import HotdataClient @@ -426,12 +446,12 @@ def empty(namespace: Sequence[str]) -> None: ) for ranking in RANKINGS } - kept = False + failed = True try: yield stores, empty - kept = keep + failed = False finally: - if not kept: + if delete == "always" or (delete == "on_error" and failed): client.delete_managed_database(store.database) client.close() @@ -505,6 +525,7 @@ class Job: replay: bool total: int kept: str | None = None + reuse: bool = False @dataclass(frozen=True) @@ -550,6 +571,7 @@ def row_of(question: Question, result: dict[str, Arm]) -> dict[str, Any]: "facts_written": found.extracted, "sessions_at_limit": found.at_limit, "episodes_in_block": list(found.episodes), + "block_sha256": found.block_sha256, "reader_input_tokens": found.answer.input_tokens, "answer": found.answer.text, "verdict": found.verdict.text, @@ -564,8 +586,10 @@ def run_shard(job: Job, shard: Shard) -> Done: With `job.kept`, each question runs over a new database named `-`, and the database keeps the memory of the question after the run. A question that - raises deletes its database. The worker opens its own model cache, embedder, and - store, so it can run in its own process. It prints one line for each question. + raises deletes its database. With `job.reuse` as well, each question opens its kept + database, skips `capture`, and never deletes the database. The worker opens its own + model cache, embedder, and store, so it can run in its own process. It prints one line + for each question. """ from bench.models import Offline, Remote @@ -588,9 +612,17 @@ def finish(question: Question, result: dict[str, Arm]) -> None: if job.kept is not None: for question in shard.questions: name = f"{job.kept}-{question.id}" - with local_store(job.url, name, embedder, keep=True) as (stores, _): + delete: Delete = "never" if job.reuse else "on_error" + with local_store(job.url, name, embedder, delete) as (stores, _): result = run_question( - question, models, stores, _keep, KEPT_RUN, job.arms, job.config + question, + models, + stores, + _keep, + KEPT_RUN, + job.arms, + job.config, + capture=not job.reuse, ) finish(question, result) else: @@ -655,6 +687,11 @@ def main(argv: Sequence[str] | None = None) -> Path | None: action="store_true", help="keep the memory of each question in a local database of its own", ) + parser.add_argument( + "--reuse", + action="store_true", + help="recall from the memory that a run with --keep kept, and skip capture", + ) parser.add_argument("--tag", default="", help="a suffix for the name of the result file") parser.add_argument("--url", default="http://localhost:3000") args = parser.parse_args(argv) @@ -663,8 +700,10 @@ def main(argv: Sequence[str] | None = None) -> Path | None: parser.error(f"arms must be among {', '.join(ARMS)}") if args.workers < 1: parser.error("workers must be at least 1") - if args.keep and arms != ("memory",): - parser.error("--keep needs --arms memory") + if args.keep and args.reuse: + parser.error("pass --keep or --reuse, not both") + if (args.keep or args.reuse) and arms != ("memory",): + parser.error(f"--{'keep' if args.keep else 'reuse'} needs --arms memory") try: config = Config( retrieval_ranking=args.retrieval_ranking, @@ -700,22 +739,26 @@ def main(argv: Sequence[str] | None = None) -> Path | None: return None kept = None - if args.keep: + if args.keep or args.reuse: try: kept = kept_name(args.sample, config, library_commit()) except RuntimeError as error: parser.error(str(error)) - found = existing_databases(args.url, [f"{kept}-{question.id}" for question in questions]) - if found: + names = [f"{kept}-{question.id}" for question in questions] + found = existing_databases(args.url, names) + if args.keep and found: parser.error(f"{len(found)} kept databases exist already, for example {found[0]}") - print(f"keeping the memory as {kept}-") + missing = sorted(set(names) - set(found)) + if args.reuse and missing: + parser.error(f"{len(missing)} kept databases do not exist, for example {missing[0]}") + print(f"{'keeping' if args.keep else 'reusing'} the memory as {kept}-") commit = _commit() started = datetime.now(UTC) run_id = f"{started:%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" clock_start = time.perf_counter() shards = shards_of(questions, args.workers, run_id) - job = Job(args.url, arms, config, args.replay, len(questions), kept) + job = Job(args.url, arms, config, args.replay, len(questions), kept, args.reuse) if len(shards) == 1: done = [run_shard(job, shards[0])] else: @@ -772,6 +815,7 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "session_episodes": config.session_episodes, "episodes": config.episodes, "kept": kept, + "reused": args.reuse, }, "arms": { "full": "every session of the question", @@ -792,7 +836,9 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "workers": len(shards), "spent_usd": round(spent, 2), "model_calls": calls, - "sessions_at_limit": sum( + "sessions_at_limit": None + if args.reuse + else sum( row["arms"]["memory"]["sessions_at_limit"] for row in rows if "memory" in row["arms"] diff --git a/bench/tests/test_e2e.py b/bench/tests/test_e2e.py index a7fa80a..1dbcdac 100644 --- a/bench/tests/test_e2e.py +++ b/bench/tests/test_e2e.py @@ -414,10 +414,43 @@ def git(*args: str) -> str: library_commit(tmp_path) -def test_keep_needs_the_memory_arm_only(capsys: pytest.CaptureFixture[str]) -> None: +@pytest.mark.parametrize( + ("argv", "error"), + [ + (["--keep", "--arms", "retrieval,memory"], "--keep needs --arms memory"), + (["--reuse", "--arms", "full"], "--reuse needs --arms memory"), + (["--keep", "--reuse", "--arms", "memory"], "pass --keep or --reuse, not both"), + ], +) +def test_keep_and_reuse_refuse_other_arms( + argv: list[str], error: str, capsys: pytest.CaptureFixture[str] +) -> None: with pytest.raises(SystemExit): - main(["--keep", "--arms", "retrieval,memory"]) - assert "--keep needs --arms memory" in capsys.readouterr().err + main(argv) + assert error in capsys.readouterr().err + + +def test_run_question_without_capture_recalls_the_kept_memory( + questions: dict[str, Question], +) -> None: + question = questions["q_one"] + store = MemoryStore(embedder=fake_embedder, clock=FakeClock()) + stores = {"fused": store, "vector": store} + config = Config(budget=100_000, session_episodes=True, episodes="search") + captured = FakeModels() + first = run_question(question, captured, stores, lambda _: None, "kept", ("memory",), config)[ + "memory" + ] + reused = FakeModels() + second = run_question( + question, reused, stores, lambda _: None, "kept", ("memory",), config, capture=False + )["memory"] + assert [model for model, _ in reused.prompts] == [READER, JUDGE] + assert reused.prompts[0] == next(p for p in captured.prompts if p[0] == READER) + assert second.block_sha256 == first.block_sha256 is not None + assert (second.context_chars, second.episodes) == (first.context_chars, first.episodes) + assert first.extracted and first.at_limit == 0 + assert second.extracted is None and second.at_limit is None def _embed(texts: Sequence[str]) -> list[list[float]]: @@ -425,26 +458,27 @@ def _embed(texts: Sequence[str]) -> list[list[float]]: @pytest.mark.hotdata -def test_local_store_keeps_the_database_only_on_success() -> None: +def test_local_store_deletes_the_database_as_asked() -> None: url = os.environ.get("HOTMEMORY_TEST_URL") if not url: pytest.skip("HOTMEMORY_TEST_URL is not set") - kept = f"hotmemory-test-kept-{uuid.uuid4().hex[:8]}" - failed = f"{kept}-failed" - temporary = f"{kept}-temporary" - with local_store(url, kept, _embed, keep=True): + base = f"hotmemory-test-kept-{uuid.uuid4().hex[:8]}" + names = {case: f"{base}-{case}" for case in ("always", "error", "kept", "never")} + with local_store(url, names["always"], _embed): pass - with pytest.raises(ValueError), local_store(url, failed, _embed, keep=True): + with pytest.raises(ValueError), local_store(url, names["error"], _embed, "on_error"): raise ValueError("stop") - with local_store(url, temporary, _embed): + with local_store(url, names["kept"], _embed, "on_error"): pass + with pytest.raises(ValueError), local_store(url, names["never"], _embed, "never"): + raise ValueError("stop") try: - assert existing_databases(url, [kept, failed, temporary]) == [kept] + assert existing_databases(url, names.values()) == [names["kept"], names["never"]] finally: from hotdata_framework import HotdataClient client = HotdataClient("local", "local", host=url) for database in client.list_managed_databases(): - if database.description == kept: + if database.description in (names["kept"], names["never"]): client.delete_managed_database(database) client.close() From 49564da0468c079e6a141a434c33009e073c9960 Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 15:40:38 +0530 Subject: [PATCH 3/9] Phase 9 task 3: diagnose reads the kept memory When the result file names kept databases (a run with --keep or --reuse), bench.diagnose opens the kept database of each missed question and reads its memory, with no capture replay. It refuses when a kept database does not exist, and never deletes one. A result file without kept databases replays capture as before. The diagnosis of one miss moves into `diagnose`, and the output records the kept name. --- bench/diagnose.py | 128 ++++++++++++++++++++++------------- bench/tests/test_diagnose.py | 49 ++++++++++++-- 2 files changed, 127 insertions(+), 50 deletions(-) diff --git a/bench/diagnose.py b/bench/diagnose.py index 99a38d1..45b1da1 100644 --- a/bench/diagnose.py +++ b/bench/diagnose.py @@ -9,6 +9,9 @@ some of them ("some present"). If the run recalled episodes, each miss also records whether the block held an evidence session as an episode, from the episode keys of the result file. +If the result file names kept databases (a run with `--keep` or `--reuse`), this reads the +memory of each question from its kept database and does not run `capture`. + uv run --group bench --env-file .env python -m bench.diagnose --result """ @@ -26,11 +29,13 @@ from bench.data import DATA_DIR, DATA_FILE, QUESTION_TYPES, Question, load from bench.e2e import ( BASELINE, + KEPT_RUN, RECALL_BUDGET, RECALL_K, Config, capture_session, episode_key, + existing_databases, fact_source, local_store, ) @@ -187,53 +192,37 @@ def main(argv: Sequence[str] | None = None) -> Path: questions = {question.id: question for question in load(DATA_DIR / DATA_FILE)} replay = CachedModels(Offline(), CACHE_DIR / "replies.sqlite") marker = CachedModels(Remote(), CACHE_DIR / "replies.sqlite") + kept: str | None = result["config"].get("memory", {}).get("kept") rows: list[dict[str, Any]] = [] - run_id = f"diagnose-{datetime.now(UTC):%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" - with local_store(args.url, f"hotmemory-bench-{run_id}") as (stores, empty): - memory = Memory(stores[config.memory_ranking]) - for count, question_id in enumerate(missed, start=1): - question = questions[question_id] - namespace = ("bench", "lme_s", run_id, question.id) - sessions = {fact_source(session.id): session.id for session in question.sessions} - sessions |= {episode_key(session.id): session.id for session in question.sessions} - for session in question.sessions: - capture_session(memory, namespace, session, replay, config, []) - store = stores[config.memory_ranking] - facts = store.list(namespace, Filter(kind=RECALLED_KINDS), limit=10_000) - marked = _mark(marker, question.question, question.answer, facts) - holding = {facts[n - 1].id for n in marked} - ranks: dict[str, int | None] = {} - for name in RANKINGS: - ranked = _search(stores[name], question.question, namespace, len(facts)) - found = [n for n, r in enumerate(ranked, start=1) if r.id in holding] - ranks[name] = found[0] if found else None - top = _search(store, question.question, namespace, config.k) - block = block_ids(top, question.as_of, config.k, config.budget) - cause = verdict(holding, ranks[config.memory_ranking], block, config.k) - present = None - if cause == RECALLED: - shown = [fact.content for fact in facts if fact.id in holding & block] - whole = _complete(marker, question.question, question.answer, shown) - present = "all" if whole else "some" - rows.append( - { - "id": question.id, - "type": question.type, - "facts": len(facts), - "holding": [facts[n - 1].content for n in marked], - "holding_sessions": [session_of_fact(facts[n - 1], sessions) for n in marked], - "best_rank": ranks, - "cause": cause, - "present": present, - "evidence_in_block": ( - None - if config.episodes == "none" - else evidence_in_block(question, in_block[question.id]) - ), - } - ) - empty(namespace) - print(f"{count}/{len(missed)} {question.id} {group(rows[-1])} {ranks}", flush=True) + + def diagnosed(question: Question, stores: dict[str, Store], namespace: tuple[str, ...]) -> None: + rows.append(diagnose(question, stores, namespace, config, marker, in_block[question.id])) + print( + f"{len(rows)}/{len(missed)} {question.id} {group(rows[-1])} {rows[-1]['best_rank']}", + flush=True, + ) + + if kept is not None: + names = [f"{kept}-{question_id}" for question_id in missed] + absent = sorted(set(names) - set(existing_databases(args.url, names))) + if absent: + parser.error(f"{len(absent)} kept databases do not exist, for example {absent[0]}") + for question_id in missed: + name = f"{kept}-{question_id}" + with local_store(args.url, name, delete="never") as (stores, _): + namespace: tuple[str, ...] = ("bench", "lme_s", KEPT_RUN, "memory", question_id) + diagnosed(questions[question_id], stores, namespace) + else: + run_id = f"diagnose-{datetime.now(UTC):%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" + with local_store(args.url, f"hotmemory-bench-{run_id}") as (stores, empty): + memory = Memory(stores[config.memory_ranking]) + for question_id in missed: + question = questions[question_id] + namespace = ("bench", "lme_s", run_id, question.id) + for session in question.sessions: + capture_session(memory, namespace, session, replay, config, []) + diagnosed(question, stores, namespace) + empty(namespace) replay.close() marker.close() causes = Counter(row["cause"] for row in rows) @@ -249,6 +238,7 @@ def main(argv: Sequence[str] | None = None) -> Path: "session_episodes": config.session_episodes, "episodes": config.episodes, }, + "kept": kept, "marker": JUDGE, "hotmemory_commit": commit, "spent_usd": round(marker.spent, 4), @@ -265,6 +255,52 @@ def main(argv: Sequence[str] | None = None) -> Path: return path +def diagnose( + question: Question, + stores: dict[str, Store], + namespace: tuple[str, ...], + config: Config, + marker: CachedModels, + shown: Sequence[str], +) -> dict[str, Any]: + """Return the diagnosis of one miss, from the memory that `namespace` holds. + + `shown` holds the episode keys that the block of the run held. + """ + sessions = {fact_source(session.id): session.id for session in question.sessions} + sessions |= {episode_key(session.id): session.id for session in question.sessions} + store = stores[config.memory_ranking] + facts = store.list(namespace, Filter(kind=RECALLED_KINDS), limit=10_000) + marked = _mark(marker, question.question, question.answer, facts) + holding = {facts[n - 1].id for n in marked} + ranks: dict[str, int | None] = {} + for name in RANKINGS: + ranked = _search(stores[name], question.question, namespace, len(facts)) + found = [n for n, r in enumerate(ranked, start=1) if r.id in holding] + ranks[name] = found[0] if found else None + top = _search(store, question.question, namespace, config.k) + block = block_ids(top, question.as_of, config.k, config.budget) + cause = verdict(holding, ranks[config.memory_ranking], block, config.k) + present = None + if cause == RECALLED: + listed = [fact.content for fact in facts if fact.id in holding & block] + whole = _complete(marker, question.question, question.answer, listed) + present = "all" if whole else "some" + return { + "id": question.id, + "type": question.type, + "facts": len(facts), + "holding": [facts[n - 1].content for n in marked], + "holding_sessions": [session_of_fact(facts[n - 1], sessions) for n in marked], + "best_rank": ranks, + "cause": cause, + "present": present, + "evidence_in_block": ( + None if config.episodes == "none" else evidence_in_block(question, shown) + ), + } + + def session_of_fact(fact: Record, sessions: dict[str, str]) -> str | None: """Return the session id of `fact` from its sources, or None if no source names one. diff --git a/bench/tests/test_diagnose.py b/bench/tests/test_diagnose.py index d2fc473..3efdcbf 100644 --- a/bench/tests/test_diagnose.py +++ b/bench/tests/test_diagnose.py @@ -5,9 +5,11 @@ import pytest -from bench.data import Question +from bench.data import Question, load from bench.diagnose import ( + RECALLED, block_ids, + diagnose, evidence_in_block, group, memory_config, @@ -15,9 +17,11 @@ table, verdict, ) -from bench.e2e import BASELINE, episode_key, fact_source -from bench.models import READER, CachedModels, CacheMiss, Offline, Reply -from hotmemory import Record +from bench.e2e import BASELINE, KEPT_RUN, Config, episode_key, fact_source, run_question +from bench.models import JUDGE, READER, CachedModels, CacheMiss, Offline, Reply +from bench.tests.test_e2e import FIXTURE, FakeModels +from hotmemory import MemoryStore, Record +from tests.conftest import FakeClock, fake_embedder AS_OF = datetime(2023, 5, 30, 23, 59, 59, tzinfo=UTC) @@ -141,3 +145,40 @@ def test_offline_replays_the_cache_and_stops_at_a_miss(tmp_path: Path) -> None: with pytest.raises(CacheMiss): replay(READER, "q", 10) assert (replay.calls, replay.spent) == (0, 0.0) + + +class FakeMarker: + """Marks the facts that hold `word`, and says that every needed value is present.""" + + def __init__(self, word: str) -> None: + self.word = word + self.calls = 0 + + def __call__( + self, model: str, prompt: str, max_tokens: int, schema: dict[str, Any] | None = None + ) -> Reply: + assert model == JUDGE + self.calls += 1 + if "Facts:\n1. " not in prompt: + return Reply('{"complete": true}', 10, 3) + lines = prompt.split("Facts:\n", 1)[1].splitlines() + marked = [n for n, line in enumerate(lines, 1) if self.word in line] + return Reply(f'{{"facts": {marked}}}', 10, 3) + + +def test_diagnose_reads_the_memory_of_the_namespace() -> None: + question = next(q for q in load(FIXTURE, expected_sha256=None) if q.id == "q_one") + store = MemoryStore(embedder=fake_embedder, clock=FakeClock()) + stores = {"fused": store, "vector": store} + config = Config(budget=100_000, session_episodes=True, episodes="search") + result = run_question( + question, FakeModels(), stores, lambda _: None, KEPT_RUN, ("memory",), config + ) + namespace = ("bench", "lme_s", KEPT_RUN, "memory", question.id) + marker = FakeMarker("bike") + shown = result["memory"].episodes + row = diagnose(question, stores, namespace, config, marker, shown) # type: ignore[arg-type] + assert len(row["holding"]) == 3 and row["facts"] > 3 + assert set(row["holding_sessions"]) == {"answer_a_1", "answer_a_2", "answer_a_late"} + assert row["cause"] == RECALLED and row["present"] == "all" + assert row["evidence_in_block"] is True From 102dba2cd2a75e34dfecdfc16fedde8c81d0b030 Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 15:42:21 +0530 Subject: [PATCH 4/9] Phase 9 task 4: a smoke run of 18 questions With --smoke, the runner runs the 18 questions of bench/sample-smoke.json, 3 of each type drawn from sample A with seed 20261016, over one MemoryStore with a fixed clock, in the process. The result file records the smoke sample, the memory driver, and the vector ranking, and is named e2e-smoke-.json. --smoke refuses --keep, --reuse, --workers, and --sample. MemoryStore ranks by vector only, so capture shows the extractor a different list of known facts than a run on the local engine, and the first smoke run of a configuration pays for its extraction (observed: a replay stops at the first extractor call). --- bench/e2e.py | 106 +++++++++++++++++++++++++++++++--------- bench/sample-smoke.json | 20 ++++++++ bench/tests/test_e2e.py | 30 ++++++++++++ 3 files changed, 134 insertions(+), 22 deletions(-) create mode 100644 bench/sample-smoke.json diff --git a/bench/e2e.py b/bench/e2e.py index a199afc..279a006 100644 --- a/bench/e2e.py +++ b/bench/e2e.py @@ -7,8 +7,10 @@ uv run --group bench --env-file .env python -m bench.e2e uv run --group bench --env-file .env python -m bench.e2e --run uv run --group bench --env-file .env python -m bench.e2e --replay --arms memory + uv run --group bench --env-file .env python -m bench.e2e --run --smoke --arms memory -It writes one JSON file into `bench/results/` with the configuration and the results. +With `--smoke`, it runs the 18 questions of `bench/sample-smoke.json` on `MemoryStore`, in +the process. It writes one JSON file into `bench/results/` with the configuration and the results. """ from __future__ import annotations @@ -24,7 +26,7 @@ from collections.abc import Callable, Collection, Iterator, Mapping, Sequence from concurrent.futures import ProcessPoolExecutor from contextlib import contextmanager -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal, cast @@ -52,8 +54,8 @@ CachedModels, Reply, ) -from bench.retrieval import RANKINGS, RESULTS_DIR, _commit -from hotmemory import Embedder, Fact, Filter, Memory, Record, Store +from bench.retrieval import FIXED_TIME, RANKINGS, RESULTS_DIR, _commit +from hotmemory import Embedder, Fact, Filter, Memory, MemoryStore, Record, Store from hotmemory.contract import Episodes HERE = Path(__file__).resolve().parent @@ -128,6 +130,8 @@ class Sample: "a": Sample("bench/sample.json", 20261010), "b": Sample("bench/sample-b.json", 20261013, excludes=("a",)), } +SMOKE = Sample("bench/sample-smoke.json", 20261016) +SMOKE_PER_TYPE = 3 def choose_sample( @@ -164,6 +168,26 @@ def sample_ids(name: str, questions: Sequence[Question], root: Path = HERE.paren return ids +def smoke_ids(questions: Sequence[Question], root: Path = HERE.parent) -> list[str]: + """Return the ids of the smoke sample, and write its file first if it does not exist. + + The smoke sample draws SMOKE_PER_TYPE questions of each type from sample A with the + seed of SMOKE, sorted by type and id. + """ + path = root / SMOKE.file + if not path.exists(): + types = {question.id: question.type for question in questions} + sample = sample_ids("a", questions, root) + rng = random.Random(SMOKE.seed) + chosen = [] + for kind in QUESTION_TYPES: + of_kind = sorted(i for i in sample if types[i] == kind) + chosen.extend(sorted(rng.sample(of_kind, SMOKE_PER_TYPE))) + path.write_text(json.dumps(chosen, indent=1) + "\n") + ids: list[str] = json.loads(path.read_text()) + return ids + + def fact_source(session_id: str) -> str: """Return the source of the facts of a session: `session:` and 12 hex digits of its SHA-256. @@ -526,6 +550,7 @@ class Job: total: int kept: str | None = None reuse: bool = False + smoke: bool = False @dataclass(frozen=True) @@ -587,9 +612,10 @@ def run_shard(job: Job, shard: Shard) -> Done: With `job.kept`, each question runs over a new database named `-`, and the database keeps the memory of the question after the run. A question that raises deletes its database. With `job.reuse` as well, each question opens its kept - database, skips `capture`, and never deletes the database. The worker opens its own - model cache, embedder, and store, so it can run in its own process. It prints one line - for each question. + database, skips `capture`, and never deletes the database. With `job.smoke`, the + questions run over one `MemoryStore` with a fixed clock, in the process. The worker + opens its own model cache, embedder, and store, so it can run in its own process. It + prints one line for each question. """ from bench.models import Offline, Remote @@ -609,7 +635,20 @@ def finish(question: Question, result: dict[str, Arm]) -> None: ) try: - if job.kept is not None: + if job.smoke: + store = MemoryStore(embedder=embedder, clock=lambda: FIXED_TIME) + + def empty(namespace: Sequence[str]) -> None: + for record in store.list(namespace, limit=100_000): + store.delete(record.namespace, record.key) + + for question in shard.questions: + in_process: dict[str, Store] = dict.fromkeys(RANKINGS, store) + result = run_question( + question, models, in_process, empty, shard.run_id, job.arms, job.config + ) + finish(question, result) + elif job.kept is not None: for question in shard.questions: name = f"{job.kept}-{question.id}" delete: Delete = "never" if job.reuse else "on_error" @@ -627,10 +666,10 @@ def finish(question: Question, result: dict[str, Arm]) -> None: finish(question, result) else: name = f"hotmemory-bench-{shard.run_id}" - with local_store(job.url, name, embedder) as (stores, empty): + with local_store(job.url, name, embedder) as (stores, emptied): for question in shard.questions: result = run_question( - question, models, stores, empty, shard.run_id, job.arms, job.config + question, models, stores, emptied, shard.run_id, job.arms, job.config ) finish(question, result) finally: @@ -653,6 +692,11 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "--limit", type=int, default=None, help="the first N questions of the sample" ) parser.add_argument("--sample", choices=tuple(SAMPLES), default="a", help="the sample to run") + parser.add_argument( + "--smoke", + action="store_true", + help="run the 18 questions of the smoke sample on MemoryStore, in the process", + ) parser.add_argument("--retrieval-ranking", choices=RANKINGS, default="fused") parser.add_argument("--memory-ranking", choices=RANKINGS, default="fused") parser.add_argument("--recall-k", type=int, default=RECALL_K, help="k of the memory recall") @@ -702,6 +746,8 @@ def main(argv: Sequence[str] | None = None) -> Path | None: parser.error("workers must be at least 1") if args.keep and args.reuse: parser.error("pass --keep or --reuse, not both") + if args.smoke and (args.keep or args.reuse or args.workers != 1 or args.sample != "a"): + parser.error("--smoke runs alone: no --keep, --reuse, --workers, or --sample") if (args.keep or args.reuse) and arms != ("memory",): parser.error(f"--{'keep' if args.keep else 'reuse'} needs --arms memory") try: @@ -717,9 +763,14 @@ def main(argv: Sequence[str] | None = None) -> Path | None: ) except ValueError as error: parser.error(str(error)) + if args.smoke: + config = replace(config, retrieval_ranking="vector", memory_ranking="vector") all_questions = {question.id: question for question in load(DATA_DIR / DATA_FILE)} - sample = sample_ids(args.sample, list(all_questions.values())) + if args.smoke: + sample = smoke_ids(list(all_questions.values())) + else: + sample = sample_ids(args.sample, list(all_questions.values())) questions = [all_questions[question_id] for question_id in sample][: args.limit] tokens = embed.TiktokenTokens(encoding="o200k_base") @@ -758,7 +809,7 @@ def main(argv: Sequence[str] | None = None) -> Path | None: run_id = f"{started:%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" clock_start = time.perf_counter() shards = shards_of(questions, args.workers, run_id) - job = Job(args.url, arms, config, args.replay, len(questions), kept, args.reuse) + job = Job(args.url, arms, config, args.replay, len(questions), kept, args.reuse, args.smoke) if len(shards) == 1: done = [run_shard(job, shards[0])] else: @@ -778,13 +829,23 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "data_file": DATA_FILE, "data_revision": DATA_REVISION, "data_sha256": DATA_SHA256, - "sample": { - "name": args.sample, - "file": SAMPLES[args.sample].file, - "per_type": PER_TYPE, - "seed": SAMPLES[args.sample].seed, - "excludes": list(SAMPLES[args.sample].excludes), - }, + "sample": ( + { + "name": "smoke", + "file": SMOKE.file, + "per_type": SMOKE_PER_TYPE, + "seed": SMOKE.seed, + "from": SAMPLES["a"].file, + } + if args.smoke + else { + "name": args.sample, + "file": SAMPLES[args.sample].file, + "per_type": PER_TYPE, + "seed": SAMPLES[args.sample].seed, + "excludes": list(SAMPLES[args.sample].excludes), + } + ), "reader": { "model": READER, "temperature": 0, @@ -803,8 +864,8 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "max_tokens": EXTRACTOR_MAX_TOKENS, }, "embedder": {"model": embed.MODEL, "dimensions": embed.DIMENSIONS}, - "driver": "hotdata", - "location": "local", + "driver": "memory" if args.smoke else "hotdata", + "location": "in-process" if args.smoke else "local", "retrieval": {"ranking": config.retrieval_ranking, "sessions": RETRIEVED}, "memory": { "ranking": config.memory_ranking, @@ -853,7 +914,8 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "questions": rows, } RESULTS_DIR.mkdir(exist_ok=True) - name = f"e2e-sample-{args.sample}" + (f"-{args.tag}" if args.tag else "") + name = "e2e-smoke" if args.smoke else f"e2e-sample-{args.sample}" + name += f"-{args.tag}" if args.tag else "" name += f"-first{args.limit}" if args.limit is not None else "" path = RESULTS_DIR / f"{name}.json" path.write_text(json.dumps(result_file, indent=1) + "\n") diff --git a/bench/sample-smoke.json b/bench/sample-smoke.json new file mode 100644 index 0000000..caaa525 --- /dev/null +++ b/bench/sample-smoke.json @@ -0,0 +1,20 @@ +[ + "51a45a95", + "6f9b354f", + "8e9d538c", + "3249768e", + "7a8d0b71", + "c4f10528", + "195a1a1b", + "1da05512", + "32260d93", + "21d02d0d", + "28dc39ac", + "92a0aa75", + "07741c45", + "b6019101", + "e66b632c", + "gpt4_65aabe59", + "gpt4_74aed68e", + "gpt4_a1b77f9c" +] diff --git a/bench/tests/test_e2e.py b/bench/tests/test_e2e.py index 1dbcdac..970eaf1 100644 --- a/bench/tests/test_e2e.py +++ b/bench/tests/test_e2e.py @@ -13,6 +13,7 @@ from bench.data import QUESTION_TYPES, Question, load from bench.e2e import ( SAMPLES, + SMOKE, Config, capture_session, choose_sample, @@ -28,6 +29,7 @@ run_question, sample_ids, shards_of, + smoke_ids, summarize, ) from bench.models import EXTRACTOR, JUDGE, READER, CachedModels, Offline, Reply @@ -112,6 +114,32 @@ def test_sample_b_leaves_out_sample_a(questions: dict[str, Question], tmp_path: assert sample_ids("b", pool[:60], root=tmp_path) == b +def test_smoke_ids_draw_three_of_each_type_from_sample_a( + questions: dict[str, Question], tmp_path: Path +) -> None: + base = questions["q_one"] + pool = [ + replace(base, id=f"{kind}-{n}", type=kind) for kind in QUESTION_TYPES for n in range(20) + ] + (tmp_path / "bench").mkdir() + a = sample_ids("a", pool, root=tmp_path) + smoke = smoke_ids(pool, root=tmp_path) + types = {question.id: question.type for question in pool} + assert len(smoke) == len(set(smoke)) == 18 + assert set(smoke) <= set(a) + assert [types[i] for i in smoke] == [kind for kind in QUESTION_TYPES for _ in range(3)] + assert json.loads((tmp_path / SMOKE.file).read_text()) == smoke + assert smoke_ids(pool[:30], root=tmp_path) == smoke + + +def test_committed_smoke_sample_is_in_sample_a() -> None: + root = Path(__file__).resolve().parents[2] + smoke = json.loads((root / SMOKE.file).read_text()) + a = json.loads((root / SAMPLES["a"].file).read_text()) + assert len(smoke) == len(set(smoke)) == 18 + assert set(smoke) <= set(a) + + def test_run_question_on_every_arm(questions: dict[str, Question]) -> None: question = questions["q_one"] store = MemoryStore(embedder=fake_embedder, clock=FakeClock()) @@ -420,6 +448,8 @@ def git(*args: str) -> str: (["--keep", "--arms", "retrieval,memory"], "--keep needs --arms memory"), (["--reuse", "--arms", "full"], "--reuse needs --arms memory"), (["--keep", "--reuse", "--arms", "memory"], "pass --keep or --reuse, not both"), + (["--smoke", "--workers", "4"], "--smoke runs alone"), + (["--smoke", "--sample", "b"], "--smoke runs alone"), ], ) def test_keep_and_reuse_refuse_other_arms( From 630e3be77f1a2d04ae972aacdf0a3ac1cb9f6a5a Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 15:47:07 +0530 Subject: [PATCH 5/9] Phase 9 task 5: keep finished questions, and --resume The runner writes bench/results/.rows.jsonl as it runs: a header line with the arms, the configuration, the sample, the limit, and the kept name, then one line for each finished question, appended under a file lock. With --resume, it reads the file, refuses a header of another run, and runs only the questions that the file does not hold. Without --resume, it refuses to start over an existing rows file. The result file is written, and the rows file deleted, only when every question finished. A question that raises is listed as failed, its namespaces are emptied, and the next question runs. The error text crosses the process pool as a string, so an APIConnectionError no longer breaks the pool. CacheMiss still stops the shard. A resumed --keep run deletes the kept databases of the questions that did not finish before it starts again. --- bench/e2e.py | 228 +++++++++++++++++++++++++++++++++------- bench/tests/test_e2e.py | 65 ++++++++++++ 2 files changed, 254 insertions(+), 39 deletions(-) diff --git a/bench/e2e.py b/bench/e2e.py index 279a006..3c55ca4 100644 --- a/bench/e2e.py +++ b/bench/e2e.py @@ -16,6 +16,7 @@ from __future__ import annotations import argparse +import fcntl import hashlib import json import multiprocessing @@ -26,7 +27,7 @@ from collections.abc import Callable, Collection, Iterator, Mapping, Sequence from concurrent.futures import ProcessPoolExecutor from contextlib import contextmanager -from dataclasses import dataclass, replace +from dataclasses import asdict, dataclass, replace from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal, cast @@ -52,6 +53,7 @@ READER, READER_MAX_TOKENS, CachedModels, + CacheMiss, Reply, ) from bench.retrieval import FIXED_TIME, RANKINGS, RESULTS_DIR, _commit @@ -551,6 +553,7 @@ class Job: kept: str | None = None reuse: bool = False smoke: bool = False + rows_file: Path | None = None @dataclass(frozen=True) @@ -563,12 +566,41 @@ class Shard: @dataclass(frozen=True) class Done: - """The rows of one shard, and what its model calls cost.""" + """The rows of one shard, the questions that failed, and what its model calls cost.""" rows: list[dict[str, Any]] spent: float calls: int embedder_misses: int + failed: list[str] + + +def append_row(path: Path, row: dict[str, Any]) -> None: + """Append `row` to the file at `path` as one JSON line, under a lock on the file.""" + with path.open("a") as file: + fcntl.flock(file, fcntl.LOCK_EX) + try: + file.write(json.dumps(row) + "\n") + file.flush() + finally: + fcntl.flock(file, fcntl.LOCK_UN) + + +def read_rows(path: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]: + """Return the header and the rows of a rows file that `main` wrote. + + The first line is the header, and each later line is the row of one finished question. + A last line that is not complete JSON is left out. + """ + lines = path.read_text().splitlines() + header: dict[str, Any] = json.loads(lines[0])["header"] + rows = [] + for line in lines[1:]: + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + continue + return header, rows def shards_of(questions: Sequence[Question], workers: int, run_id: str) -> list[Shard]: @@ -615,16 +647,21 @@ def run_shard(job: Job, shard: Shard) -> Done: database, skips `capture`, and never deletes the database. With `job.smoke`, the questions run over one `MemoryStore` with a fixed clock, in the process. The worker opens its own model cache, embedder, and store, so it can run in its own process. It - prints one line for each question. + prints one line for each question, and appends the row of each finished question to + `job.rows_file`. A question that raises is listed in `failed`, its namespaces are + emptied, and the next question runs. CacheMiss stops the shard. """ from bench.models import Offline, Remote models = CachedModels(Offline() if job.replay else Remote(), CACHE_DIR / "replies.sqlite") embedder = embed.openai_embedder() rows: list[dict[str, Any]] = [] + failed: list[str] = [] def finish(question: Question, result: dict[str, Arm]) -> None: rows.append(row_of(question, result)) + if job.rows_file is not None: + append_row(job.rows_file, rows[-1]) marks = " ".join( f"{arm}={'yes' if found.correct else 'no'}" for arm, found in result.items() ) @@ -634,6 +671,10 @@ def finish(question: Question, result: dict[str, Arm]) -> None: flush=True, ) + def fail(question: Question, error: Exception) -> None: + failed.append(f"{question.id}: {type(error).__name__}: {error}") + print(f"{shard.run_id} {question.id} failed: {failed[-1]}", flush=True) + try: if job.smoke: store = MemoryStore(embedder=embedder, clock=lambda: FIXED_TIME) @@ -642,40 +683,88 @@ def empty(namespace: Sequence[str]) -> None: for record in store.list(namespace, limit=100_000): store.delete(record.namespace, record.key) + in_process: dict[str, Store] = dict.fromkeys(RANKINGS, store) for question in shard.questions: - in_process: dict[str, Store] = dict.fromkeys(RANKINGS, store) - result = run_question( - question, models, in_process, empty, shard.run_id, job.arms, job.config - ) + try: + result = run_question( + question, models, in_process, empty, shard.run_id, job.arms, job.config + ) + except CacheMiss: + raise + except Exception as error: + fail(question, error) + clean(empty, shard.run_id, question) + continue finish(question, result) elif job.kept is not None: + delete: Delete = "never" if job.reuse else "on_error" for question in shard.questions: name = f"{job.kept}-{question.id}" - delete: Delete = "never" if job.reuse else "on_error" - with local_store(job.url, name, embedder, delete) as (stores, _): - result = run_question( - question, - models, - stores, - _keep, - KEPT_RUN, - job.arms, - job.config, - capture=not job.reuse, - ) + try: + with local_store(job.url, name, embedder, delete) as (stores, _): + result = run_question( + question, + models, + stores, + _keep, + KEPT_RUN, + job.arms, + job.config, + capture=not job.reuse, + ) + except CacheMiss: + raise + except Exception as error: + fail(question, error) + continue finish(question, result) else: name = f"hotmemory-bench-{shard.run_id}" with local_store(job.url, name, embedder) as (stores, emptied): for question in shard.questions: - result = run_question( - question, models, stores, emptied, shard.run_id, job.arms, job.config - ) + try: + result = run_question( + question, models, stores, emptied, shard.run_id, job.arms, job.config + ) + except CacheMiss: + raise + except Exception as error: + fail(question, error) + clean(emptied, shard.run_id, question) + continue finish(question, result) finally: models.close() embedder.close() - return Done(rows, models.spent, models.calls, embedder.misses) + return Done(rows, models.spent, models.calls, embedder.misses, failed) + + +def clean(empty: Callable[[Sequence[str]], None], run_id: str, question: Question) -> None: + """Empty the namespaces of the arms of `question` after it failed. + + Raises RuntimeError with the text of the error if a call fails, so that the error + crosses a process pool. + """ + try: + for arm in ("retrieval", "memory"): + empty(("bench", "lme_s", run_id, arm, question.id)) + except Exception as error: + raise RuntimeError( + f"could not empty the namespaces of {question.id}: {type(error).__name__}: {error}" + ) from None + + +def _delete_databases(url: str, names: Collection[str]) -> None: + """Delete each local database whose name is in `names`.""" + from hotdata_framework import HotdataClient + + client = HotdataClient("local", "local", host=url) + try: + for database in client.list_managed_databases(): + if database.description in names: + client.delete_managed_database(database) + finally: + client.close() def _keep(namespace: Sequence[str]) -> None: @@ -736,6 +825,11 @@ def main(argv: Sequence[str] | None = None) -> Path | None: action="store_true", help="recall from the memory that a run with --keep kept, and skip capture", ) + parser.add_argument( + "--resume", + action="store_true", + help="skip the questions that the rows file of an earlier run of this command holds", + ) parser.add_argument("--tag", default="", help="a suffix for the name of the result file") parser.add_argument("--url", default="http://localhost:3000") args = parser.parse_args(argv) @@ -795,30 +889,89 @@ def main(argv: Sequence[str] | None = None) -> Path | None: kept = kept_name(args.sample, config, library_commit()) except RuntimeError as error: parser.error(str(error)) + + name = "e2e-smoke" if args.smoke else f"e2e-sample-{args.sample}" + name += f"-{args.tag}" if args.tag else "" + name += f"-first{args.limit}" if args.limit is not None else "" + path = RESULTS_DIR / f"{name}.json" + rows_file = RESULTS_DIR / f"{name}.rows.jsonl" + header = { + "arms": list(arms), + "config": asdict(config), + "sample": "smoke" if args.smoke else args.sample, + "limit": args.limit, + "kept": kept, + "reuse": args.reuse, + } + earlier: list[dict[str, Any]] = [] + if args.resume: + if not rows_file.exists(): + parser.error(f"--resume found no rows file {rows_file}") + found_header, earlier = read_rows(rows_file) + if found_header != header: + parser.error(f"the rows file {rows_file} is of another run: {found_header}") + elif rows_file.exists(): + parser.error(f"the rows file {rows_file} exists: pass --resume, or delete it") + finished = {row["id"] for row in earlier} + left = [question for question in questions if question.id not in finished] + if args.resume: + print(f"resuming: {len(finished)} questions finished, {len(left)} left") + + if kept is not None: names = [f"{kept}-{question.id}" for question in questions] found = existing_databases(args.url, names) - if args.keep and found: - parser.error(f"{len(found)} kept databases exist already, for example {found[0]}") - missing = sorted(set(names) - set(found)) - if args.reuse and missing: - parser.error(f"{len(missing)} kept databases do not exist, for example {missing[0]}") + if args.reuse: + missing = sorted(set(names) - set(found)) + if missing: + parser.error( + f"{len(missing)} kept databases do not exist, for example {missing[0]}" + ) + else: + done_names = {f"{kept}-{question_id}" for question_id in finished} + partial = [found_name for found_name in found if found_name not in done_names] + if partial and not args.resume: + parser.error(f"{len(found)} kept databases exist already, for example {found[0]}") + if partial: + _delete_databases(args.url, partial) + print(f"deleted {len(partial)} kept databases of questions that did not finish") print(f"{'keeping' if args.keep else 'reusing'} the memory as {kept}-") commit = _commit() started = datetime.now(UTC) run_id = f"{started:%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" clock_start = time.perf_counter() - shards = shards_of(questions, args.workers, run_id) - job = Job(args.url, arms, config, args.replay, len(questions), kept, args.reuse, args.smoke) - if len(shards) == 1: - done = [run_shard(job, shards[0])] + RESULTS_DIR.mkdir(exist_ok=True) + if not args.resume: + rows_file.write_text(json.dumps({"header": header}) + "\n") + shards = shards_of(left, args.workers, run_id) if left else [] + job = Job( + args.url, + arms, + config, + args.replay, + len(left), + kept, + args.reuse, + args.smoke, + rows_file, + ) + if len(shards) <= 1: + done = [run_shard(job, shard) for shard in shards] else: context = multiprocessing.get_context("spawn") with ProcessPoolExecutor(len(shards), mp_context=context) as pool: done = list(pool.map(run_shard, [job] * len(shards), shards)) - by_id = {row["id"]: row for found in done for row in found.rows} - rows = [by_id[question.id] for question in questions] spent = sum(found.spent for found in done) + failed = [failure for found in done for failure in found.failed] + if failed: + print(f"{len(failed)} questions failed; spent ${spent:.2f}") + for failure in failed: + print(f" {failure}") + print(f"the finished rows are in {rows_file}; run the command again with --resume") + return None + by_id = {row["id"]: row for row in earlier} + by_id |= {row["id"]: row for found in done for row in found.rows} + rows = [by_id[question.id] for question in questions] calls = sum(found.calls for found in done) misses = sum(found.embedder_misses for found in done) @@ -896,6 +1049,7 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "replay": args.replay, "workers": len(shards), "spent_usd": round(spent, 2), + "resumed_questions": len(finished), "model_calls": calls, "sessions_at_limit": None if args.reuse @@ -913,12 +1067,8 @@ def main(argv: Sequence[str] | None = None) -> Path | None: "summary": summaries, "questions": rows, } - RESULTS_DIR.mkdir(exist_ok=True) - name = "e2e-smoke" if args.smoke else f"e2e-sample-{args.sample}" - name += f"-{args.tag}" if args.tag else "" - name += f"-first{args.limit}" if args.limit is not None else "" - path = RESULTS_DIR / f"{name}.json" path.write_text(json.dumps(result_file, indent=1) + "\n") + rows_file.unlink() for arm in arms: found = summaries[arm]["all"] print(f"{arm}: {found['correct']}/{found['questions']} = {found['accuracy']:.3f}") diff --git a/bench/tests/test_e2e.py b/bench/tests/test_e2e.py index 970eaf1..6a8c8de 100644 --- a/bench/tests/test_e2e.py +++ b/bench/tests/test_e2e.py @@ -15,6 +15,9 @@ SAMPLES, SMOKE, Config, + Job, + Shard, + append_row, capture_session, choose_sample, episode_key, @@ -26,7 +29,9 @@ library_commit, local_store, main, + read_rows, run_question, + run_shard, sample_ids, shards_of, smoke_ids, @@ -512,3 +517,63 @@ def test_local_store_deletes_the_database_as_asked() -> None: if database.description in (names["kept"], names["never"]): client.delete_managed_database(database) client.close() + + +def test_rows_file_keeps_the_rows_and_drops_a_cut_line(tmp_path: Path) -> None: + path = tmp_path / "run.rows.jsonl" + path.write_text(json.dumps({"header": {"arms": ["memory"]}}) + "\n") + append_row(path, {"id": "q1"}) + append_row(path, {"id": "q2"}) + with path.open("a") as file: + file.write('{"id": "q3", "ar') + assert read_rows(path) == ({"arms": ["memory"]}, [{"id": "q1"}, {"id": "q2"}]) + + +class FailingModels(FakeModels): + """FakeModels whose reader raises for one question, like a lost connection.""" + + def __init__(self, question: str) -> None: + super().__init__() + self.question = question + self.spent = 0.0 + self.calls = 0 + + def __call__( + self, model: str, prompt: str, max_tokens: int, schema: dict[str, Any] | None = None + ) -> Reply: + if model == READER and self.question in prompt: + raise ConnectionError("Connection error.") + return super().__call__(model, prompt, max_tokens, schema) + + def close(self) -> None: + pass + + +class ClosingEmbedder: + misses = 0 + + def __call__(self, texts: Sequence[str]) -> list[list[float]]: + return fake_embedder(texts) + + def close(self) -> None: + pass + + +def test_run_shard_fails_one_question_and_keeps_the_rest( + questions: dict[str, Question], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import bench.e2e as e2e + from bench import embed + + first, second = questions["q_one"], questions["q_three"] + models = FailingModels(first.question) + monkeypatch.setattr(e2e, "CachedModels", lambda backend, path: models) + monkeypatch.setattr(embed, "openai_embedder", ClosingEmbedder) + rows_file = tmp_path / "run.rows.jsonl" + rows_file.write_text(json.dumps({"header": {}}) + "\n") + config = Config(session_episodes=True, episodes="search") + job = Job("", ("memory",), config, True, 2, smoke=True, rows_file=rows_file) + done = run_shard(job, Shard([first, second], "r1")) + assert [row["id"] for row in done.rows] == [second.id] + assert done.failed == [f"{first.id}: ConnectionError: Connection error."] + assert read_rows(rows_file)[1] == done.rows From 0856a0470493751c3b8cad794b9cd1d7e6867c3d Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 16:02:27 +0530 Subject: [PATCH 6/9] Phase 9 task 6: prove the reuse docs/benchmarks.md records the proof. On sample A, from the caches, with 4 workers: search at 64,000 with --keep took 762.4 s and equals the phase 8 run on all 54 rows. search at 32,000 and cited at 64,000 with --reuse took 13.0 s and 15.3 s, and equal their phase 8 runs on all 54 rows. search at 64,000 with --reuse took 11.1 s, with the same block SHA-256 as the --keep run for each question. The 54 kept databases hold 192 MB. The diagnosis of the --keep run read the kept databases in 7 s and equals the phase 8 diagnosis on all 9 misses. --- docs/benchmarks.md | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/docs/benchmarks.md b/docs/benchmarks.md index b59f10f..94cba85 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -420,6 +420,31 @@ session. A question of the type multi-session can need more sessions than the bl The type single-session-preference has the most misses with `search` at 64,000: 3 on sample A and 4 on sample B. +### Faster evaluation + +Phase 9 kept the memory of a run and reused it for runs that change only `recall`. +The section "Faster runs" of [bench/README.md](../bench/README.md) gives the options. Each run below is +`--replay` on sample A, with `--extractor detail --session-episodes` and 4 workers, so no +run called a model. Observed on 2026-10-16, at commit `630e3be`. The last commit that +changed `src/hotmemory` is `b5443ac`, the same code as the phase 8 runs. + +| Run | Correct of 54 | Time | Rows equal to | File | +|---|---|---|---|---| +| `search`, 64,000, `--keep` | 45 | 762.4 s | phase 8 `search` 64,000: 54 of 54 | `e2e-sample-a-p9-keep-search-64k.json` | +| `search`, 32,000, `--reuse` | 43 | 13.0 s | phase 8 `search` 32,000: 54 of 54 | `e2e-sample-a-p9-reuse-search-32k.json` | +| `cited`, 64,000, `--reuse` | 44 | 15.3 s | phase 8 `cited` 64,000: 54 of 54 | `e2e-sample-a-p9-reuse-cited-64k.json` | +| `search`, 64,000, `--reuse` | 45 | 11.1 s | the `--keep` run: 54 of 54, with the same block SHA-256 | `e2e-sample-a-p9-reuse-search-64k.json` | + +Two rows are equal in four values: the answer, the verdict, the block length, and the +episodes in the block. The `--keep` run also wrote the same count of facts for each question as +phase 8. The phase 8 runs of the same configurations took 1,133 to 2,085 seconds, and +those times include model calls that were not in the cache. The 54 kept databases hold +192 MB in the storage of the local engine. + +The diagnosis of the `--keep` run read the kept databases, and did not run `capture`. It +took 7 seconds and called no model. For each of the 9 misses, it gave the same cause, +marked facts, and evidence check as the phase 8 diagnosis of `search` at 64,000. + ## Known limits - The end-to-end numbers come from samples of 54 questions, so one question moves an arm From 298fbe7822e42a7a02f4b0877fb8e64773c620e3 Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 16:03:24 +0530 Subject: [PATCH 7/9] Phase 9 task 7: document the faster runs bench/README.md gets the options --keep, --reuse, --smoke, and --resume, and a section "Faster runs": the name of a kept database and its parts, the refusals, the smoke sample and why its first run pays for extraction, and the rows file. The diagnosis paragraph says that a result file with kept databases is read without capture. The result file name of a smoke run and the new field block_sha256 are listed. --- bench/README.md | 75 ++++++++++++++++++++++++++++++++++++++++++---- docs/benchmarks.md | 8 ++--- 2 files changed, 74 insertions(+), 9 deletions(-) diff --git a/bench/README.md b/bench/README.md index da92ae4..c7ce434 100644 --- a/bench/README.md +++ b/bench/README.md @@ -91,11 +91,16 @@ The options are these: | `--episodes` | `none`, `cited`, or `search`, the `episodes` of the memory recall; a value other than `none` needs `--session-episodes` | `none` | | `--replay` | call no model, and stop at the first reply that the cache does not hold | off | | `--workers` | the number of processes that run the questions | `1` | +| `--keep` | keep the memory of each question in a local database of its own; needs `--arms memory` | off | +| `--reuse` | recall from the kept memory, and skip `capture`; needs `--arms memory` | off | +| `--smoke` | run the 18 questions of the smoke sample on `MemoryStore`, in the process | off | +| `--resume` | run only the questions that the rows file of an earlier run does not hold | off | | `--tag` | a suffix for the name of the result file | none | | `--limit` | the first N questions of the sample | every question | -Each worker gets the questions in turn, and runs them over a new local database of its -own. Each table then holds one question at a time, as in a run with one worker. BM25 ranks +Each worker gets the questions in turn. Without `--keep` or `--reuse`, it runs them over a +new local database of its own, and empties the namespaces of each question after it. Each +table then holds one question at a time, as in a run with one worker. BM25 ranks over the whole table before the scope filter, so the rows of a second question in a table can change the ranking. An extractor variant names a prompt and a model. Each fact carries the source `session:`, where the hash is the first 12 hex digits of the SHA-256 of @@ -110,8 +115,66 @@ With `--session-episodes`, `capture` stores each session as an episode under the `session:` source. The block shows the source as `session-@1`. The source holds no run id, so a second run with another `--episodes` or `--recall-budget` replays `capture` from the cache. The result file lists the episode keys of each block in -`episodes_in_block`. The result file is `bench/results/e2e-sample-[-].json`, -and it records each option. +`episodes_in_block`, and the SHA-256 of each block in `block_sha256`. The result file is +`bench/results/e2e-sample-[-].json`, or `bench/results/e2e-smoke[-].json` +for a smoke run, and it records each option. + +## Faster runs + +The episodes value, the budget, and `k` act in `recall`, after `capture`. So a run that +changes only these values, or the reader, can use the memory of an earlier run. The +memory arm captures the memory once with `--keep`, and a later run with `--reuse` skips +`capture`: + +```sh +uv run --group bench --env-file .env python -m bench.e2e --run --arms memory --extractor detail --session-episodes --episodes search --recall-budget 64000 --workers 4 --keep +uv run --group bench --env-file .env python -m bench.e2e --run --arms memory --extractor detail --session-episodes --episodes cited --recall-budget 32000 --workers 4 --reuse +``` + +With `--keep`, each question runs over a new local database of its own, and the database +stays after the run. One database holds one question, because BM25 ranks the whole table +before the namespace filter, and the memory of a second question changes the block. The +name of the database is `hotmemory-kept------`: + +| Part | What it holds | +|---|---| +| `` | The variant name, and 6 hex digits of the SHA-256 of its model and prompt. | +| `` | `episodes` with `--session-episodes`, `sources` with fact sources, or `plain`. | +| `` | The `--memory-ranking`. | +| `` | The last commit that changed `src/hotmemory`. | + +The runner refuses `--keep` if `src/hotmemory` holds a change that is not committed, or if +a kept database of the run exists. If a question fails, its database is deleted. With +`--reuse`, the runner computes the same names, and refuses to run if a database does not +exist. It never deletes a kept database. A reused row has `null` for `facts_written` and +`sessions_at_limit`, because `capture` did not run. The result file records the name in +`memory.kept`. The cost estimate of a run with `--reuse` still counts the extraction, so it +is too high. If a commit changes `src/hotmemory`, no later run can use the kept databases. +Delete them by hand. + +A smoke run checks that a change runs and that the block looks right: + +```sh +uv run --group bench --env-file .env python -m bench.e2e --run --smoke --arms memory --extractor detail --session-episodes --episodes search --recall-budget 64000 +``` + +It runs the 18 questions of `bench/sample-smoke.json`, 3 of each type drawn from sample A +with the seed `20261016`, over one `MemoryStore` in the process. It decides no gain: 18 +questions can be off by about 23 points, and `MemoryStore` ranks by vector only. Because +the ranking differs, `capture` shows the extractor another list of known facts than a run +on the local engine. So the first smoke run of a configuration pays for its extraction, +and a later one replays it from the cache. `--smoke` refuses `--keep`, `--reuse`, +`--workers`, and `--sample`. + +Each run writes the file `bench/results/.rows.jsonl` as it goes, where `` is +the name of its result file. The first line holds the arms, the configuration, the sample, +the limit, and the kept name. Each later line holds the row of one finished question. If a +call to a model or to the engine fails, the question fails, and the next question runs. A +`CacheMiss` of `--replay` still stops the run. If a question failed, or the run stopped, +the runner writes no result file. Run the same command again with `--resume`: it skips the +questions that the rows file holds. It refuses a rows file of another configuration, and a +run without `--resume` refuses to start over a rows file. When every question has +finished, the runner writes the result file and deletes the rows file. ## The checks of the results @@ -131,7 +194,9 @@ uv run --group bench --env-file .env python -m bench.diagnose --result bench/res The command runs `capture` again with the ranking, the `k`, the `budget`, and the extractor variant that the result file records. The replay calls no model, and it stops at the first -reply that the cache does not hold. GPT-4o marks the stored facts that hold each answer. +reply that the cache does not hold. If the result file names kept databases, from a run +with `--keep` or `--reuse`, the command reads the memory of each question from its kept +database and does not run `capture`. GPT-4o marks the stored facts that hold each answer. Each miss then gets one of four causes: | Cause | Meaning | diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 94cba85..108cd25 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -422,10 +422,10 @@ sample A and 4 on sample B. ### Faster evaluation -Phase 9 kept the memory of a run and reused it for runs that change only `recall`. -The section "Faster runs" of [bench/README.md](../bench/README.md) gives the options. Each run below is -`--replay` on sample A, with `--extractor detail --session-episodes` and 4 workers, so no -run called a model. Observed on 2026-10-16, at commit `630e3be`. The last commit that +Phase 9 kept the memory of a run and reused it for runs that change only `recall`. The +section "Faster runs" of [bench/README.md](../bench/README.md) gives the options. Each run +below is `--replay` on sample A, with `--extractor detail --session-episodes` and 4 +workers, so no run called a model. Observed on 2026-10-16, at commit `630e3be`. The last commit that changed `src/hotmemory` is `b5443ac`, the same code as the phase 8 runs. | Run | Correct of 54 | Time | Rows equal to | File | From e9639ebe0d2442ab0af962df3bbbee25e9242ae2 Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 17:03:20 +0530 Subject: [PATCH 8/9] Phase 9: record the smoke run, and the docs audit docs/benchmarks.md records the smoke run: the first run paid $1.07 for 842 extractor calls in 2,080 s, and the replay took 12.7 s at $0 with the same answer, verdict, and block SHA-256 for each of the 18 questions. The first replay command waited 563 s in all, for a cause that was not found, and a later one took 14.9 s. The docs audit: the docstring of bench.e2e names --keep, --reuse, the rows file, and --resume, and bench/README.md says that a worker stops if it cannot empty the namespaces of a failed question. --- bench/README.md | 3 ++- bench/e2e.py | 5 ++++- docs/benchmarks.md | 16 ++++++++++++++++ 3 files changed, 22 insertions(+), 2 deletions(-) diff --git a/bench/README.md b/bench/README.md index c7ce434..b79165d 100644 --- a/bench/README.md +++ b/bench/README.md @@ -169,7 +169,8 @@ and a later one replays it from the cache. `--smoke` refuses `--keep`, `--reuse` Each run writes the file `bench/results/.rows.jsonl` as it goes, where `` is the name of its result file. The first line holds the arms, the configuration, the sample, the limit, and the kept name. Each later line holds the row of one finished question. If a -call to a model or to the engine fails, the question fails, and the next question runs. A +call to a model or to the engine fails, the question fails, and the next question runs. If +the runner then cannot empty the namespaces of that question, the worker stops. A `CacheMiss` of `--replay` still stops the run. If a question failed, or the run stopped, the runner writes no result file. Run the same command again with `--resume`: it skips the questions that the rows file holds. It refuses a rows file of another configuration, and a diff --git a/bench/e2e.py b/bench/e2e.py index 3c55ca4..ee2df81 100644 --- a/bench/e2e.py +++ b/bench/e2e.py @@ -10,7 +10,10 @@ uv run --group bench --env-file .env python -m bench.e2e --run --smoke --arms memory With `--smoke`, it runs the 18 questions of `bench/sample-smoke.json` on `MemoryStore`, in -the process. It writes one JSON file into `bench/results/` with the configuration and the results. +the process. With `--keep`, it keeps the memory of each question, and `--reuse` recalls from +it with no `capture`. It appends the row of each finished question to a rows file, and +`--resume` skips those rows. At the end, it writes one JSON file into `bench/results/` with +the configuration and the results. """ from __future__ import annotations diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 108cd25..06f4236 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -445,6 +445,22 @@ The diagnosis of the `--keep` run read the kept databases, and did not run `capt took 7 seconds and called no model. For each of the 9 misses, it gave the same cause, marked facts, and evidence check as the phase 8 diagnosis of `search` at 64,000. +The smoke run uses the same options with `search` at 64,000 characters, on the 18 +questions of `bench/sample-smoke.json` and `MemoryStore`, in one process. Its first run +paid for the extraction. `MemoryStore` ranks by vector only, so the extractor saw other +lists of known facts than in the runs on the local engine. Observed on +2026-10-16, at commit `298fbe7`: + +| Run | Correct of 18 | Time | Cost | Model calls | File | +|---|---|---|---|---|---| +| First, `--run` | 16 | 2,080.0 s | $1.07 | 842 | `e2e-smoke-p9-first.json` | +| Again, `--replay` | 16 | 12.7 s | $0 | 0 | `e2e-smoke-p9-cached.json` | + +The replay gave the same answer, verdict, and block SHA-256 for each of the 18 questions. +The time in the table is `duration_s` of the result file. The whole command took 563 +seconds the first time, and 14.9 seconds on a later try. The cause of the first wait was +not found. + ## Known limits - The end-to-end numbers come from samples of 54 questions, so one question moves an arm From 9af1620ac116298077b14b76eed5cd7d2012edc5 Mon Sep 17 00:00:00 2001 From: Rohan Dsouza Date: Fri, 16 Oct 2026 17:18:02 +0530 Subject: [PATCH 9/9] Phase 9: let a smoke run take --workers --smoke now accepts --workers. Each worker runs its questions over a MemoryStore of its own, with the fixed clock, and empties the memory of each question after it, so a question gives the same row in any worker. A replay of the smoke run with 4 workers took 3.8 s and gave the same answer, verdict, block SHA-256, and count of facts for each of the 18 questions as the first run. A first run with 4 workers, which pays for extraction, was not measured. --- bench/README.md | 9 +++++---- bench/e2e.py | 18 +++++++++--------- bench/tests/test_e2e.py | 2 +- docs/benchmarks.md | 5 ++++- 4 files changed, 19 insertions(+), 15 deletions(-) diff --git a/bench/README.md b/bench/README.md index b79165d..adec5cc 100644 --- a/bench/README.md +++ b/bench/README.md @@ -93,7 +93,7 @@ The options are these: | `--workers` | the number of processes that run the questions | `1` | | `--keep` | keep the memory of each question in a local database of its own; needs `--arms memory` | off | | `--reuse` | recall from the kept memory, and skip `capture`; needs `--arms memory` | off | -| `--smoke` | run the 18 questions of the smoke sample on `MemoryStore`, in the process | off | +| `--smoke` | run the 18 questions of the smoke sample on `MemoryStore`, one store in each worker | off | | `--resume` | run only the questions that the rows file of an earlier run does not hold | off | | `--tag` | a suffix for the name of the result file | none | | `--limit` | the first N questions of the sample | every question | @@ -159,12 +159,13 @@ uv run --group bench --env-file .env python -m bench.e2e --run --smoke --arms me ``` It runs the 18 questions of `bench/sample-smoke.json`, 3 of each type drawn from sample A -with the seed `20261016`, over one `MemoryStore` in the process. It decides no gain: 18 +with the seed `20261016`. Each worker runs its questions over a `MemoryStore` of its own, +in its own process, and empties the memory of each question after it. It decides no gain: 18 questions can be off by about 23 points, and `MemoryStore` ranks by vector only. Because the ranking differs, `capture` shows the extractor another list of known facts than a run on the local engine. So the first smoke run of a configuration pays for its extraction, -and a later one replays it from the cache. `--smoke` refuses `--keep`, `--reuse`, -`--workers`, and `--sample`. +and a later one replays it from the cache. Pass `--workers 4` to make the first run +faster. `--smoke` refuses `--keep`, `--reuse`, and `--sample`. Each run writes the file `bench/results/.rows.jsonl` as it goes, where `` is the name of its result file. The first line holds the arms, the configuration, the sample, diff --git a/bench/e2e.py b/bench/e2e.py index ee2df81..a488ec0 100644 --- a/bench/e2e.py +++ b/bench/e2e.py @@ -9,11 +9,11 @@ uv run --group bench --env-file .env python -m bench.e2e --replay --arms memory uv run --group bench --env-file .env python -m bench.e2e --run --smoke --arms memory -With `--smoke`, it runs the 18 questions of `bench/sample-smoke.json` on `MemoryStore`, in -the process. With `--keep`, it keeps the memory of each question, and `--reuse` recalls from -it with no `capture`. It appends the row of each finished question to a rows file, and -`--resume` skips those rows. At the end, it writes one JSON file into `bench/results/` with -the configuration and the results. +With `--smoke`, it runs the 18 questions of `bench/sample-smoke.json` on `MemoryStore`, one +store in each worker. With `--keep`, it keeps the memory of each question, and `--reuse` +recalls from it with no `capture`. It appends the row of each finished question to a rows +file, and `--resume` skips those rows. At the end, it writes one JSON file into +`bench/results/` with the configuration and the results. """ from __future__ import annotations @@ -648,7 +648,7 @@ def run_shard(job: Job, shard: Shard) -> Done: id>`, and the database keeps the memory of the question after the run. A question that raises deletes its database. With `job.reuse` as well, each question opens its kept database, skips `capture`, and never deletes the database. With `job.smoke`, the - questions run over one `MemoryStore` with a fixed clock, in the process. The worker + questions run over one `MemoryStore` of the worker, with a fixed clock. The worker opens its own model cache, embedder, and store, so it can run in its own process. It prints one line for each question, and appends the row of each finished question to `job.rows_file`. A question that raises is listed in `failed`, its namespaces are @@ -787,7 +787,7 @@ def main(argv: Sequence[str] | None = None) -> Path | None: parser.add_argument( "--smoke", action="store_true", - help="run the 18 questions of the smoke sample on MemoryStore, in the process", + help="run the 18 questions of the smoke sample on MemoryStore, in each worker", ) parser.add_argument("--retrieval-ranking", choices=RANKINGS, default="fused") parser.add_argument("--memory-ranking", choices=RANKINGS, default="fused") @@ -843,8 +843,8 @@ def main(argv: Sequence[str] | None = None) -> Path | None: parser.error("workers must be at least 1") if args.keep and args.reuse: parser.error("pass --keep or --reuse, not both") - if args.smoke and (args.keep or args.reuse or args.workers != 1 or args.sample != "a"): - parser.error("--smoke runs alone: no --keep, --reuse, --workers, or --sample") + if args.smoke and (args.keep or args.reuse or args.sample != "a"): + parser.error("--smoke runs alone: no --keep, --reuse, or --sample") if (args.keep or args.reuse) and arms != ("memory",): parser.error(f"--{'keep' if args.keep else 'reuse'} needs --arms memory") try: diff --git a/bench/tests/test_e2e.py b/bench/tests/test_e2e.py index 6a8c8de..e5eeeef 100644 --- a/bench/tests/test_e2e.py +++ b/bench/tests/test_e2e.py @@ -453,7 +453,7 @@ def git(*args: str) -> str: (["--keep", "--arms", "retrieval,memory"], "--keep needs --arms memory"), (["--reuse", "--arms", "full"], "--reuse needs --arms memory"), (["--keep", "--reuse", "--arms", "memory"], "pass --keep or --reuse, not both"), - (["--smoke", "--workers", "4"], "--smoke runs alone"), + (["--smoke", "--keep"], "--smoke runs alone"), (["--smoke", "--sample", "b"], "--smoke runs alone"), ], ) diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 06f4236..037d061 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -455,8 +455,11 @@ lists of known facts than in the runs on the local engine. Observed on |---|---|---|---|---|---| | First, `--run` | 16 | 2,080.0 s | $1.07 | 842 | `e2e-smoke-p9-first.json` | | Again, `--replay` | 16 | 12.7 s | $0 | 0 | `e2e-smoke-p9-cached.json` | +| Again, `--replay --workers 4` | 16 | 3.8 s | $0 | 0 | `e2e-smoke-p9-cached-w4.json` | -The replay gave the same answer, verdict, and block SHA-256 for each of the 18 questions. +Each replay gave the same answer, verdict, and block SHA-256 for each of the 18 questions. +With 4 workers, each worker holds a `MemoryStore` of its own. A first run with 4 workers +was not measured. The time in the table is `duration_s` of the result file. The whole command took 563 seconds the first time, and 14.9 seconds on a later try. The cause of the first wait was not found.