diff --git a/CHANGELOG.md b/CHANGELOG.md index 23b9a34..d543a8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ changelog, see the public [kimetsu CHANGELOG](../CHANGELOG.md). ## Bench tool +### Unreleased — paired production MCP evidence + +- Persistent MCP query measurements, delivered capsule text and optional final + answerability; scoped Windows working-set and peak memory observations. +- Explicit per-side reranker, cutoff, threads and fact-guard settings with + effective configuration checks and binary/fixture/runner fingerprints. +- Separate positive and negative denominators, auditable failures, stale-query + accounting, dated compressed-capsule matching and timeout descendant cleanup. +- Rust harness validation: 132 tests passed; paired-runner validation: 18 Python + tests passed. The linked Kimetsu audit evaluates metadata on every repeat. + ### v0.5 — 2026-06-05 — `kstress` brain stress test New second binary `kstress` profiles the brain (not agent tasks) at scale — diff --git a/Cargo.lock b/Cargo.lock index 8d586b3..12e8961 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1787,6 +1787,7 @@ dependencies = [ "tempfile", "time", "ureq 2.12.1", + "windows-sys 0.61.2", ] [[package]] @@ -1798,6 +1799,7 @@ dependencies = [ "hf-hub", "ignore", "kimetsu-core", + "ort", "regex", "rusqlite", "serde", diff --git a/Cargo.toml b/Cargo.toml index aa3995e..f5722b5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,3 +62,6 @@ rusqlite = { version = "0.37", features = ["bundled"] } # Used by the LongMemEval driver for per-instance temp brain workspaces. tempfile = "3" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_System_ProcessStatus", "Win32_System_Threading"] } diff --git a/README.md b/README.md index d05cf9d..f3caf48 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,21 @@ taking ours on faith. ## What it measures +### Paired memory evidence (unreleased) + +BrainBenchmark now measures persistent production MCP context delivery, retaining +rankings, delivered capsules, optional `answerability`, latency, serialized byte +cost and supported Windows process-memory measurements. The paired runner verifies +effective per-side model, floor and guard settings, alternates repeat order, and +records binary/fixture fingerprints. Errors and unpaired cases are not successful +quality observations. Historical binaries without answerability remain supported. + +See the public [structured-fact report](https://github.com/RodCor/kimetsu/blob/281c6df/docs/audits/2026-09-07-structured-facts.md) +for the frozen fixture, commands and all-repeat metadata scoring. It reports +evidence delivery rather than generated-answer accuracy; it does not replace +older overall BrainBench scores. This branch requires the companion Kimetsu +hardening changes in its parent checkout for the path dependencies. + `kbench` runs one or more Terminal-Bench tasks under multiple agent configurations and produces a side-by-side comparison: diff --git a/datasets/brainbench/agent-memory-contract.json b/datasets/brainbench/agent-memory-contract.json new file mode 100644 index 0000000..9a3e5af --- /dev/null +++ b/datasets/brainbench/agent-memory-contract.json @@ -0,0 +1,262 @@ +{ + "scenarios": [ + { + "id": "exact-code-evidence", + "dimension": "retrieval", + "tier": "easy", + "description": "Exact identifiers, multi-fact recall, and unrelated no-answer queries.", + "memories": [ + { + "key": "stderr", + "text": "MCP diagnostic logs must go to stderr. Stdout is reserved for JSON-RPC messages." + }, + { + "key": "foreign-keys", + "text": "Enable SQLite foreign key constraints explicitly on every new connection using PRAGMA foreign_keys = ON." + }, + { + "key": "blocking", + "text": "Move long synchronous operations into tokio::task::spawn_blocking instead of blocking asynchronous executor threads." + }, + { + "key": "test-isolation", + "text": "Initialize a Git boundary in each temporary test workspace so repository discovery cannot write to the real project." + } + ], + "queries": [ + { + "query": "Where must MCP diagnostic logs be written?", + "relevant": [ + "stderr" + ] + }, + { + "query": "How do I enable SQLite foreign key constraints on a new connection?", + "relevant": [ + "foreign-keys" + ] + }, + { + "query": "How should I handle a long synchronous operation in the Tokio executor?", + "relevant": [ + "blocking" + ] + }, + { + "query": "What is the weather forecast for tomorrow?", + "relevant": [] + }, + { + "query": "What is the current production database password?", + "relevant": [] + } + ] + }, + { + "id": "related-but-unanswerable", + "dimension": "retrieval", + "tier": "hard", + "description": "Related vocabulary must not imply that a stored fact answers a different question.", + "memories": [ + { + "key": "http-port", + "text": "The local development HTTP server listens on port 4317." + }, + { + "key": "storage", + "text": "The project stores local agent memories in a SQLite database." + }, + { + "key": "logs", + "text": "Development request logs include the route and elapsed milliseconds." + } + ], + "queries": [ + { + "query": "Which port does the local development HTTP server use?", + "relevant": [ + "http-port" + ] + }, + { + "query": "What authentication password does the local development HTTP server require?", + "relevant": [] + }, + { + "query": "Which encryption key protects the SQLite database?", + "relevant": [] + }, + { + "query": "What is the production request log retention duration?", + "relevant": [] + } + ] + }, + { + "id": "cross-language-code", + "dimension": "retrieval", + "tier": "hard", + "description": "Small exploratory Spanish query track against English code memories.", + "memories": [ + { + "key": "stderr", + "text": "Write MCP diagnostic messages to stderr because stdout carries the JSON-RPC protocol." + }, + { + "key": "foreign-keys", + "text": "SQLite foreign key enforcement must be enabled separately for every connection with PRAGMA foreign_keys = ON." + }, + { + "key": "blocking", + "text": "Use tokio::task::spawn_blocking for long synchronous work inside an asynchronous application." + } + ], + "queries": [ + { + "query": "¿Dónde deben escribirse los mensajes de diagnóstico MCP para no romper JSON-RPC?", + "relevant": [ + "stderr" + ] + }, + { + "query": "¿Cómo se activan las claves foráneas de SQLite en cada conexión?", + "relevant": [ + "foreign-keys" + ] + }, + { + "query": "¿Cómo ejecuto trabajo síncrono largo sin bloquear Tokio?", + "relevant": [ + "blocking" + ] + }, + { + "query": "¿Cuál es el precio actual de la electricidad?", + "relevant": [] + } + ] + }, + { + "id": "multi-fact-retrieval", + "dimension": "retrieval", + "tier": "medium", + "description": "Fraction recall requires both expected facts; finding either is only a hit.", + "memories": [ + { + "key": "git-boundary", + "text": "For safe temporary memory tests, run git init in the temporary workspace to establish a repository boundary." + }, + { + "key": "user-brain", + "text": "For isolated memory tests, set KIMETSU_USER_BRAIN=0 to prevent cross-project memory leakage." + }, + { + "key": "release", + "text": "Release packaging produces a compressed archive containing the executable." + } + ], + "queries": [ + { + "query": "How do temporary memory tests establish a Git boundary and prevent user brain leakage?", + "relevant": [ + "git-boundary", + "user-brain" + ] + }, + { + "query": "Which executable signing certificate is used for releases?", + "relevant": [] + } + ] + }, + { + "id": "live-temporal-applicability", + "dimension": "retrieval", + "tier": "hard", + "description": "Future starts and expired claims must not appear in current context; these are storage validity fields, not text hints.", + "memories": [ + { + "key": "expired", + "text": "The Atlas staging listener uses port 4001.", + "valid_to": "2020-01-01T00:00:00Z" + }, + { + "key": "future", + "text": "The Atlas staging listener uses port 4003.", + "valid_from": "2099-01-01T00:00:00Z" + }, + { + "key": "current", + "text": "The Atlas staging listener uses port 4002." + } + ], + "queries": [ + { + "query": "Which port does the Atlas staging listener currently use?", + "relevant": [ + "current" + ], + "stale": [ + "expired", + "future" + ] + }, + { + "query": "What authentication password does the Atlas staging listener require?", + "relevant": [], + "stale": [ + "expired", + "future" + ] + } + ] + }, + { + "id": "persistent-mcp-observes-new-writes", + "dimension": "workflow", + "tier": "medium", + "description": "A persistent MCP process must see new claims written by another process, while related unknowns remain unanswered.", + "workflow": { + "seed": [], + "episodes": [ + { + "task": "Which port does the Zephyr local development HTTP server use?", + "relevant": [], + "record": [ + { + "key": "port", + "text": "The Zephyr local development HTTP server listens on port 5243." + } + ] + }, + { + "task": "Which port does the Zephyr local development HTTP server use?", + "relevant": [ + "port" + ] + }, + { + "task": "What authentication password does the Zephyr HTTP server require?", + "relevant": [] + }, + { + "task": "Where should MCP diagnostic logs be written?", + "relevant": [], + "record": [ + { + "key": "logs", + "text": "MCP diagnostic logs go to stderr; stdout is reserved for JSON-RPC messages." + } + ] + }, + { + "task": "Where should MCP diagnostic logs be written?", + "relevant": [ + "logs" + ] + } + ] + } + } + ] +} diff --git a/scripts/PAIRED_BRAINBENCH.md b/scripts/PAIRED_BRAINBENCH.md new file mode 100644 index 0000000..c086cd9 --- /dev/null +++ b/scripts/PAIRED_BRAINBENCH.md @@ -0,0 +1,36 @@ +# Compare two Kimetsu builds without a reader model + +Build the current `kbench` once and use that identical harness for both binaries: + +```powershell +cargo build --release --bin kbench --locked --offline +python scripts/compare_brainbench.py --kbench target/release/kbench.exe --baseline path/to/baseline/kimetsu.exe --candidate path/to/candidate/kimetsu.exe --dataset datasets/brainbench/agent-memory-contract.json --budget-tokens 512 --repeats 3 --out local/paired-memory +``` + +Use an existing shared `FASTEMBED_CACHE_DIR` to avoid model downloads. The binaries must support embeddings. The runner disables the user brain and permits only non-generative dimensions. Every scenario uses a temporary isolated project. It never uses your live brain. + +The runner alternates baseline/candidate order, fixes one scenario worker, retains every report and stderr log, and records SHA-256 fingerprints for binaries, the harness, the dataset and referenced fixture files. It does not force identical model defaults: a binary-default comparison intentionally includes changes to those defaults. Pin relevant model/environment settings for an algorithm-only experiment; recorded overrides make that distinction reviewable. + +Each scenario is paired by dimension and ID. Repeats are averaged within a scenario, not counted as additional independent cases. A scenario with a skipped or errored observation on either side is reported as unpaired and excluded from quality deltas; execution failures are not scores. The exploratory confidence interval bootstraps scenario IDs. Related scenarios are still correlated: a release claim requires held-out task/repository families and real task-success measurements. Changed/duplicate scenario identities are errors. + +`comparison.json` is written as the run progresses. A command failure, timeout, invalid JSON/report response, or final identity-validation failure leaves it with `status: "incomplete"`, structured failure evidence, and all completed run records; the process exits nonzero and does not write a completed Markdown comparison. + +The whole-run timeout terminates descendants before reaping `kbench` (Windows `taskkill /T`, Unix process group). This prevents a hung MCP inference child from surviving a timed-out comparison and competing with the next run. The deadline covers seeding and queries together; it is not a per-query latency cutoff. + +BrainBench's headline now weights measured dimensions equally; the old scenario-weighted average remains a diagnostic. No-answer queries score abstention, not the vacuous recall of an empty relevant set. Positive recall and negative injection rates use separate denominators, and stale correctness is reported as unavailable when no stale cases exist. Unmatched or ambiguous returned capsules retain their rank and count as injected material. + +Retrieval and workflow scenarios query the production `kimetsu_brain_context` tool through a persistent stdio MCP process. Every temporary project sets `broker.warm_start=false`, and requests set `include_ambient=false`: gold labels cover returned capsules, not unsolicited digest/profile/resume text. Measurements therefore cover production MCP retrieval under this controlled session configuration; they exclude warm-start construction/delivery and are not default whole-first-turn costs. Continuity is tested separately with warm starts enabled. Workflow writes happen through a separate process while MCP stays alive, exercising index freshness. The render-contract dimension retains its explicit CLI rendering check. Invalid MCP responses fail the scenario instead of masquerading as abstention. Current-context retrieval scores become zero if any explicitly stale gold item is delivered in the top four, even below the correct answer; the older ordering-only resolution metric remains diagnostic. + +Query observations retain hit@4, fraction recall@4, MRR, negative injection and stale injection separately. Repeats are averaged within the same query for quality summaries. Timing reports distinguish the first query in each process from subsequent queries. First-query timing is not a disk-cache-cold measurement; process startup is recorded separately. Subsequent p50/p95 are descriptive pooled measurements, not independent trials or a confidence interval. Full-run timings also include seeding and process startup. + +Response sizes include UTF-8 model text, the serialized MCP result, and the full JSON-RPC response line. `reported_used_tokens` is retained as a diagnostic: old heuristic estimates and new conservative byte bounds are not directly comparable. These byte measurements are not provider tokenizer counts or billed-token measurements. + +On Windows, each query also samples the MCP child's current and lifetime peak working set via `GetProcessMemoryInfo`, after the latency clock stops. The comparison reports the largest observed process peak; it excludes `kbench`, separate ingestion processes, other agents and system-wide model caches. A working set includes resident shared pages, so it is not private allocation or a sum of whole-agent RAM. Unsupported platforms or failed samples report unavailable. [Windows counter definitions](https://learn.microsoft.com/en-us/windows/win32/api/psapi/ns-psapi-process_memory_counters). + +For runtime experiments use `--baseline-threads 0 --candidate-threads 4` with the same binary on both sides (`0` removes `KIMETSU_INTRA_THREADS`; omission inherits it). `KBENCH_RERANKER` sets `embedder.reranker` only inside temporary benchmark projects and is recorded; the binary must actually honor that setting on MCP for a model comparison to be valid. Keep other settings fixed and avoid concurrent builds or other inference during timing runs. + +For a paired model comparison, use the same binary on both sides with `--baseline-reranker ms-marco-tinybert-l-2-v2 --candidate-reranker ms-marco-minilm-l-4-v2`. Side overrides take precedence over `KBENCH_RERANKER`; `off` explicitly disables reranking. The harness first sets `retrieval.level=custom` in each temporary project so the new-project `deep` preset cannot overwrite the requested model. Run order still alternates, and the comparison fingerprints the Python runner as well as `kbench` and the binaries. Model comparisons require compatible locally cached models and a binary that honors its MCP reranker configuration. + +The small checked-in fixture is a regression and exploratory language track, not a comprehensive held-out benchmark. It includes temporal validity fields and persistent-process write visibility. Final host token consumption and complete-agent success require the corresponding serving and agent evaluations. + +Use `--baseline-rerank-floor` and `--candidate-rerank-floor` for explicit final cross-encoder admission thresholds in [0, 1]. Omission inherits `KBENCH_RERANK_FLOOR`, or the binary default when unset. The harness writes `broker.rerank_min_score` only in temporary projects. Older binaries need this option omitted. Thresholds are model-specific scores, not calibrated probabilities; cosine gates remain active. Each side records its effective override alongside the model. diff --git a/scripts/compare_brainbench.py b/scripts/compare_brainbench.py new file mode 100644 index 0000000..120f812 --- /dev/null +++ b/scripts/compare_brainbench.py @@ -0,0 +1,410 @@ +"""Paired, reader-free BrainBench runs through the same harness and fixture. + +Standard library only. Alternates baseline/candidate order and averages repeats +within each scenario before estimating uncertainty. Never starts a reader or +write-precision generation task. Timings include CLI startup, seeding and queries. +""" +import argparse +import hashlib +import json +import math +import os +from pathlib import Path +import random +import signal +import statistics +import subprocess +import time + + +def run_owned_tree(cmd, *, env, timeout): + """Kill descendants while the parent still exists, before reaping it. + + subprocess.run kills only the parent on timeout. Inference children can + otherwise survive, including on Windows where process groups do not die + with their parent. The timeout applies to the whole benchmark invocation. + """ + options = dict(creationflags=subprocess.CREATE_NEW_PROCESS_GROUP) if os.name == "nt" else dict(start_new_session=True) + process = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + encoding="utf-8", errors="strict", **options) + try: + stdout, stderr = process.communicate(timeout=timeout) + except BaseException as error: + if os.name == "nt": + # /T walks descendants before /F terminates their parent; do not + # call process.kill first, which loses Windows tree ancestry. + subprocess.run(["taskkill", "/PID", str(process.pid), "/T", "/F"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=15) + else: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + if process.poll() is None: + process.kill() + stdout, stderr = process.communicate(timeout=15) + if isinstance(error, subprocess.TimeoutExpired): + raise subprocess.TimeoutExpired(cmd, timeout, output=stdout, stderr=stderr) from error + raise + return subprocess.CompletedProcess(cmd, process.returncode, stdout, stderr) + +OFFLINE_DIMENSIONS = { + "retrieval", "dedup", "importance", "forgetting", "calibration", + "poisoning", "render-contract", "graph", "workflow", +} + +REPORT_ROW_FIELDS = {"id", "dimension", "score", "skipped", "detail"} + + +def validate_report(report): + if not isinstance(report, dict) or not isinstance(report.get("scenarios"), list): + raise ValueError("report must be an object with a scenarios array") + configuration = report.get("session_configuration") + if (not isinstance(configuration, dict) or configuration.get("warm_start") is not False + or configuration.get("include_ambient") is not False): + raise ValueError("session_configuration must confirm warm_start=false and include_ambient=false") + for index, row in enumerate(report["scenarios"]): + if not isinstance(row, dict): + raise ValueError(f"scenario {index} must be an object") + missing = REPORT_ROW_FIELDS - set(row) + if missing: + raise ValueError(f"scenario {index} lacks fields: {', '.join(sorted(missing))}") + if not isinstance(row["id"], str) or not isinstance(row["dimension"], str): + raise ValueError(f"scenario {index} identity must contain strings") + if (isinstance(row["score"], bool) or not isinstance(row["score"], (int, float)) + or not math.isfinite(row["score"]) or not 0 <= row["score"] <= 1): + raise ValueError(f"scenario {index} has an invalid score") + if not isinstance(row["skipped"], bool) or not isinstance(row["detail"], str): + raise ValueError(f"scenario {index} has invalid status fields") + + +def indexed(report): + validate_report(report) + result = {} + for row in report["scenarios"]: + key = f"{row['dimension']}/{row['id']}" + if key in result: + raise ValueError(f"duplicate scenario identity: {key}") + result[key] = row + return result + + +def measurement_summary(reports, paired_keys): + groups = {} + for run in reports: + for scenario in run["scenarios"]: + identity = f"{scenario['dimension']}/{scenario['id']}" + if identity not in paired_keys: + continue + observations = scenario.get("observations", []) + if not isinstance(observations, list): + raise ValueError("observations must be an array") + for index, observation in enumerate(observations): + if not isinstance(observation, dict) or not isinstance(observation.get("query"), str): + raise ValueError("query observation requires a query string") + key = (identity, index, observation["query"]) + groups.setdefault(key, []).append(observation) + if not groups: + return None + metric_names = ["positive_recall_at_4", "positive_hit_at_4", "positive_mrr", "negative_injection", "stale_injection"] + metrics = {name: [] for name in metric_names} + first, subsequent, text_bytes, result_bytes = [], [], [], [] + working_sets, peak_working_sets = [], [] + count = 0 + for observations in groups.values(): + for name in metric_names: + values = [o.get(name) for o in observations if o.get(name) is not None] + if any(not isinstance(v, (int, float, bool)) or not math.isfinite(v) or not 0 <= v <= 1 for v in values): + raise ValueError(f"invalid query metric {name}") + if values: + metrics[name].append(statistics.mean(values)) + for observation in observations: + count += 1 + if not isinstance(observation.get("first_query"), bool): + raise ValueError("query observation requires first-query classification") + for field in ["latency_ms", "model_text_bytes", "mcp_result_bytes"]: + value = observation.get(field) + if isinstance(value, bool) or not isinstance(value, (int,float)) or not math.isfinite(value) or value < 0: + raise ValueError(f"invalid query measurement {field}") + (first if observation["first_query"] else subsequent).append(observation["latency_ms"]) + text_bytes.append(observation["model_text_bytes"]) + result_bytes.append(observation["mcp_result_bytes"]) + for field, values in [("working_set_bytes", working_sets), ("peak_working_set_bytes", peak_working_sets)]: + value = observation.get(field) + if value is not None: + if isinstance(value, bool) or not isinstance(value, (int,float)) or not math.isfinite(value) or value < 0: + raise ValueError(f"invalid memory measurement {field}") + values.append(value) + avg = lambda values: statistics.mean(values) if values else None + def percentile(values, p): + return sorted(values)[max(0, math.ceil(len(values)*p)-1)] if values else None + return dict(unique_queries=len(groups), query_observations=count, + positive_queries=len(metrics["positive_recall_at_4"]), negative_queries=len(metrics["negative_injection"]), + stale_queries=len(metrics["stale_injection"]), + positive_recall_at_4=avg(metrics["positive_recall_at_4"]), + positive_hit_at_4=avg(metrics["positive_hit_at_4"]), positive_mrr=avg(metrics["positive_mrr"]), + negative_injection_rate=avg(metrics["negative_injection"]), stale_injection_rate=avg(metrics["stale_injection"]), + first_query_mean_ms=avg(first), subsequent_query_p50_ms=percentile(subsequent,.5), + subsequent_query_p95_ms=percentile(subsequent,.95), subsequent_observations=len(subsequent), + mean_model_text_bytes=avg(text_bytes), mean_mcp_result_bytes=avg(result_bytes), + memory_observations=len(working_sets), mean_mcp_working_set_bytes=avg(working_sets), + max_mcp_peak_working_set_bytes=max(peak_working_sets) if peak_working_sets else None, + note="Quality averages repeats per query; latency percentiles pool repeated observations descriptively, not as independent evidence. First query includes model loading where applicable; server/process initialization is recorded separately.") + + +def compare_reports(baseline, candidate): + if not baseline or len(baseline) != len(candidate): + raise ValueError("equal nonempty repeat counts are required") + bases, candidates = list(map(indexed, baseline)), list(map(indexed, candidate)) + keys = set(bases[0]) + if not keys or any(set(run) != keys for run in bases + candidates): + raise ValueError("scenario sets differ or are empty; comparisons must use identical fixtures") + rows, unpaired, unpaired_details = [], [], [] + for key in sorted(keys): + reasons = [] + for label, runs in (("baseline", bases), ("candidate", candidates)): + for repeat, run in enumerate(runs, 1): + row = run[key] + if row["skipped"]: + reasons.append(dict(side=label, repeat=repeat, kind="skipped", + detail=row["detail"])) + elif row["detail"].startswith("error:"): + reasons.append(dict(side=label, repeat=repeat, kind="error", + detail=row["detail"])) + if reasons: + unpaired.append(key) + unpaired_details.append(dict(identity=key, reasons=reasons)) + continue + a = statistics.mean(run[key]["score"] for run in bases) + b = statistics.mean(run[key]["score"] for run in candidates) + rows.append(dict(identity=key, dimension=bases[0][key]["dimension"], + baseline=a, candidate=b, delta=b-a)) + dimensions = {} + for dimension in sorted({row["dimension"] for row in rows}): + group = [row for row in rows if row["dimension"] == dimension] + deltas = [row["delta"] for row in group] + interval = None + if len(deltas) >= 2: + rng = random.Random(20260904) + boot = sorted(statistics.mean(rng.choices(deltas, k=len(deltas))) for _ in range(5000)) + interval = [boot[125], boot[4874]] + dimensions[dimension] = dict( + n_scenarios=len(group), baseline=statistics.mean(row["baseline"] for row in group), + candidate=statistics.mean(row["candidate"] for row in group), + mean_delta=statistics.mean(deltas), ci95=interval, + wins=sum(d > 0 for d in deltas), ties=sum(d == 0 for d in deltas), + losses=sum(d < 0 for d in deltas)) + errors = lambda reports: sum(row["detail"].startswith("error:") + for report in reports for row in report["scenarios"]) + paired_keys = {row["identity"] for row in rows} + measurements = {"baseline": measurement_summary(baseline, paired_keys), "candidate": measurement_summary(candidate, paired_keys)} + return dict(measurement_summary=measurements, by_dimension=dimensions, scenarios=rows, unpaired_scenarios=unpaired, + unpaired_details=unpaired_details, + baseline_errors=errors(baseline), candidate_errors=errors(candidate), + repeats=len(baseline), + uncertainty_note="Exploratory paired bootstrap over scenario IDs after averaging repeats; correlated task families require a separate grouped holdout.") + + +def fingerprint(path): + path = Path(path).resolve(strict=True) + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024*1024), b""): + digest.update(chunk) + return dict(path=str(path), sha256=digest.hexdigest(), bytes=path.stat().st_size) + + +def dataset_fingerprints(path): + path = Path(path).resolve(strict=True) + data = json.loads(path.read_text(encoding="utf-8")) + sources = list(data.get("eval_fixtures", [])) + list(data.get("workflow_gen", [])) + if data.get("calibration_gen"): + sources.append(data["calibration_gen"]) + files = {path} + for source in sources: + if source.get("source"): + files.add((path.parent / source["source"]).resolve(strict=True)) + elif source.get("path"): + files.add((path.parent / source["path"]).resolve(strict=True)) + return [fingerprint(p) for p in sorted(files)] + + +def markdown(result): + lines = ["# Paired BrainBench comparison", "", + "Same harness and fixture; run order alternates. Positive delta favors the candidate.", "", + "| Dimension | Scenarios | Baseline | Candidate | Delta | Exploratory 95% interval |", + "|---|---:|---:|---:|---:|---|"] + for name, row in result["comparison"]["by_dimension"].items(): + ci = row["ci95"] + interval = "n/a" if ci is None else f"[{ci[0]:+.3f}, {ci[1]:+.3f}]" + lines.append(f"| {name} | {row['n_scenarios']} | {row['baseline']:.3f} | {row['candidate']:.3f} | {row['mean_delta']:+.3f} | {interval} |") + compare = result["comparison"] + lines += ["", f"Errors: baseline {compare['baseline_errors']}, candidate {compare['candidate_errors']}.", + f"Unpaired/skipped scenarios: {len(compare['unpaired_scenarios'])}.", "", + compare["uncertainty_note"], "", + "Wall times include process/model startup, corpus seeding and queries; they are not warm inference latency.", ""] + for label in ["baseline", "candidate"]: + values = [run["wall_seconds"] for run in result["runs"] if run["label"] == label] + lines.append(f"{label}: mean complete-run time {statistics.mean(values):.2f} s ({len(values)} repeats).") + if any(compare["measurement_summary"].values()): + lines += ["", "Query measurements through persistent MCP (subsequent queries reuse the process):", "", + "| Build | Positive hit@4 | Positive recall@4 | False injection | Subsequent p50 / p95 ms | Mean MCP result bytes | Peak MCP working set MiB |", + "|---|---:|---:|---:|---:|---:|---:|"] + def fmt(value): + return "n/a" if value is None else f"{value:.3f}" + for label, summary in compare["measurement_summary"].items(): + if summary is not None: + peak = summary['max_mcp_peak_working_set_bytes'] + lines.append(f"| {label} | {fmt(summary['positive_hit_at_4'])} | {fmt(summary['positive_recall_at_4'])} | {fmt(summary['negative_injection_rate'])} | {fmt(summary['subsequent_query_p50_ms'])} / {fmt(summary['subsequent_query_p95_ms'])} | {fmt(summary['mean_mcp_result_bytes'])} | {fmt(peak / 1048576 if peak is not None else None)} |") + lines.append("\nMeasured bytes include JSON escaping; reported token estimates are retained per query but may use different accounting rules across builds. Query timing excludes the separately recorded MCP initialization and corpus seeding.") + return "\n".join(lines) + "\n" + + +def text_output(value): + if value is None: + return "" + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return value + + +def persist_result(path, result): + path.write_text(json.dumps(result, indent=2), encoding="utf-8") + + +def environment_for_side(base, threads, reranker=None, rerank_floor=None, explicit_fact_guard=None): + result = dict(base) + if threads == 0: + result.pop("KIMETSU_INTRA_THREADS", None) + elif threads is not None: + result["KIMETSU_INTRA_THREADS"] = str(threads) + if reranker is not None: + result["KBENCH_RERANKER"] = reranker + if rerank_floor is not None: + if not 0 <= rerank_floor <= 1: + raise ValueError("rerank floor must be finite and between 0 and 1") + result["KBENCH_RERANK_FLOOR"] = str(rerank_floor) + if explicit_fact_guard is not None: + if explicit_fact_guard not in ("true", "false"): + raise ValueError("explicit fact guard must be true or false") + result["KBENCH_EXPLICIT_FACT_GUARD"] = explicit_fact_guard + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + for name in ["kbench", "baseline", "candidate", "dataset", "out"]: + parser.add_argument(f"--{name}", type=Path, required=True) + parser.add_argument("--repeats", type=int, default=3) + parser.add_argument("--budget-tokens", type=int, default=512) + parser.add_argument("--dimensions", default="retrieval,workflow,render-contract,poisoning") + parser.add_argument("--timeout-seconds", type=int, default=1800) + parser.add_argument("--baseline-threads", type=int, help="0 unsets the override; omitted inherits environment") + parser.add_argument("--candidate-threads", type=int, help="0 unsets the override; omitted inherits environment") + parser.add_argument("--baseline-reranker", help="Override reranker only in baseline temporary projects") + parser.add_argument("--candidate-reranker", help="Override reranker only in candidate temporary projects") + parser.add_argument("--baseline-rerank-floor", type=float) + parser.add_argument("--candidate-rerank-floor", type=float) + parser.add_argument("--baseline-explicit-fact-guard", choices=["true", "false"]) + parser.add_argument("--candidate-explicit-fact-guard", choices=["true", "false"]) + args = parser.parse_args() + if any(v is not None and not 0 <= v <= 1 for v in [args.baseline_rerank_floor, args.candidate_rerank_floor]): + parser.error("rerank floors must be finite and between 0 and 1") + dimensions = set(args.dimensions.split(",")) + if not dimensions or not dimensions <= OFFLINE_DIMENSIONS: + parser.error("only reader-free, non-generative dimensions are supported") + if args.repeats < 1 or args.budget_tokens < 1 or args.timeout_seconds < 1: + parser.error("repeats, budget and timeout must be positive") + if any(value is not None and not 0 <= value <= 1024 for value in [args.baseline_threads, args.candidate_threads]): + parser.error("thread overrides must be between 0 (unset) and 1024") + binaries = {label: getattr(args, label).resolve(strict=True) for label in ["baseline", "candidate"]} + args.out.mkdir(parents=True, exist_ok=True) + env = dict(os.environ, KIMETSU_USER_BRAIN="0") + # Record only relevant non-secret overrides, never the full environment. + overrides = {key: env[key] for key in ["KIMETSU_BRAIN_EMBEDDER", "KIMETSU_ABSTAIN_EVIDENCE", + "KIMETSU_DETECT_CONFLICTS", "KIMETSU_RESOLVE_CONFLICTS", "KIMETSU_INTRA_THREADS", + "FASTEMBED_CACHE_DIR", "HF_HOME", "KBENCH_RERANKER", "KBENCH_RERANK_FLOOR", "KBENCH_EXPLICIT_FACT_GUARD"] if key in env} + result = dict(schema_version=1, status="running", harness=fingerprint(args.kbench), runner=fingerprint(Path(__file__)), + binaries={k: fingerprint(v) for k,v in binaries.items()}, + datasets=dataset_fingerprints(args.dataset), + settings=dict(budget_tokens=args.budget_tokens, dimensions=sorted(dimensions), + jobs=1, warm_start=False, include_ambient=False, overrides=overrides, + baseline_threads=args.baseline_threads, candidate_threads=args.candidate_threads, + baseline_reranker=args.baseline_reranker, candidate_reranker=args.candidate_reranker, + baseline_rerank_floor=args.baseline_rerank_floor, candidate_rerank_floor=args.candidate_rerank_floor), runs=[]) + reports = {"baseline": [], "candidate": []} + for repeat in range(args.repeats): + for label in (["baseline", "candidate"] if repeat % 2 == 0 else ["candidate", "baseline"]): + cmd = [str(args.kbench.resolve()), "brainbench", "--dataset", str(args.dataset.resolve()), + "--kimetsu-binary", str(binaries[label]), "--budget-tokens", str(args.budget_tokens), + "--dimensions", ",".join(sorted(dimensions)), "--jobs", "1", "--output", "json"] + print(f"repeat {repeat+1}/{args.repeats}: {label}", flush=True) + start = time.perf_counter() + stem = args.out / f"{repeat+1}-{label}" + run_record = dict(label=label, repeat=repeat+1) + run_env = environment_for_side(env, getattr(args, f"{label}_threads"), getattr(args, f"{label}_reranker"), getattr(args, f"{label}_rerank_floor"), getattr(args, f"{label}_explicit_fact_guard")) + run_record["intra_threads_override"] = run_env.get("KIMETSU_INTRA_THREADS") + run_record["rerank_floor_override"] = run_env.get("KBENCH_RERANK_FLOOR") + run_record["explicit_fact_guard_override"] = run_env.get("KBENCH_EXPLICIT_FACT_GUARD") + run_record["reranker_override"] = run_env.get("KBENCH_RERANKER") + try: + completed = run_owned_tree(cmd, env=run_env, timeout=args.timeout_seconds) + except subprocess.TimeoutExpired as error: + run_record["wall_seconds"] = time.perf_counter() - start + run_record["failure"] = dict(kind="timeout", timeout_seconds=args.timeout_seconds, + message=str(error)) + stem.with_suffix(".stdout.log").write_text(text_output(error.output), encoding="utf-8") + stem.with_suffix(".stderr.log").write_text(text_output(error.stderr), encoding="utf-8") + result["runs"].append(run_record) + result["status"] = "incomplete" + persist_result(args.out / "comparison.json", result) + return 1 + elapsed = time.perf_counter() - start + stem.with_suffix(".stdout.log").write_text(completed.stdout, encoding="utf-8") + stem.with_suffix(".stderr.log").write_text(completed.stderr, encoding="utf-8") + run_record["wall_seconds"] = elapsed + if completed.returncode: + run_record["failure"] = dict(kind="nonzero_exit", + returncode=completed.returncode, + message=f"{label} exited {completed.returncode}") + result["runs"].append(run_record) + result["status"] = "incomplete" + persist_result(args.out / "comparison.json", result) + return 1 + try: + report = json.loads(completed.stdout) + except json.JSONDecodeError as error: + run_record["failure"] = dict(kind="invalid_json", message=str(error)) + result["runs"].append(run_record) + result["status"] = "incomplete" + persist_result(args.out / "comparison.json", result) + return 1 + report_path = stem.with_suffix(".json") + report_path.write_text(json.dumps(report, indent=2), encoding="utf-8") + run_record["report_file"] = report_path.name + try: + validate_report(report) + except ValueError as error: + run_record["failure"] = dict(kind="invalid_report", message=str(error)) + result["runs"].append(run_record) + result["status"] = "incomplete" + persist_result(args.out / "comparison.json", result) + return 1 + reports[label].append(report) + result["runs"].append(run_record) + persist_result(args.out / "comparison.json", result) + try: + result["comparison"] = compare_reports(reports["baseline"], reports["candidate"]) + except ValueError as error: + result["status"] = "incomplete" + result["failure"] = dict(kind="comparison_validation", message=str(error)) + persist_result(args.out / "comparison.json", result) + return 1 + result["status"] = "complete" + persist_result(args.out / "comparison.json", result) + (args.out / "comparison.md").write_text(markdown(result), encoding="utf-8") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/test_compare_brainbench.py b/scripts/test_compare_brainbench.py new file mode 100644 index 0000000..c616a81 --- /dev/null +++ b/scripts/test_compare_brainbench.py @@ -0,0 +1,218 @@ +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +import compare_brainbench +from compare_brainbench import compare_reports + + +class ProcessOwnershipTests(unittest.TestCase): + def test_timeout_terminates_descendants_before_parent_exit(self): + import os + import time + with tempfile.TemporaryDirectory() as folder: + marker = Path(folder) / "orphan-ran" + child = "import time,pathlib; time.sleep(2); pathlib.Path(%r).write_text('orphan')" % str(marker) + parent = "import subprocess,sys,time; subprocess.Popen([sys.executable,'-c',%r]); print('ready',flush=True); time.sleep(30)" % child + start = time.monotonic() + with self.assertRaises(subprocess.TimeoutExpired): + compare_brainbench.run_owned_tree([sys.executable, "-c", parent], env=dict(os.environ), timeout=.5) + elapsed = time.monotonic() - start + time.sleep(max(0, 2.4 - elapsed)) + self.assertFalse(marker.exists(), "timed-out inference descendant survived") + + +def report(rows): + return {"session_configuration":{"warm_start":False,"include_ambient":False}, "scenarios": [dict(id=key, dimension=dim, score=value, + skipped=False, detail="ok") for key, dim, value in rows]} + + +class PairedComparisonTests(unittest.TestCase): + def test_guard_override_is_explicit_and_isolated(self): + base = {"KBENCH_EXPLICIT_FACT_GUARD": "true"} + self.assertEqual(compare_brainbench.environment_for_side(base, None, explicit_fact_guard="false")["KBENCH_EXPLICIT_FACT_GUARD"], "false") + self.assertEqual(base["KBENCH_EXPLICIT_FACT_GUARD"], "true") + with self.assertRaises(ValueError): + compare_brainbench.environment_for_side(base, None, explicit_fact_guard="maybe") + + def test_pairing_rejects_missing_or_false_isolation_provenance(self): + for configuration in [None, {}, {"warm_start":True,"include_ambient":False}, + {"warm_start":0,"include_ambient":False}]: + sample = report([("a", "retrieval", 1)]) + sample["session_configuration"] = configuration + with self.assertRaisesRegex(ValueError, "session_configuration"): + compare_reports([sample], [report([("a", "retrieval", 1)])]) + + def test_memory_summary_reports_unavailable_and_observed_peak_separately(self): + base = report([("a", "retrieval", 1)]) + base["scenarios"][0]["observations"] = [ + dict(query="a", positive_recall_at_4=1, positive_hit_at_4=True, + positive_mrr=1, negative_injection=None, stale_injection=None, + latency_ms=1, first_query=True, model_text_bytes=80, mcp_result_bytes=100, + working_set_bytes=1000, peak_working_set_bytes=2000), + ] + summary = compare_reports([base], [base])["measurement_summary"]["baseline"] + self.assertEqual(summary["max_mcp_peak_working_set_bytes"], 2000) + self.assertEqual(summary["memory_observations"], 1) + base["scenarios"][0]["observations"][0]["working_set_bytes"] = None + base["scenarios"][0]["observations"][0]["peak_working_set_bytes"] = None + summary = compare_reports([base], [base])["measurement_summary"]["baseline"] + self.assertIsNone(summary["max_mcp_peak_working_set_bytes"]) + self.assertEqual(summary["memory_observations"], 0) + + def test_side_thread_overrides_preserve_base_and_allow_unset(self): + base = {"KIMETSU_USER_BRAIN":"0", "KIMETSU_INTRA_THREADS":"8"} + self.assertNotIn("KIMETSU_INTRA_THREADS", compare_brainbench.environment_for_side(base, 0)) + self.assertEqual(compare_brainbench.environment_for_side(base, 4)["KIMETSU_INTRA_THREADS"], "4") + self.assertEqual(compare_brainbench.environment_for_side(base, None)["KIMETSU_INTRA_THREADS"], "8") + self.assertEqual(base["KIMETSU_INTRA_THREADS"], "8") + + def test_side_model_overrides_allow_a_paired_reranker_comparison(self): + base = {"KBENCH_RERANKER":"ms-marco-tinybert-l-2-v2"} + self.assertEqual(compare_brainbench.environment_for_side(base, None, "off")["KBENCH_RERANKER"], "off") + self.assertEqual(compare_brainbench.environment_for_side(base, 4, "ms-marco-minilm-l-4-v2")["KBENCH_RERANKER"], "ms-marco-minilm-l-4-v2") + self.assertEqual(compare_brainbench.environment_for_side(base, None)["KBENCH_RERANKER"], base["KBENCH_RERANKER"]) + + def test_side_cutoff_isolated_and_validated(self): + base = {"KBENCH_RERANK_FLOOR": "0.3"} + self.assertEqual(compare_brainbench.environment_for_side(base, None, None, .9)["KBENCH_RERANK_FLOOR"], "0.9") + self.assertEqual(base["KBENCH_RERANK_FLOOR"], "0.3") + for invalid in [-.1, 1.1, float("nan"), float("inf")]: + with self.assertRaises(ValueError): + compare_brainbench.environment_for_side(base, None, None, invalid) + + def test_query_measurements_separate_recall_hit_and_first_query_latency(self): + base = report([("a", "retrieval", .5)]) + base["scenarios"][0]["observations"] = [ + dict(query="a", positive_recall_at_4=.5, positive_hit_at_4=True, + positive_mrr=1, negative_injection=None, stale_injection=None, + latency_ms=100, first_query=True, model_text_bytes=80, mcp_result_bytes=100), + dict(query="b", positive_recall_at_4=1, positive_hit_at_4=True, + positive_mrr=1, negative_injection=None, stale_injection=None, + latency_ms=20, first_query=False, model_text_bytes=180, mcp_result_bytes=200), + dict(query="negative", positive_recall_at_4=None, positive_hit_at_4=None, + positive_mrr=None, negative_injection=False, stale_injection=None, + latency_ms=40, first_query=False, model_text_bytes=280, mcp_result_bytes=300), + ] + summary = compare_reports([base, base], [base, base])["measurement_summary"]["baseline"] + self.assertEqual(summary["unique_queries"], 3) + self.assertEqual(summary["positive_queries"], 2) + self.assertEqual(summary["query_observations"], 6) + self.assertEqual(summary["positive_recall_at_4"], .75) + self.assertEqual(summary["positive_hit_at_4"], 1) + self.assertEqual(summary["negative_injection_rate"], 0) + self.assertEqual(summary["subsequent_query_p95_ms"], 40) + self.assertEqual(summary["mean_mcp_result_bytes"], 200) + + def test_repeats_are_not_independent_scenarios(self): + base = report([("a", "retrieval", 0), ("b", "retrieval", 1)]) + candidate = report([("b", "retrieval", 1), ("a", "retrieval", 1)]) + result = compare_reports([base, base, base], [candidate, candidate, candidate]) + self.assertEqual(result["by_dimension"]["retrieval"]["n_scenarios"], 2) + self.assertEqual(result["by_dimension"]["retrieval"]["mean_delta"], .5) + + def test_changed_scenario_set_is_rejected(self): + with self.assertRaises(ValueError): + compare_reports([report([("a", "retrieval", 0)])], + [report([("b", "retrieval", 1)])]) + + def test_failed_observation_is_unpaired_and_cannot_create_a_gain(self): + base = report([("a", "retrieval", 0), ("b", "retrieval", 0)]) + base["scenarios"][0]["detail"] = "error: process exited" + candidate = report([("a", "retrieval", 1), ("b", "retrieval", 1)]) + candidate["scenarios"][1]["skipped"] = True + result = compare_reports([base], [candidate]) + self.assertEqual(result["baseline_errors"], 1) + self.assertEqual(result["unpaired_scenarios"], ["retrieval/a", "retrieval/b"]) + self.assertEqual(result["scenarios"], []) + self.assertEqual(result["by_dimension"], {}) + + def test_duplicate_identity_is_rejected(self): + repeated = report([("a", "retrieval", 0), ("a", "retrieval", 1)]) + with self.assertRaises(ValueError): + compare_reports([repeated], [repeated]) + + +class RunnerFailureTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) + for name in ["kbench.exe", "baseline.exe", "candidate.exe"]: + (self.root / name).write_bytes(name.encode()) + (self.root / "dataset.json").write_text('{"scenarios": []}', encoding="utf-8") + + def tearDown(self): + self.temp.cleanup() + + def argv(self): + return ["compare_brainbench.py", + "--kbench", str(self.root / "kbench.exe"), + "--baseline", str(self.root / "baseline.exe"), + "--candidate", str(self.root / "candidate.exe"), + "--dataset", str(self.root / "dataset.json"), + "--out", str(self.root / "out"), "--repeats", "1"] + + def assert_incomplete_failure(self, side_effect, kind): + run_patch = (mock.patch.object(compare_brainbench, "run_owned_tree", side_effect=side_effect) + if isinstance(side_effect, BaseException) + else mock.patch.object(compare_brainbench, "run_owned_tree", return_value=side_effect)) + with mock.patch.object(sys, "argv", self.argv()), run_patch: + rc = compare_brainbench.main() + artifact = json.loads((self.root / "out" / "comparison.json").read_text(encoding="utf-8")) + self.assertEqual(rc, 1) + self.assertEqual(artifact["status"], "incomplete") + self.assertNotIn("comparison", artifact) + self.assertEqual(artifact["runs"][-1]["failure"]["kind"], kind) + + def test_nonzero_exit_is_persisted_before_runner_fails(self): + completed = subprocess.CompletedProcess([], 7, stdout="partial", stderr="boom") + self.assert_incomplete_failure(completed, "nonzero_exit") + + def test_timeout_is_persisted_before_runner_fails(self): + timeout = subprocess.TimeoutExpired(["kbench"], 1800, output="partial", stderr="slow") + self.assert_incomplete_failure(timeout, "timeout") + + def test_invalid_json_is_persisted_before_runner_fails(self): + completed = subprocess.CompletedProcess([], 0, stdout="not json", stderr="warning") + self.assert_incomplete_failure(completed, "invalid_json") + + def test_valid_json_with_wrong_report_shape_is_persisted_as_failure(self): + for payload in ["null", "{}", '{"scenarios": [{}]}']: + with self.subTest(payload=payload): + completed = subprocess.CompletedProcess([], 0, stdout=payload, stderr="") + self.assert_incomplete_failure(completed, "invalid_report") + + def test_changed_scenario_identity_finalizes_as_incomplete(self): + baseline = subprocess.CompletedProcess( + [], 0, stdout=json.dumps(report([("a", "retrieval", 1)])), stderr="") + candidate = subprocess.CompletedProcess( + [], 0, stdout=json.dumps(report([("b", "retrieval", 1)])), stderr="") + with mock.patch.object(sys, "argv", self.argv()), \ + mock.patch.object(compare_brainbench, "run_owned_tree", side_effect=[baseline, candidate]): + rc = compare_brainbench.main() + artifact = json.loads((self.root / "out" / "comparison.json").read_text(encoding="utf-8")) + self.assertEqual(rc, 1) + self.assertEqual(artifact["status"], "incomplete") + self.assertEqual(artifact["failure"]["kind"], "comparison_validation") + self.assertNotIn("comparison", artifact) + + def test_duplicate_scenario_identity_finalizes_as_incomplete(self): + duplicate = subprocess.CompletedProcess( + [], 0, stdout=json.dumps(report([ + ("a", "retrieval", 0), ("a", "retrieval", 1) + ])), stderr="") + with mock.patch.object(sys, "argv", self.argv()), \ + mock.patch.object(compare_brainbench, "run_owned_tree", return_value=duplicate): + rc = compare_brainbench.main() + artifact = json.loads((self.root / "out" / "comparison.json").read_text(encoding="utf-8")) + self.assertEqual(rc, 1) + self.assertEqual(artifact["status"], "incomplete") + self.assertEqual(artifact["failure"]["kind"], "comparison_validation") + + +if __name__ == "__main__": + unittest.main() diff --git a/src/drivers/brain_mcp.rs b/src/drivers/brain_mcp.rs new file mode 100644 index 0000000..fa3bd8e --- /dev/null +++ b/src/drivers/brain_mcp.rs @@ -0,0 +1,247 @@ +//! Persistent production MCP client for reader-free measurement. +use serde_json::Value; +use std::io::{BufRead, BufReader, Write}; +use std::path::Path; +use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; +use std::time::Instant; + +fn decode_tool_result(response: &Value, id: u64) -> Result<(Value, usize, usize), String> { + if response.get("id").and_then(Value::as_u64) != Some(id) { + return Err("MCP response ID mismatch".into()); + } + if let Some(error) = response.get("error") { + return Err(format!("MCP error: {error}")); + } + let result = response.get("result").ok_or("MCP result missing")?; + if result.get("isError").and_then(Value::as_bool) == Some(true) { + return Err(format!("MCP tool failed: {result}")); + } + let blocks = result + .get("content") + .and_then(Value::as_array) + .ok_or("MCP content missing")?; + if blocks.len() != 1 || blocks[0].get("type").and_then(Value::as_str) != Some("text") { + return Err("expected one MCP text content block".into()); + } + let text = blocks[0] + .get("text") + .and_then(Value::as_str) + .ok_or("MCP text missing")?; + let payload: Value = + serde_json::from_str(text).map_err(|e| format!("MCP text is not JSON: {e}"))?; + if !payload.is_object() { + return Err("MCP payload must be an object".into()); + } + if payload.get("ok").and_then(Value::as_bool) == Some(false) { + return Err(format!("context tool rejected request: {payload}")); + } + Ok((payload, text.len(), result.to_string().len())) +} + +pub(super) struct McpMeasurement { + pub payload: Value, + pub text_bytes: usize, + pub result_bytes: usize, + pub wire_bytes: usize, + pub latency_ms: f64, + pub first_query: bool, + pub server_startup_ms: f64, + pub working_set_bytes: Option, + pub peak_working_set_bytes: Option, +} + +#[cfg(windows)] +fn process_memory(handle: windows_sys::Win32::Foundation::HANDLE) -> Option<(u64, u64)> { + use windows_sys::Win32::System::ProcessStatus::{ + K32GetProcessMemoryInfo, PROCESS_MEMORY_COUNTERS, + }; + let mut counters: PROCESS_MEMORY_COUNTERS = unsafe { std::mem::zeroed() }; + counters.cb = std::mem::size_of::() as u32; + // The caller retains the live process handle. The initialized C-layout + // output buffer and cb match the Windows API's required structure size. + if unsafe { K32GetProcessMemoryInfo(handle, &mut counters, counters.cb) } == 0 { + None + } else { + Some(( + counters.WorkingSetSize as u64, + counters.PeakWorkingSetSize as u64, + )) + } +} + +fn child_memory(child: &Child) -> Option<(u64, u64)> { + #[cfg(windows)] + { + use std::os::windows::io::AsRawHandle; + process_memory(child.as_raw_handle()) + } + #[cfg(not(windows))] + { + let _ = child; + None // Unavailable is not zero; platform-specific collectors can extend this. + } +} + +pub(super) struct BrainMcp { + child: Child, + stdin: Option, + stdout: BufReader, + next_id: u64, + queries: usize, + startup_ms: f64, +} + +impl BrainMcp { + pub fn start(workspace: &Path, binary: &str) -> Result { + let start = Instant::now(); + let mut child = Command::new(binary) + .current_dir(workspace) + .env("KIMETSU_USER_BRAIN", "0") + .args(["mcp", "serve", "--workspace"]) + .arg(workspace) + .arg("--no-user-skills") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .map_err(|e| format!("start MCP: {e}"))?; + let stdin = child.stdin.take().ok_or("MCP stdin unavailable")?; + let stdout = child.stdout.take().ok_or("MCP stdout unavailable")?; + let mut client = Self { + child, + stdin: Some(stdin), + stdout: BufReader::new(stdout), + next_id: 1, + queries: 0, + startup_ms: 0.0, + }; + let response = client + .request( + "initialize", + serde_json::json!({ + "protocolVersion":"2024-11-05", "capabilities":{}, + "clientInfo":{"name":"BrainBenchmark","version":"1"} + }), + )? + .0; + if response.get("error").is_some() { + return Err(format!("MCP initialize failed: {response}")); + } + client.send(&serde_json::json!({"jsonrpc":"2.0","method":"notifications/initialized"}))?; + client.startup_ms = start.elapsed().as_secs_f64() * 1000.0; + Ok(client) + } + + fn send(&mut self, value: &Value) -> Result<(), String> { + let stream = self.stdin.as_mut().ok_or("MCP stdin closed")?; + writeln!(stream, "{value}") + .and_then(|_| stream.flush()) + .map_err(|e| format!("MCP write: {e}")) + } + + fn request(&mut self, method: &str, params: Value) -> Result<(Value, usize), String> { + let id = self.next_id; + self.next_id += 1; + self.send(&serde_json::json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}))?; + loop { + let mut line = String::new(); + if self + .stdout + .read_line(&mut line) + .map_err(|e| format!("MCP read: {e}"))? + == 0 + { + return Err("MCP exited before responding".into()); + } + let value: Value = + serde_json::from_str(&line).map_err(|e| format!("MCP stdout is not JSON: {e}"))?; + if value.get("id").is_none() { + continue; + } // server notification + if value.get("id").and_then(Value::as_u64) != Some(id) { + return Err("MCP response ID mismatch".into()); + } + return Ok((value, line.len())); + } + } + + pub fn context(&mut self, query: &str, budget: usize) -> Result { + let start = Instant::now(); + let id = self.next_id; + let (response, wire_bytes) = self.request( + "tools/call", + serde_json::json!({ + "name":"kimetsu_brain_context", "arguments":{ + "query":query, "budget_tokens":budget, "include_ambient":false, + "max_capsules":4 + } + }), + )?; + let latency_ms = start.elapsed().as_secs_f64() * 1000.0; + let (payload, text_bytes, result_bytes) = decode_tool_result(&response, id)?; + let first_query = self.queries == 0; + self.queries += 1; + // Sample only the MCP child, after stopping the request latency clock. + let memory = child_memory(&self.child); + Ok(McpMeasurement { + payload, + text_bytes, + result_bytes, + wire_bytes, + latency_ms, + first_query, + server_startup_ms: if first_query { self.startup_ms } else { 0.0 }, + working_set_bytes: memory.map(|m| m.0), + peak_working_set_bytes: memory.map(|m| m.1), + }) + } +} + +impl Drop for BrainMcp { + fn drop(&mut self) { + self.stdin.take(); + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(windows)] + #[test] + fn windows_process_memory_reports_live_working_set_and_peak() { + use windows_sys::Win32::System::Threading::GetCurrentProcess; + let (current, peak) = process_memory(unsafe { GetCurrentProcess() }).unwrap(); + assert!(current > 0); + assert!(peak >= current); + } + + #[test] + fn measures_serialized_result_and_utf8_text_without_losing_escapes() { + let payload = + serde_json::json!({"capsules":[{"summary":"配置\n\"quoted\""}],"used_tokens":12}); + let text = payload.to_string(); + let result = serde_json::json!({"content":[{"type":"text","text":text}]}); + let response = serde_json::json!({"jsonrpc":"2.0","id":2,"result":result}); + let (decoded, text_bytes, result_bytes) = decode_tool_result(&response, 2).unwrap(); + assert_eq!(decoded, payload); + assert_eq!(text_bytes, text.len()); + assert_eq!(result_bytes, result.to_string().len()); + assert!(result_bytes > text_bytes); + } + + #[test] + fn malformed_or_failed_protocol_output_is_not_abstention() { + for response in [ + serde_json::json!({"id":9,"result":{"content":[{"type":"text","text":"{}"}]}}), + serde_json::json!({"id":2,"error":{"message":"failed"}}), + serde_json::json!({"id":2,"result":{"isError":true,"content":[]}}), + serde_json::json!({"id":2,"result":{"content":[]}}), + serde_json::json!({"id":2,"result":{"content":[{"type":"text","text":"null"}]}}), + ] { + assert!(decode_tool_result(&response, 2).is_err()); + } + } +} diff --git a/src/drivers/brainbench.rs b/src/drivers/brainbench.rs index 7571a66..d22b6be 100644 --- a/src/drivers/brainbench.rs +++ b/src/drivers/brainbench.rs @@ -191,6 +191,10 @@ pub struct Memory { pub scope: String, #[serde(default = "default_kind")] pub kind: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub valid_from: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub valid_to: Option, } /// A retrieval/importance probe against an ingested scenario. @@ -447,11 +451,15 @@ pub struct EvalFixtureFile { pub cases: Vec, } -/// One memory in an EvalFixture file. Extra fields (e.g. `valid_to`) are ignored. +/// One memory in an EvalFixture file. Temporal applicability is retained at ingest. #[derive(Debug, Clone, Deserialize, Serialize)] pub struct EvalFixMemory { pub key: String, pub text: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub valid_from: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub valid_to: Option, } /// One gold-labeled retrieval case in an EvalFixture file. @@ -609,8 +617,38 @@ pub struct RenderContractSpec { pub must_not_contain: Vec, } +/// One query through the production MCP surface. The first query may load the +/// model; subsequent queries reuse the server. OS/model disk cache is not reset. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct QueryObservation { + pub query: String, + pub ranked: Vec, + /// Delivered evidence, retained for auditing text matching and compression. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub delivered_capsules: Vec, + /// Final evidence accounting as actually delivered, absent on older binaries. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub answerability: Option, + pub positive_recall_at_4: Option, + pub positive_hit_at_4: Option, + pub positive_mrr: Option, + pub negative_injection: Option, + pub stale_injection: Option, + pub latency_ms: f64, + pub first_query: bool, + pub server_startup_ms: f64, + pub model_text_bytes: usize, + pub mcp_result_bytes: usize, + pub wire_bytes: usize, + pub reported_used_tokens: Option, + pub working_set_bytes: Option, + pub peak_working_set_bytes: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ScenarioResult { + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub observations: Vec, pub id: String, pub dimension: Dimension, pub tier: Tier, @@ -639,6 +677,9 @@ pub struct BrainBenchReport { pub generated_at: String, /// Path to the dataset file. pub dataset: String, + /// Controlled setup used by this harness; absent on historical reports. + #[serde(default)] + pub session_configuration: Option, /// Per-scenario results (full detail). pub scenarios: Vec, /// "dimension/tier" -> (sum_of_scores, count). Skipped scenarios excluded. @@ -646,8 +687,11 @@ pub struct BrainBenchReport { /// Per-dimension mean + n + 95% CI (skipped scenarios excluded). #[serde(default)] pub by_dimension: BTreeMap, - /// Mean score over NON-skipped scenarios (0.0–1.0). + /// Equal-weight mean over measured dimensions (0.0–1.0). pub overall_index: f64, + /// Historical scenario-weighted mean, retained as a diagnostic only. + #[serde(default)] + pub scenario_weighted_index: f64, } impl BrainBenchReport { @@ -665,6 +709,10 @@ impl BrainBenchReport { total, skipped )); + out.push_str(&format!( + "The headline weights measured dimensions equally. Scenario-weighted diagnostic: {:.1}%. Neither score estimates agent task success.\n\n", + self.scenario_weighted_index * 100.0 + )); out.push_str("## By dimension (n, 95% CI)\n\n"); out.push_str("| dimension | score | n | 95% CI |\n"); @@ -869,7 +917,7 @@ pub fn score_workflow_episode( (None, _) => false, }; let r = recall_at_k(ranked, relevant, k); - let resolution_ok = stale.is_empty() || resolution_correct(ranked, relevant, stale); + let resolution_ok = stale_hit_rate(ranked, stale, k) == 0.0; let score = if trap_hit || !resolution_ok { 0.0 } else { r }; EpisodeScore { score, @@ -957,6 +1005,8 @@ pub fn expand_eval_fixtures( .memories .iter() .map(|m| Memory { + valid_from: m.valid_from.clone(), + valid_to: m.valid_to.clone(), key: m.key.clone(), text: m.text.clone(), scope: default_scope(), @@ -1084,18 +1134,24 @@ pub fn expand_calibration_gen( ), memories: vec![ Memory { + valid_from: None, + valid_to: None, key: "good".to_string(), text: pool[gi].text.clone(), scope: default_scope(), kind: default_kind(), }, Memory { + valid_from: None, + valid_to: None, key: "neutral".to_string(), text: pool[ni].text.clone(), scope: default_scope(), kind: default_kind(), }, Memory { + valid_from: None, + valid_to: None, key: "bad".to_string(), text: pool[bi].text.clone(), scope: default_scope(), @@ -1167,6 +1223,11 @@ pub fn expand_workflow_gen( .iter() .map(|m| (m.key.clone(), m.text.clone())) .collect(); + let mem_validity_by_key: std::collections::HashMap<_, _> = fixture + .memories + .iter() + .map(|m| (m.key.clone(), (m.valid_from.clone(), m.valid_to.clone()))) + .collect(); let resolvable = |keys: &[String]| keys.iter().all(|k| mem_text_by_key.contains_key(k)); // Plain cases: at least one relevant key, all keys resolvable. Update @@ -1270,6 +1331,8 @@ pub fn expand_workflow_gen( keys.iter() .filter(|k| taken.insert((*k).clone())) .map(|k| Memory { + valid_from: mem_validity_by_key[k].0.clone(), + valid_to: mem_validity_by_key[k].1.clone(), key: k.clone(), text: mem_text_by_key[k].clone(), scope: default_scope(), @@ -1723,6 +1786,8 @@ pub fn expand_workflow_gen( } taken.insert(m.key.clone()); seed.push(Memory { + valid_from: m.valid_from.clone(), + valid_to: m.valid_to.clone(), key: m.key.clone(), text: m.text.clone(), scope: default_scope(), @@ -1787,6 +1852,16 @@ fn resolve_kimetsu_bin(cfg: &BrainBenchConfig) -> String { /// /// EVERY kimetsu call sets `KIMETSU_USER_BRAIN=0` so the global cross-project /// brain cannot leak pre-existing memories into measurements. +fn brain_config_overrides(reranker: Option<&str>) -> Vec<(&str, &str)> { + // New projects select the deep preset, which overwrites concrete models + // during configuration loading. Leave preset mode before selecting one. + let mut settings = vec![("broker.warm_start", "false")]; + if let Some(model) = reranker { + settings.extend([("retrieval.level", "custom"), ("embedder.reranker", model)]); + } + settings +} + fn setup_brain(kimetsu_bin: &str) -> Result { let tmp = tempfile::Builder::new() .prefix("kbench-brain-") @@ -1821,6 +1896,82 @@ fn setup_brain(kimetsu_bin: &str) -> Result ))); } + let reranker = std::env::var("KBENCH_RERANKER").ok(); + let floor = std::env::var("KBENCH_RERANK_FLOOR").ok(); + let guard = std::env::var("KBENCH_EXPLICIT_FACT_GUARD").ok(); + let mut settings = brain_config_overrides(reranker.as_deref()); + if let Some(value) = guard.as_deref() { + if !matches!(value, "true" | "false") { + return Err(BrainBenchError::Other( + "KBENCH_EXPLICIT_FACT_GUARD must be true or false".into(), + )); + } + settings.push(("broker.explicit_fact_guard", value)); + } + if let Some(value) = floor.as_deref() { + let parsed = value + .parse::() + .map_err(|_| BrainBenchError::Other("invalid KBENCH_RERANK_FLOOR".into()))?; + if !parsed.is_finite() || !(0.0..=1.0).contains(&parsed) { + return Err(BrainBenchError::Other( + "KBENCH_RERANK_FLOOR must be finite and between 0 and 1".into(), + )); + } + settings.push(("broker.rerank_min_score", value)); + } + for (key, value) in settings { + let configured = Command::new(kimetsu_bin) + .current_dir(workspace) + .env("KIMETSU_USER_BRAIN", "0") + .args(["config", "set", key, value]) + .output() + .map_err(|e| BrainBenchError::KimetsuError(format!("set {key}: {e}")))?; + if !configured.status.success() { + return Err(BrainBenchError::KimetsuError(format!( + "set {key} failed: {}", + String::from_utf8_lossy(&configured.stderr) + ))); + } + } + + if let Some(expected) = guard.as_deref() { + let actual = Command::new(kimetsu_bin) + .current_dir(workspace) + .env("KIMETSU_USER_BRAIN", "0") + .args(["config", "get", "broker.explicit_fact_guard"]) + .output() + .map_err(|e| { + BrainBenchError::KimetsuError(format!("read effective fact guard: {e}")) + })?; + if !actual.status.success() || String::from_utf8_lossy(&actual.stdout).trim() != expected { + return Err(BrainBenchError::KimetsuError(format!( + "binary did not apply broker.explicit_fact_guard={expected}" + ))); + } + } + if let Some(expected) = floor.as_deref() { + // Older binaries accept unknown TOML keys but omit them from effective + // config. Never label that silent no-op as a measured threshold. + let actual = Command::new(kimetsu_bin) + .current_dir(workspace) + .env("KIMETSU_USER_BRAIN", "0") + .args(["config", "get", "broker.rerank_min_score"]) + .output() + .map_err(|e| { + BrainBenchError::KimetsuError(format!("read effective rerank floor: {e}")) + })?; + let parsed = String::from_utf8_lossy(&actual.stdout) + .trim() + .parse::() + .ok(); + if !actual.status.success() || parsed != expected.parse::().ok() { + return Err(BrainBenchError::KimetsuError(format!( + "binary did not apply broker.rerank_min_score={expected}: {} {}", + String::from_utf8_lossy(&actual.stdout), + String::from_utf8_lossy(&actual.stderr) + ))); + } + } Ok(tmp) } @@ -1834,6 +1985,8 @@ fn ingest(workspace: &Path, kimetsu_bin: &str, memories: &[Memory]) -> Result<() "text": m.text, "scope": m.scope, "kind": m.kind, + "valid_from": m.valid_from, + "valid_to": m.valid_to, }) .to_string() }) @@ -1883,13 +2036,8 @@ fn strip_prefix_summary(summary: &str) -> &str { } } -/// Retrieve the ranked fixture keys for a query. -/// -/// Runs `brain context "" --no-ambient --json --budget-tokens `, -/// parses the `capsules` array (already score-sorted), strips each capsule -/// `summary`'s prefix, normalizes it, and matches against the normalized text of -/// each fixture Memory to recover its `key`. Capsules that match no fixture -/// memory are dropped. Order is preserved. +/// A single query uses the same MCP surface as a host agent. Retrieval and +/// workflow runners instead keep one BrainMcp alive across all their queries. fn retrieve_ranked_keys( workspace: &Path, kimetsu_bin: &str, @@ -1897,39 +2045,62 @@ fn retrieve_ranked_keys( budget: usize, memories: &[Memory], ) -> Result, BrainBenchError> { - let budget_str = budget.to_string(); - let out = Command::new(kimetsu_bin) - .current_dir(workspace) - .env("KIMETSU_USER_BRAIN", "0") - .args([ - "brain", - "context", - query, - "--no-ambient", - "--json", - "--budget-tokens", - &budget_str, - ]) - .output() - .map_err(|e| { - BrainBenchError::KimetsuError(format!( - "could not spawn `{kimetsu_bin} brain context`: {e}" - )) - })?; - if !out.status.success() { - let stderr = String::from_utf8_lossy(&out.stderr); - return Err(BrainBenchError::KimetsuError(format!( - "brain context failed: {stderr}" - ))); - } + let mut client = super::brain_mcp::BrainMcp::start(workspace, kimetsu_bin) + .map_err(BrainBenchError::KimetsuError)?; + let measurement = client + .context(query, budget) + .map_err(BrainBenchError::KimetsuError)?; + ranked_fixture_keys(&measurement.payload, memories) +} - let stdout = String::from_utf8_lossy(&out.stdout); - let v: serde_json::Value = serde_json::from_str(&stdout).map_err(|e| { - BrainBenchError::KimetsuError(format!( - "brain context returned non-JSON output: {e}\n{stdout}" - )) - })?; +fn observe_query( + query: &str, + ranked: &[String], + relevant: &[String], + stale: &[String], + measurement: &super::brain_mcp::McpMeasurement, +) -> QueryObservation { + let positive = !relevant.is_empty(); + QueryObservation { + query: query.into(), + ranked: ranked.to_vec(), + answerability: measurement.payload.get("answerability").cloned(), + delivered_capsules: measurement + .payload + .get("capsules") + .and_then(serde_json::Value::as_array) + .cloned() + .unwrap_or_default(), + positive_recall_at_4: positive.then(|| recall_at_k(ranked, relevant, 4)), + positive_hit_at_4: positive.then(|| ranked.iter().take(4).any(|id| relevant.contains(id))), + positive_mrr: positive.then(|| mrr(ranked, relevant)), + negative_injection: (!positive).then_some(!ranked.is_empty()), + stale_injection: (!stale.is_empty()) + .then(|| ranked.iter().take(4).any(|id| stale.contains(id))), + latency_ms: measurement.latency_ms, + first_query: measurement.first_query, + server_startup_ms: measurement.server_startup_ms, + model_text_bytes: measurement.text_bytes, + mcp_result_bytes: measurement.result_bytes, + wire_bytes: measurement.wire_bytes, + working_set_bytes: measurement.working_set_bytes, + peak_working_set_bytes: measurement.peak_working_set_bytes, + reported_used_tokens: measurement + .payload + .get("used_tokens") + .and_then(serde_json::Value::as_u64), + } +} +fn ranked_fixture_keys( + v: &serde_json::Value, + memories: &[Memory], +) -> Result, BrainBenchError> { + if !v.get("capsules").is_some_and(serde_json::Value::is_array) { + return Err(BrainBenchError::KimetsuError( + "context response lacks a capsules array".into(), + )); + } // Precompute normalized fixture text -> key. let norm_to_key: Vec<(String, String)> = memories .iter() @@ -1938,25 +2109,60 @@ fn retrieve_ranked_keys( let mut ranked: Vec = Vec::new(); if let Some(capsules) = v.get("capsules").and_then(|c| c.as_array()) { - for cap in capsules { - let summary = cap.get("summary").and_then(|s| s.as_str()).unwrap_or(""); + for (index, cap) in capsules.iter().enumerate() { + let summary = cap.get("summary").and_then(|s| s.as_str()).ok_or_else(|| { + BrainBenchError::KimetsuError(format!( + "context response capsule {index} lacks a string summary" + )) + })?; let body = normalize(strip_prefix_summary(summary)); - if body.is_empty() { - continue; - } + // Chronological rendering adds a date before the visible memory. + // Compression may remove its tail, making neither whole string a + // substring of the other. Ignore only this known date decoration. + let dated_body = body.strip_prefix('[').and_then(|rest| { + let (date, text) = rest.split_once("] ")?; + let bytes = date.as_bytes(); + (bytes.len() == 10 + && bytes[4] == b'-' + && bytes[7] == b'-' + && bytes + .iter() + .enumerate() + .all(|(i, b)| i == 4 || i == 7 || b.is_ascii_digit())) + .then_some(text) + }); // Match the capsule body against fixture memory texts. Prefer exact // normalized equality; fall back to substring containment either way // (the summary may truncate or lightly reword the stored text). - let matched = norm_to_key + let exact: Vec<_> = norm_to_key .iter() - .find(|(norm, _)| *norm == body) - .or_else(|| { - norm_to_key - .iter() - .find(|(norm, _)| body.contains(norm.as_str()) || norm.contains(&body)) - }); - if let Some((_, key)) = matched { + .filter(|(norm, _)| !body.is_empty() && *norm == body) + .collect(); + let matches = if exact.is_empty() { + norm_to_key + .iter() + .filter(|(norm, _)| { + !body.is_empty() + && !norm.is_empty() + && (body.contains(norm.as_str()) + || norm.contains(&body) + || dated_body.is_some_and(|text| { + !text.is_empty() + && (text.contains(norm.as_str()) || norm.contains(text)) + })) + }) + .collect::>() + } else { + exact + }; + if let [(_, key)] = matches.as_slice() { ranked.push(key.clone()); + } else { + let mut unknown = format!("__unmatched_{}__", ranked.len()); + while memories.iter().any(|m| m.key == unknown) { + unknown.push('_'); + } + ranked.push(unknown); } } } @@ -1987,7 +2193,20 @@ fn detect_conflicts_payload(workspace: &Path, kimetsu_bin: &str) -> String { // ─── Per-dimension runners ─────────────────────────────────────────────────── -/// Retrieval correctness: recall@4 + MRR + resolution for knowledge-update. +/// Current-context correctness: recall@4, with no explicitly stale capsule. +/// Ordering stale evidence below the answer still exposes a contradictory claim. +fn retrieval_score(ranked: &[String], relevant: &[String], stale: &[String]) -> f64 { + if relevant.is_empty() { + return if ranked.is_empty() { 1.0 } else { 0.0 }; + } + let recall = recall_at_k(ranked, relevant, 4); + if stale_hit_rate(ranked, stale, 4) == 0.0 { + recall + } else { + 0.0 + } +} + fn run_retrieval( scenario: &Scenario, workspace: &Path, @@ -2009,36 +2228,53 @@ fn run_retrieval( let mut mrrs: Vec = Vec::new(); let mut stale_hits: Vec = Vec::new(); let mut resolutions: Vec = Vec::new(); + let mut negative_injections = Vec::new(); + let mut client = super::brain_mcp::BrainMcp::start(workspace, kimetsu_bin) + .map_err(BrainBenchError::KimetsuError)?; + let mut observations = Vec::new(); for q in &scenario.queries { - let ranked = - retrieve_ranked_keys(workspace, kimetsu_bin, &q.query, budget, &scenario.memories)?; + let measurement = client + .context(&q.query, budget) + .map_err(BrainBenchError::KimetsuError)?; + let ranked = ranked_fixture_keys(&measurement.payload, &scenario.memories)?; + observations.push(observe_query( + &q.query, + &ranked, + &q.relevant, + &q.stale, + &measurement, + )); let r4 = recall_at_k(&ranked, &q.relevant, 4); let m = mrr(&ranked, &q.relevant); let sh = stale_hit_rate(&ranked, &q.stale, 4); let res = resolution_correct(&ranked, &q.relevant, &q.stale); - recalls.push(r4); - mrrs.push(m); - stale_hits.push(sh); - resolutions.push(if res { 1.0 } else { 0.0 }); - - // Per-query blended score: recall@4, gated by resolution when the query - // plants a stale memory (a "current value" must outrank the old one). - let blended = if q.stale.is_empty() { - r4 + if q.relevant.is_empty() { + negative_injections.push(if ranked.is_empty() { 0.0 } else { 1.0 }); } else { - r4 * if res { 1.0 } else { 0.0 } - }; + recalls.push(r4); + mrrs.push(m); + } + if !q.stale.is_empty() { + stale_hits.push(sh); + resolutions.push(if res { 1.0 } else { 0.0 }); + } + + // Explicitly stale claims must be absent from delivered current context. + let blended = retrieval_score(&ranked, &q.relevant, &q.stale); per_query_scores.push(blended); } let score = mean(&per_query_scores); let detail = format!( - "recall@4={:.2} mrr={:.2} stale-hit={:.2} resolution={:.2} ({} quer{})", + "positive-recall@4={:.2} mrr={:.2} stale-hit={} resolution={} false-injection={} positive-n={} negative-n={} ({} quer{})", mean(&recalls), mean(&mrrs), - mean(&stale_hits), - mean(&resolutions), + optional_metric(&stale_hits), + optional_metric(&resolutions), + optional_metric(&negative_injections), + recalls.len(), + negative_injections.len(), scenario.queries.len(), if scenario.queries.len() == 1 { "y" @@ -2046,7 +2282,17 @@ fn run_retrieval( "ies" } ); - Ok(skeleton_result(scenario, score, detail)) + let mut result = skeleton_result(scenario, score, detail); + result.observations = observations; + Ok(result) +} + +fn optional_metric(values: &[f64]) -> String { + if values.is_empty() { + "n/a".into() + } else { + format!("{:.3} (n={})", mean(values), values.len()) + } } /// Workflow stream: one persistent brain, ordered episodes; per episode the @@ -2094,8 +2340,21 @@ fn run_workflow( let mut forbidden_eps = 0usize; let mut trap_hits = 0usize; + let mut client = super::brain_mcp::BrainMcp::start(workspace, kimetsu_bin) + .map_err(BrainBenchError::KimetsuError)?; + let mut observations = Vec::new(); for (ep_idx, ep) in spec.episodes.iter().enumerate() { - let ranked = retrieve_ranked_keys(workspace, kimetsu_bin, &ep.task, budget, &known)?; + let measurement = client + .context(&ep.task, budget) + .map_err(BrainBenchError::KimetsuError)?; + let ranked = ranked_fixture_keys(&measurement.payload, &known)?; + observations.push(observe_query( + &ep.task, + &ranked, + &ep.relevant, + &ep.stale, + &measurement, + )); let es = score_workflow_episode(&ranked, &ep.relevant, &ep.stale, &ep.forbidden, ep.top_k); episode_scores.push(es.score); @@ -2184,7 +2443,9 @@ fn run_workflow( abstentions, known.len() ); - Ok(skeleton_result(scenario, score, detail)) + let mut result = skeleton_result(scenario, score, detail); + result.observations = observations; + Ok(result) } /// Importance ranking: expect_key must land within top_k. @@ -3242,6 +3503,7 @@ fn run_render_contract( fn skeleton_result(scenario: &Scenario, score: f64, detail: String) -> ScenarioResult { ScenarioResult { + observations: Vec::new(), id: scenario.id.clone(), dimension: scenario.dimension, tier: scenario.tier, @@ -3614,15 +3876,20 @@ fn build_report(results: Vec, dataset_path: &Path) -> BrainBench .filter(|r| !r.skipped) .map(|r| r.score) .collect(); - let overall_index = mean(&scored); + let scenario_weighted_index = mean(&scored); + let overall_index = mean(&by_dimension.values().map(|d| d.mean).collect::>()); BrainBenchReport { generated_at: now, dataset: dataset_path.to_string_lossy().to_string(), + session_configuration: Some( + serde_json::json!({"warm_start":false,"include_ambient":false}), + ), scenarios: results, by_dimension_tier, by_dimension, overall_index, + scenario_weighted_index, } } @@ -3645,18 +3912,21 @@ pub fn synthetic_fixture() -> BrainBenchDataset { description: "plain recall of a build command".to_string(), memories: vec![ Memory { + valid_from: None, valid_to: None, key: "build-cmd".to_string(), text: "The project is built with `cargo build --release` from the bench directory.".to_string(), scope: "project".to_string(), kind: "fact".to_string(), }, Memory { + valid_from: None, valid_to: None, key: "test-cmd".to_string(), text: "Run the test suite with `cargo test` from the workspace root.".to_string(), scope: "project".to_string(), kind: "fact".to_string(), }, Memory { + valid_from: None, valid_to: None, key: "lint-cmd".to_string(), text: "Lint the codebase with `cargo clippy --all-targets`.".to_string(), scope: "project".to_string(), @@ -3689,12 +3959,14 @@ pub fn synthetic_fixture() -> BrainBenchDataset { description: "knowledge update: env var supersedes config.toml".to_string(), memories: vec![ Memory { + valid_from: None, valid_to: None, key: "cheap-model-old".to_string(), text: "The cheap model is configured in config.toml under the [models] section.".to_string(), scope: "project".to_string(), kind: "fact".to_string(), }, Memory { + valid_from: None, valid_to: None, key: "cheap-model-new".to_string(), text: "The cheap model is now set via the KIMETSU_CHEAP_MODEL environment variable, not config.toml.".to_string(), scope: "project".to_string(), @@ -3727,18 +3999,21 @@ pub fn synthetic_fixture() -> BrainBenchDataset { description: "salient security memory should rank within top-4".to_string(), memories: vec![ Memory { + valid_from: None, valid_to: None, key: "secret-rule".to_string(), text: "Never commit API keys or secrets to the repository; use the .env file which is gitignored.".to_string(), scope: "project".to_string(), kind: "fact".to_string(), }, Memory { + valid_from: None, valid_to: None, key: "format-pref".to_string(), text: "The team prefers 4-space indentation in Python files.".to_string(), scope: "project".to_string(), kind: "fact".to_string(), }, Memory { + valid_from: None, valid_to: None, key: "ci-note".to_string(), text: "CI runs on GitHub Actions on every push to main.".to_string(), scope: "project".to_string(), @@ -3771,18 +4046,21 @@ pub fn synthetic_fixture() -> BrainBenchDataset { description: "two paraphrases of the same DB-path fact".to_string(), memories: vec![ Memory { + valid_from: None, valid_to: None, key: "db-path-a".to_string(), text: "The brain database lives at .kimetsu/brain.db in the workspace.".to_string(), scope: "project".to_string(), kind: "fact".to_string(), }, Memory { + valid_from: None, valid_to: None, key: "db-path-b".to_string(), text: "The workspace stores its brain database at .kimetsu/brain.db.".to_string(), scope: "project".to_string(), kind: "fact".to_string(), }, Memory { + valid_from: None, valid_to: None, key: "editor-pref".to_string(), text: "The default editor for commit messages is vim.".to_string(), scope: "project".to_string(), @@ -3818,6 +4096,40 @@ pub fn synthetic_fixture() -> BrainBenchDataset { mod tests { use super::*; + #[test] + fn model_override_survives_new_project_retrieval_preset() { + for model in ["ms-marco-minilm-l-4-v2", "off"] { + let mut config = kimetsu_core::config::ProjectConfig::default_for_project("benchmark"); + config.retrieval.level = "deep".into(); + for (key, value) in brain_config_overrides(Some(model)) { + match key { + "broker.warm_start" => config.broker.warm_start = value.parse().unwrap(), + "retrieval.level" => config.retrieval.level = value.into(), + "embedder.reranker" => config.embedder.reranker = value.into(), + _ => panic!("unexpected override {key}"), + } + config.apply_retrieval_level(); + } + assert_eq!(config.embedder.reranker, model); + } + } + + #[test] + fn retrieval_measurements_disable_unscored_warm_start_for_every_model() { + for model in [None, Some("ms-marco-tinybert-l-2-v2"), Some("off")] { + let mut config = kimetsu_core::config::ProjectConfig::default_for_project("benchmark"); + for (key, value) in brain_config_overrides(model) { + if key == "broker.warm_start" { + config.broker.warm_start = value.parse().unwrap(); + } + } + assert!( + !config.broker.warm_start, + "warm text is outside capsule gold coverage" + ); + } + } + fn s(v: &[&str]) -> Vec { v.iter().map(|x| x.to_string()).collect() } @@ -3837,8 +4149,10 @@ mod tests { // Stale outranks relevant -> gated to 0 despite recall hit. let es = score_workflow_episode(&s(&["old", "new"]), &s(&["new"]), &s(&["old"]), &[], 4); assert_eq!(es.score, 0.0); - // Relevant outranks stale -> full recall credit. + // Even below the correct answer, stale context violates the contract. let es = score_workflow_episode(&s(&["new", "old"]), &s(&["new"]), &s(&["old"]), &[], 4); + assert_eq!(es.score, 0.0); + let es = score_workflow_episode(&s(&["new"]), &s(&["new"]), &s(&["old"]), &[], 4); assert_eq!(es.score, 1.0); } @@ -4445,10 +4759,126 @@ mod tests { // ── Report building ─────────────────────────────────────────────────────── + #[test] + fn report_does_not_let_calibration_volume_hide_failed_retrieval() { + let mut results = vec![ScenarioResult { + observations: Vec::new(), + id: "retrieval-miss".into(), + dimension: Dimension::Retrieval, + tier: Tier::Easy, + score: 0.0, + skipped: false, + detail: "miss".into(), + }]; + for i in 0..9 { + results.push(ScenarioResult { + observations: Vec::new(), + id: format!("calibration-{i}"), + dimension: Dimension::Calibration, + tier: Tier::Easy, + score: 1.0, + skipped: false, + detail: "ok".into(), + }); + } + let report = build_report(results, Path::new("test.json")); + assert_eq!(report.overall_index, 0.5); + } + + #[test] + fn irrelevant_memory_is_not_perfect_no_answer_retrieval() { + assert_eq!(retrieval_score(&["unrelated".into()], &[], &[]), 0.0); + assert_eq!(retrieval_score(&[], &[], &[]), 1.0); + } + + #[test] + fn retrieval_rejects_stale_context_even_below_correct_answer() { + assert_eq!( + retrieval_score( + &s(&["current", "expired"]), + &s(&["current"]), + &s(&["expired"]) + ), + 0.0 + ); + assert_eq!( + retrieval_score(&s(&["current"]), &s(&["current"]), &s(&["expired"])), + 1.0 + ); + } + + #[test] + fn dated_compressed_capsule_matches_visible_fixture_text() { + let memories = vec![Memory { key: "gold".into(), text: "Use stdout for protocol. Send diagnostics to stderr. Keep logs separate. Extra explanation.".into(), scope: "project".into(), kind: "fact".into(), valid_from: None, valid_to: None }]; + let payload = serde_json::json!({"capsules":[{"summary":"project:fact - [2026-09-07] Use stdout for protocol. Send diagnostics to stderr. Keep logs separate."}]}); + assert_eq!( + ranked_fixture_keys(&payload, &memories).unwrap(), + vec!["gold"] + ); + let unrelated = serde_json::json!({"capsules":[{"summary":"project:fact - [2026-09-07] Use a different database."}]}); + assert_ne!( + ranked_fixture_keys(&unrelated, &memories).unwrap(), + vec!["gold"] + ); + } + + #[test] + fn query_observation_retains_delivered_answerability() { + let measurement = super::super::brain_mcp::McpMeasurement { + payload: serde_json::json!({"answerability":{"status":"partial","missing":["timeout"]}}), + text_bytes: 0, + result_bytes: 0, + wire_bytes: 0, + latency_ms: 1.0, + first_query: false, + server_startup_ms: 0.0, + working_set_bytes: None, + peak_working_set_bytes: None, + }; + let observation = observe_query("q", &[], &[], &[], &measurement); + assert_eq!( + serde_json::to_value(observation).unwrap()["answerability"]["missing"], + serde_json::json!(["timeout"]) + ); + } + #[test] + fn unknown_capsules_keep_their_rank_and_count_as_injection() { + let memories = vec![Memory { + valid_from: None, + valid_to: None, + key: "known".into(), + text: "known evidence".into(), + scope: "project".into(), + kind: "fact".into(), + }]; + let payload = serde_json::json!({"capsules": [ + {"summary": "unmatched evidence"}, {"summary": "known evidence"} + ]}); + let ranked = ranked_fixture_keys(&payload, &memories).unwrap(); + assert_eq!(ranked.len(), 2); + assert_eq!(mrr(&ranked, &["known".into()]), 0.5); + assert_eq!(retrieval_score(&ranked, &[], &[]), 0.0); + } + + #[test] + fn malformed_capsule_summary_is_an_error() { + for malformed in [ + serde_json::json!(null), + serde_json::json!({}), + serde_json::json!({"summary": null}), + serde_json::json!({"summary": 42}), + ] { + let payload = serde_json::json!({"capsules": [malformed]}); + let error = ranked_fixture_keys(&payload, &[]).unwrap_err(); + assert!(error.to_string().contains("capsule 0")); + } + } + #[test] fn build_report_excludes_skipped_from_index() { let results = vec![ ScenarioResult { + observations: Vec::new(), id: "r1".to_string(), dimension: Dimension::Retrieval, tier: Tier::Easy, @@ -4457,6 +4887,7 @@ mod tests { detail: "ok".to_string(), }, ScenarioResult { + observations: Vec::new(), id: "r2".to_string(), dimension: Dimension::Retrieval, tier: Tier::Easy, @@ -4465,6 +4896,7 @@ mod tests { detail: "miss".to_string(), }, ScenarioResult { + observations: Vec::new(), id: "f1".to_string(), dimension: Dimension::Forgetting, tier: Tier::Medium, @@ -4495,14 +4927,20 @@ mod tests { let fixture = EvalFixtureFile { memories: vec![ EvalFixMemory { + valid_from: None, + valid_to: None, key: "m-old".to_string(), text: "old value".to_string(), }, EvalFixMemory { + valid_from: None, + valid_to: None, key: "m-new".to_string(), text: "new value".to_string(), }, EvalFixMemory { + valid_from: None, + valid_to: None, key: "m-plain".to_string(), text: "a plain fact".to_string(), }, @@ -4580,6 +5018,8 @@ mod tests { fn write_pool(dir: &Path, name: &str, n: usize) -> PathBuf { let memories: Vec = (0..n) .map(|i| EvalFixMemory { + valid_from: None, + valid_to: None, key: format!("m{i}"), text: format!("distinct fact number {i} about subsystem {i}"), }) diff --git a/src/drivers/mod.rs b/src/drivers/mod.rs index d115e1e..d8be883 100644 --- a/src/drivers/mod.rs +++ b/src/drivers/mod.rs @@ -5,6 +5,7 @@ //! internal corpora) drop in as new files here. pub mod beam; +mod brain_mcp; pub mod brainbench; pub mod locomo; pub mod longmemeval;