From 37b61771d27abf8f07ebdaff8682e1407eed3b13 Mon Sep 17 00:00:00 2001 From: phernandez Date: Sun, 23 Aug 2026 11:42:31 -0500 Subject: [PATCH 1/3] feat(core): add concurrent-write benchmark driver Signed-off-by: phernandez --- benchmarks/AGENTS.md | 10 +- benchmarks/README.md | 44 +- benchmarks/docs/benchmarks.md | 20 +- benchmarks/justfile | 27 + .../src/basic_memory_benchmarks/bm_runtime.py | 226 ++++ benchmarks/src/basic_memory_benchmarks/cli.py | 71 ++ .../concurrent_write.py | 987 ++++++++++++++++++ .../providers/bm_local.py | 149 +-- benchmarks/tests/test_concurrent_write.py | 420 ++++++++ 9 files changed, 1800 insertions(+), 154 deletions(-) create mode 100644 benchmarks/src/basic_memory_benchmarks/bm_runtime.py create mode 100644 benchmarks/src/basic_memory_benchmarks/concurrent_write.py create mode 100644 benchmarks/tests/test_concurrent_write.py diff --git a/benchmarks/AGENTS.md b/benchmarks/AGENTS.md index 7f8bf1457..7827cc1be 100644 --- a/benchmarks/AGENTS.md +++ b/benchmarks/AGENTS.md @@ -1,15 +1,17 @@ -# AGENTS.md - basic-memory-benchmarks Guide +# AGENTS.md - Basic Memory benchmarks guide ## Project Overview -`basic-memory-benchmarks` is a standalone benchmark harness for comparing Basic Memory against other memory systems. +`basic-memory-benchmarks` is the benchmark harness within Core's `/benchmarks` +directory for comparing Basic Memory against other memory systems. Primary goals: - Deterministic retrieval benchmarks - Optional LLM-as-a-judge benchmarks - Public, reproducible artifact publication (including provenance metadata) -This repo is intentionally isolated from `basic-memory` so benchmark dependencies do not pollute the product repo. +The benchmark package keeps its own `pyproject.toml` and lockfile so benchmark +dependencies do not pollute the Core product environment. ## Build / Test Commands @@ -44,6 +46,8 @@ Validate and publish: `just` shortcuts: - `just bench-smoke` +- `BM_LOCAL_PATH=.. just bench-concurrent-write-smoke` +- `BM_LOCAL_PATH=.. just bench-concurrent-write-load` - `just bench-fetch-locomo` - `just bench-convert-locomo` - `just bench-run-bm-local` diff --git a/benchmarks/README.md b/benchmarks/README.md index 5a9ba75c4..73c0d59ed 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -1,13 +1,15 @@ -# basic-memory-benchmarks +# Basic Memory benchmarks -Standalone, reproducible benchmark suite for comparing Basic Memory against competitor memory systems. +Reproducible benchmark suite for comparing Basic Memory against competitor memory systems. The suite +lives in Core under `/benchmarks` and keeps its own `pyproject.toml` and lockfile so benchmark-only +dependencies remain isolated from the product environment. ## Goals - Deterministic retrieval benchmarks (Recall@5/10, MRR, Precision@5, content-hit, latency) - Optional LLM-as-judge scoring (Pydantic Evals) - Public artifacts with provenance and reproducibility metadata -- Clean dependency isolation from the core `basic-memory` repository +- Clean dependency isolation from the core Basic Memory package ## Current v1 Scope @@ -169,6 +171,42 @@ Anti-leakage: raw conversations carry `containsEvidence`/`model_name` fields; rendered docs include neither and conversation ids are remapped to neutral positional ids. +## Concurrent-write benchmark (basic-memory#1248) + +Measures correctness under concurrency: independent `bm mcp` client sessions +create and edit notes in one shared Basic Memory project, with overlapping +relation targets and shared hub notes that every writer appends to (the +multi-agent shape from basic-memory#1213/#1214). + +The driver requires a local git checkout so every result records the exact +Basic Memory commit under test: + +```bash +# Run from the Core benchmarks directory. +BM_LOCAL_PATH=.. just bench-concurrent-write-smoke + +# Load shape, report-only (divergence is a valid benchmark result). +BM_LOCAL_PATH=.. just bench-concurrent-write-load writers=8 notes=200 + +# Direct invocation against another checkout or worktree. +uv run bm-bench run concurrent-write \ + --writers 4 --notes-per-writer 25 \ + --bm-local-path /path/to/basic-memory +``` + +Per run (`benchmarks/runs//`): `manifest.json`, `per-op.jsonl`, +`concurrent-write-summary.json`, and `summary.md`. After the concurrent phase +settles, the driver records convergence directly from the on-disk files and +the isolated SQLite index before optionally timing a full reindex. It checks +file/entity/row counts, duplicate permalinks, duplicate observation or relation +tuples, and unique `bmk-*` markers that detect lost or doubled writes. + +The run uses a fresh isolated home under `benchmarks/.bm-homes/`; environment +variables such as `BASIC_MEMORY_SEMANTIC_SEARCH_ENABLED` pass through, so axes +like Redis on/off are controlled the same way as the retrieval scripts. +Postgres row-integrity checks are a follow-up; the write workload itself is +database-agnostic. + ## Basic Memory source policy By default this project tracks Basic Memory from `main`. diff --git a/benchmarks/docs/benchmarks.md b/benchmarks/docs/benchmarks.md index 59a9066c8..17952f07f 100644 --- a/benchmarks/docs/benchmarks.md +++ b/benchmarks/docs/benchmarks.md @@ -1,7 +1,7 @@ # Benchmark Runbook This document is the canonical operator runbook for benchmark execution in -`basic-memory-benchmarks`. +the Core repository's `/benchmarks` package. It covers: 1. current benchmark workflows and commands, @@ -13,6 +13,7 @@ It covers: | Area | Status | | --- | --- | | Single run execution (`run retrieval`, `run full`, `run judge`) | Implemented | +| Concurrent write convergence (`run concurrent-write`) | Implemented | | `just` one-command pipelines (`bench-full`, `bench-full-judge`) | Implemented | | Artifact generation and publish/compare commands | Implemented | | Manual BM revision comparison via worktrees + `--bm-local-path` | Implemented workflow, manual orchestration | @@ -42,8 +43,8 @@ It covers: ### Repositories and paths -- benchmark repo: clone of `basicmachines-co/basic-memory-benchmarks` -- BM local repo: set `BM_LOCAL_PATH` env var (or in `.env`) to your local `basic-memory` checkout +- benchmark package: `/benchmarks` in a clone of `basicmachines-co/basic-memory` +- BM local repo: set `BM_LOCAL_PATH` env var (or in `.env`) to the Core checkout under test ### Environment @@ -53,7 +54,7 @@ It covers: ### One-time setup ```bash -cd /path/to/basic-memory-benchmarks +cd /path/to/basic-memory/benchmarks just sync ``` @@ -78,6 +79,8 @@ just bench-prepare-long - `bench-full` - `bench-full-judge` +- `bench-concurrent-write-smoke` +- `bench-concurrent-write-load` - `bench-prepare-short` - `bench-prepare-long` - `bench-run-short` @@ -96,6 +99,7 @@ Top-level commands: - `datasets fetch` - `convert locomo` - `run retrieval` +- `run concurrent-write` - `run full` - `run judge` - `compare` @@ -107,7 +111,7 @@ Top-level commands: ### One-command full retrieval run ```bash -cd /path/to/basic-memory-benchmarks +cd /path/to/basic-memory/benchmarks just bench-full ``` @@ -119,7 +123,7 @@ This runs: ### One-command full retrieval + judge ```bash -cd /path/to/basic-memory-benchmarks +cd /path/to/basic-memory/benchmarks just bench-full-judge ``` @@ -236,7 +240,7 @@ Use this workflow today to compare BM revisions while keeping benchmark tooling ```bash BM_REPO=/path/to/basic-memory -WT_ROOT=/path/to/basic-memory-benchmarks/benchmarks/worktrees/basic-memory +WT_ROOT=/path/to/basic-memory/benchmarks/benchmarks/worktrees/basic-memory mkdir -p "$WT_ROOT" @@ -250,7 +254,7 @@ git -C "$BM_REPO" worktree add "$WT_ROOT/current" HEAD ### Step 2: Prepare benchmark datasets once ```bash -cd /path/to/basic-memory-benchmarks +cd /path/to/basic-memory/benchmarks just sync just bench-prepare-short just bench-prepare-long diff --git a/benchmarks/justfile b/benchmarks/justfile index 5ac841e47..6bcff0ae9 100644 --- a/benchmarks/justfile +++ b/benchmarks/justfile @@ -210,6 +210,33 @@ bench-run-full-judge model="gpt-4o-mini": --judge \ --judge-model "{{model}}" +# --- Concurrency benchmark (basic-memory#1248) --- + +# Small-scale smoke: 4 writers x 25 notes; strict so divergence fails the command +bench-concurrent-write-smoke: + #!/usr/bin/env bash + set -euo pipefail + if [[ -z "{{bm_local_path}}" ]]; then + echo "BM_LOCAL_PATH must point to the Basic Memory git checkout under test" >&2 + exit 2 + fi + uv run bm-bench run concurrent-write \ + --writers 4 --notes-per-writer 25 \ + --bm-local-path "{{bm_local_path}}" \ + --strict + +# Load shape for the v0.22.1-vs-v0.23 comparison; report-only (divergence is the result) +bench-concurrent-write-load writers="8" notes="200": + #!/usr/bin/env bash + set -euo pipefail + if [[ -z "{{bm_local_path}}" ]]; then + echo "BM_LOCAL_PATH must point to the Basic Memory git checkout under test" >&2 + exit 2 + fi + uv run bm-bench run concurrent-write \ + --writers {{writers}} --notes-per-writer {{notes}} \ + --bm-local-path "{{bm_local_path}}" + # --- Artifacts and comparison --- bench-latest-run: diff --git a/benchmarks/src/basic_memory_benchmarks/bm_runtime.py b/benchmarks/src/basic_memory_benchmarks/bm_runtime.py new file mode 100644 index 000000000..0e6762d8b --- /dev/null +++ b/benchmarks/src/basic_memory_benchmarks/bm_runtime.py @@ -0,0 +1,226 @@ +"""Shared helpers for driving an external Basic Memory runtime. + +Everything here talks to Basic Memory through its public contracts only — the +`bm` CLI and the `bm mcp` stdio server — never through internal imports, so the +same code runs unchanged against any BM version under comparison (installed +`bm`, or a checkout via ``uv run --project basic-memory``). +""" + +from __future__ import annotations + +import asyncio +import threading +from concurrent.futures import Future +from dataclasses import dataclass +from pathlib import Path +from queue import Empty, Queue +from typing import Any + +from mcp.client.session import ClientSession +from mcp.client.stdio import StdioServerParameters, stdio_client +from mcp.types import CallToolResult + + +@dataclass +class _McpToolRequest: + name: str + arguments: dict[str, Any] + response: Future[CallToolResult] + + +class WarmMcpClient: + """One warm `bm mcp` stdio session, callable from any thread. + + The session runs on its own thread with its own subprocess; `call_tool` + marshals requests through a queue so callers pay startup cost once per + session instead of once per tool call. Requests are strictly one at a time + per session — concurrency comes from running multiple sessions. + """ + + def __init__( + self, + *, + command: str = "bm", + args: list[str] | None = None, + env: dict[str, str] | None = None, + startup_timeout_seconds: float = 30.0, + request_timeout_seconds: float = 60.0, + required_tool: str = "search_notes", + ) -> None: + self._command = command + self._args = args or ["mcp"] + self._env = env + self._startup_timeout_seconds = startup_timeout_seconds + self._request_timeout_seconds = request_timeout_seconds + self._required_tool = required_tool + self._requests: Queue[_McpToolRequest | None] = Queue() + self._ready = threading.Event() + self._startup_error: Exception | None = None + self._thread: threading.Thread | None = None + self._state_lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._serve_task: asyncio.Task[None] | None = None + + async def _serve(self) -> None: + loop = asyncio.get_running_loop() + task = asyncio.current_task() + if task is None: # pragma: no cover - asyncio always supplies the current task + raise RuntimeError("MCP session started without an asyncio task") + with self._state_lock: + self._loop = loop + self._serve_task = task + + params = StdioServerParameters(command=self._command, args=self._args, env=self._env) + async with stdio_client(params) as (read_stream, write_stream): + async with ClientSession(read_stream, write_stream) as session: + await session.initialize() + tools = await session.list_tools() + tool_names = {tool.name for tool in tools.tools} + if self._required_tool not in tool_names: + raise RuntimeError(f"bm mcp server does not expose '{self._required_tool}'") + + self._ready.set() + + while True: + request = await asyncio.to_thread(self._requests.get) + if request is None: + break + try: + result = await session.call_tool(request.name, request.arguments) + except asyncio.CancelledError: + if not request.response.done(): + request.response.set_exception( + RuntimeError("bm mcp session stopped during tool call") + ) + raise + except Exception as exc: + if not request.response.done(): + request.response.set_exception(exc) + else: + if not request.response.done(): + request.response.set_result(result) + + def _fail_pending_requests(self) -> None: + while True: + try: + request = self._requests.get_nowait() + except Empty: + return + if request is not None and not request.response.done(): + request.response.set_exception(RuntimeError("bm mcp session stopped")) + + def _thread_main(self) -> None: + try: + asyncio.run(self._serve()) + except asyncio.CancelledError: + pass + except Exception as exc: + self._startup_error = exc + finally: + self._fail_pending_requests() + with self._state_lock: + self._loop = None + self._serve_task = None + self._ready.set() + + def start(self) -> None: + if self._thread is not None and self._thread.is_alive(): + return + self._ready.clear() + self._startup_error = None + self._thread = threading.Thread( + target=self._thread_main, + name="bm-benchmark-mcp-client", + daemon=True, + ) + self._thread.start() + + if not self._ready.wait(timeout=self._startup_timeout_seconds): + self.stop() + raise TimeoutError("Timed out starting bm mcp session") + if self._startup_error is not None: + startup_error = self._startup_error + self.stop() + raise RuntimeError("Failed to start bm mcp session") from startup_error + + def call_tool(self, name: str, arguments: dict[str, Any]) -> CallToolResult: + if self._thread is None or not self._thread.is_alive(): + raise RuntimeError("bm mcp session is not running") + + response: Future[CallToolResult] = Future() + self._requests.put(_McpToolRequest(name=name, arguments=arguments, response=response)) + return response.result(timeout=self._request_timeout_seconds) + + def stop(self) -> None: + thread = self._thread + if thread is None: + return + + if thread.is_alive(): + # Trigger: stop can follow a timed-out tool call that is still running. + # Why: queued shutdown alone cannot interrupt that call, so verification + # could race an untracked writer. Outcome: task cancellation exits the + # stdio context, whose MCP transport terminates the child process. + self._requests.put(None) + with self._state_lock: + loop = self._loop + serve_task = self._serve_task + if loop is not None and serve_task is not None and loop.is_running(): + loop.call_soon_threadsafe(serve_task.cancel) + thread.join(timeout=self._startup_timeout_seconds) + if thread.is_alive(): + raise RuntimeError("bm mcp session did not stop before the shutdown deadline") + + self._thread = None + + +def resolve_bm_command_prefix(bm_local_path: str | None) -> list[str]: + """Resolve how to invoke Basic Memory: installed `bm` or a local checkout.""" + if bm_local_path: + local_path = Path(bm_local_path) + if not local_path.exists(): + raise ValueError(f"--bm-local-path not found: {local_path}") + return ["uv", "run", "--project", str(local_path), "basic-memory"] + return ["bm"] + + +def status_json_is_ready(payload: dict[str, Any]) -> bool: + """Interpret `bm status --json` output across BM versions. + + The schema varies by version; every known busy signal is checked, and an + unknown schema with no busy signal counts as ready. + """ + total = payload.get("total") + if isinstance(total, int): + return total == 0 + + for list_key in ("new", "modified", "deleted", "skipped_files"): + value = payload.get(list_key) + if isinstance(value, list) and len(value) > 0: + return False + + for dict_key in ("moves", "checksums"): + value = payload.get(dict_key) + if isinstance(value, dict) and len(value) > 0: + return False + + status = payload.get("status") + if isinstance(status, str): + lowered = status.lower() + if "no changes" in lowered or "up to date" in lowered: + return True + if "sync" in lowered or "index" in lowered or "pending" in lowered: + return False + + for key in ("is_syncing", "is_indexing", "sync_in_progress", "index_in_progress"): + value = payload.get(key) + if isinstance(value, bool): + return not value + + for key in ("pending_files", "pending", "unindexed_files", "queued_files", "queue_size"): + value = payload.get(key) + if isinstance(value, int) and value != 0: + return False + + # If the schema is unknown and no busy signal exists, treat status as ready. + return True diff --git a/benchmarks/src/basic_memory_benchmarks/cli.py b/benchmarks/src/basic_memory_benchmarks/cli.py index edeb8bdc3..42d33cc12 100644 --- a/benchmarks/src/basic_memory_benchmarks/cli.py +++ b/benchmarks/src/basic_memory_benchmarks/cli.py @@ -9,6 +9,10 @@ import typer from rich.console import Console +from basic_memory_benchmarks.concurrent_write import ( + ConcurrentWriteConfig, + run_concurrent_write, +) from basic_memory_benchmarks.converters.locomo_to_corpus import convert_locomo_to_corpus from basic_memory_benchmarks.converters.longmemeval_to_corpus import convert_longmemeval_to_corpus from basic_memory_benchmarks.datasets.locomo import LOCOMO_URL, fetch_locomo_dataset @@ -262,6 +266,73 @@ def run_retrieval_command( console.print(f"Retrieval run complete: [green]{run_dir}[/green]") +@run_app.command("concurrent-write") +def run_concurrent_write_command( + writers: int = typer.Option(4, "--writers", help="Concurrent MCP client sessions"), + notes_per_writer: int = typer.Option(25, "--notes-per-writer"), + edit_ratio: float = typer.Option( + 0.4, "--edit-ratio", help="Per-note probability of hub/own-note append edits" + ), + hub_notes: int = typer.Option(4, "--hub-notes", help="Shared contended notes all writers edit"), + relation_pool: int = typer.Option( + 8, "--relation-pool", help="Shared relation-target pool size" + ), + seed: int = typer.Option(42, "--seed"), + run_id: str | None = typer.Option(None, "--run-id"), + output_root: Path = typer.Option(Path("benchmarks/runs"), "--output-root"), + bm_source: str = typer.Option("local-checkout", "--bm-source"), + bm_local_path: Path = typer.Option( + ..., + "--bm-local-path", + exists=True, + file_okay=False, + resolve_path=True, + help="Pinned Basic Memory git checkout to benchmark", + ), + max_seconds: float | None = typer.Option( + None, "--max-seconds", help="Optional wall-clock cap for the concurrent phase" + ), + op_timeout: float = typer.Option(120.0, "--op-timeout"), + settle_timeout: float = typer.Option(180.0, "--settle-timeout"), + measure_reindex: bool = typer.Option(True, "--measure-reindex/--no-measure-reindex"), + strict: bool = typer.Option( + False, + "--strict/--no-strict", + help="Exit nonzero when convergence checks fail; the default records divergence", + ), +) -> None: + """Run independent MCP writers against one shared project (basic-memory#1248).""" + resolved_run_id = run_id or f"cw-{uuid.uuid4().hex[:12]}" + config = ConcurrentWriteConfig( + run_id=resolved_run_id, + writers=writers, + notes_per_writer=notes_per_writer, + edit_ratio=edit_ratio, + hub_notes=hub_notes, + relation_pool=relation_pool, + seed=seed, + output_root=str(output_root), + bm_source=bm_source, + bm_local_path=str(bm_local_path), + max_seconds=max_seconds, + op_timeout_seconds=op_timeout, + settle_timeout_seconds=settle_timeout, + measure_reindex=measure_reindex, + ) + run_dir = run_concurrent_write(config) + console.print(f"Concurrent-write run complete: [green]{run_dir}[/green]") + + if strict: + import json + + summary = json.loads( + (run_dir / "concurrent-write-summary.json").read_text(encoding="utf-8") + ) + if not summary["converged"]: + console.print("[red]Convergence checks failed (--strict)[/red]") + raise typer.Exit(code=1) + + @run_app.command("qa") def run_qa_command( run_dir: Path = typer.Option(..., "--run-dir"), diff --git a/benchmarks/src/basic_memory_benchmarks/concurrent_write.py b/benchmarks/src/basic_memory_benchmarks/concurrent_write.py new file mode 100644 index 000000000..6db1dc7f6 --- /dev/null +++ b/benchmarks/src/basic_memory_benchmarks/concurrent_write.py @@ -0,0 +1,987 @@ +"""Concurrent-writer benchmark driver (basicmachines-co/basic-memory#1248, axes 1 and 4). + +Measures correctness-under-concurrency: N independent MCP client sessions (one +`bm mcp` subprocess each — the basic-memory#1214 field-report shape of several +agents writing at once) create and edit notes in ONE shared Basic Memory +project, with overlapping relation targets and shared "hub" notes that every +writer appends to. The driver captures per-op latency and errors, waits for the +index to settle, then verifies the project converged: + +- markdown file count == hub notes + successful creates +- entity rows in the index DB == markdown files on disk +- no duplicate permalinks +- no duplicate (entity, category, content) observation tuples (the #1214 metric) +- no duplicate (from, to, type) relation tuples +- every observation line written by a reported-success op is present exactly + once on disk (each generated line carries a unique ``bmk-*`` marker, so lost + appends and doubled appends are both detectable) + +v0.22.1 is expected to fail under load (relation-table deadlocks #1213, +duplicate observations #1214); v0.23's generation fences must keep every check +green. Throughput and latency are secondary outputs, reported per op type. +Integrity checks read the run's isolated SQLite database directly; Postgres +integrity is a follow-up. +""" + +from __future__ import annotations + +import json +import math +import os +import random +import re +import sqlite3 +import threading +import time +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path +from statistics import mean +from typing import Any, Literal + +from pydantic import BaseModel, Field +from rich.console import Console + +from basic_memory_benchmarks.bm_runtime import ( + WarmMcpClient, + resolve_bm_command_prefix, + status_json_is_ready, +) +from basic_memory_benchmarks.models import RuntimeInfo +from basic_memory_benchmarks.utils import git_sha, run_command, runtime_info, utc_now_iso + +console = Console() + +OpType = Literal["create_hub", "create", "edit_hub", "edit_own"] + +MARKER_PATTERN = re.compile(r"bmk-[a-z0-9-]+") + +# --- Configuration and artifact models --- + + +class ConcurrentWriteConfig(BaseModel): + run_id: str + writers: int = Field(default=4, ge=1) + notes_per_writer: int = Field(default=25, ge=1) + # Probability (per created note, evaluated twice) that a writer also + # appends to a shared hub note / one of its own earlier notes. + edit_ratio: float = Field(default=0.4, ge=0.0, le=1.0) + hub_notes: int = Field(default=4, ge=0) + relation_pool: int = Field(default=8, ge=1) + seed: int = 42 + output_root: str = "benchmarks/runs" + bm_source: str = "local-checkout" + bm_local_path: str + # Optional wall-clock cap: writers stop scheduling new ops once exceeded. + max_seconds: float | None = Field(default=None, gt=0.0) + op_timeout_seconds: float = Field(default=120.0, gt=0.0) + settle_timeout_seconds: float = Field(default=180.0, gt=0.0) + measure_reindex: bool = True + + @property + def project_name(self) -> str: + return f"bm-write-{self.run_id}" + + +class OpResult(BaseModel): + writer: int + op_index: int + op_type: OpType + identifier: str + started_at_utc: str + latency_ms: float + ok: bool + error: str | None = None + error_kind: str | None = None + markers: list[str] = Field(default_factory=list) + + +class OpTypeStats(BaseModel): + count: int + ok: int + errors: int + mean_ms: float + p50_ms: float + p95_ms: float + max_ms: float + + +class IntegrityCheck(BaseModel): + name: str + passed: bool + detail: str + + +class IntegrityReport(BaseModel): + checks: list[IntegrityCheck] + markdown_files: int + entity_rows: int + observation_rows: int + distinct_observation_tuples: int + duplicate_observation_tuples: int + observation_redundancy_pct: float + duplicate_permalinks: int + duplicate_relation_tuples: int + expected_markers: int + found_markers: int + missing_markers: int + duplicated_markers: int + missing_marker_sample: list[str] = Field(default_factory=list) + duplicate_permalink_sample: list[str] = Field(default_factory=list) + duplicate_observation_sample: list[str] = Field(default_factory=list) + converged: bool + + +class ConcurrentWriteSummary(BaseModel): + run_id: str + concurrent_wall_seconds: float + settle_seconds: float + settle_mode: Literal["status-json", "fixed-delay"] + reindex_seconds: float | None = None + ops_total: int + ops_ok: int + ops_error: int + ops_not_attempted: int + terminal_writer_failures: int + error_kinds: dict[str, int] = Field(default_factory=dict) + per_op_type: dict[str, OpTypeStats] = Field(default_factory=dict) + notes_created_ok: int + creates_per_minute: float + ops_per_second: float + integrity: IntegrityReport + converged: bool + + +class ConcurrentWriteManifest(BaseModel): + run_id: str + created_at_utc: str + benchmark_git_sha: str + bm_source: str + bm_resolved_sha: str + bm_local_path: str + bm_version: str | None = None + home_dir: str + project_dir: str + project_name: str + runtime: RuntimeInfo + config: ConcurrentWriteConfig + + +# --- Workload planning (pure, deterministic) --- + + +@dataclass(frozen=True) +class PlannedOp: + writer: int + op_index: int + op_type: OpType + # For creates: the note title; edits target `identifier` instead. + title: str + identifier: str + directory: str + content: str + markers: tuple[str, ...] + + +def _observation_line(category: str, text: str, marker: str, tag: str | None = None) -> str: + suffix = f" #{tag}" if tag else "" + return f"- [{category}] {text} {marker}{suffix}" + + +def build_hub_ops(config: ConcurrentWriteConfig) -> list[PlannedOp]: + """Sequential setup ops: shared hub notes every writer will append to.""" + ops: list[PlannedOp] = [] + for hub in range(config.hub_notes): + title = f"hub-{hub}" + marker = f"bmk-setup-h{hub:02d}-l0" + topic = hub % config.relation_pool + content = "\n".join( + [ + "## Observations", + _observation_line( + "hub", f"shared hub {hub} seeded before concurrent phase", marker, "bench" + ), + "", + "## Relations", + f"- relates_to [[topic-{topic}]]", + ] + ) + ops.append( + PlannedOp( + writer=-1, + op_index=hub, + op_type="create_hub", + title=title, + identifier=f"hubs/{title}", + directory="hubs", + content=content, + markers=(marker,), + ) + ) + return ops + + +def build_writer_plan(writer: int, config: ConcurrentWriteConfig) -> list[PlannedOp]: + """Deterministic op schedule for one writer. + + Every observation line carries a unique ``bmk-w-o-l`` + marker, so post-run file scans can prove each reported-success write + survived exactly once. Relation targets are drawn from small shared pools + (topics and hubs) to create the overlapping-entity contention that + triggered #1213/#1214. + """ + rng = random.Random(config.seed * 7919 + writer) + ops: list[PlannedOp] = [] + op_index = 0 + for note_index in range(config.notes_per_writer): + title = f"w{writer:02d}-n{note_index:04d}" + markers = tuple(f"bmk-w{writer:02d}-o{op_index:04d}-l{line}" for line in range(3)) + topic_a = rng.randrange(config.relation_pool) + topic_b = rng.randrange(config.relation_pool) + lines = [ + "## Observations", + _observation_line( + "fact", f"writer {writer} note {note_index} primary fact", markers[0], "bench" + ), + _observation_line( + "detail", f"writer {writer} note {note_index} supporting detail", markers[1] + ), + _observation_line( + "status", f"writer {writer} note {note_index} status entry", markers[2] + ), + "", + "## Relations", + f"- relates_to [[topic-{topic_a}]]", + f"- part_of [[topic-{topic_b}]]", + ] + if config.hub_notes > 0: + hub = rng.randrange(config.hub_notes) + lines.append(f"- references [[hub-{hub}]]") + ops.append( + PlannedOp( + writer=writer, + op_index=op_index, + op_type="create", + title=title, + identifier=f"notes/{title}", + directory="notes", + content="\n".join(lines), + markers=markers, + ) + ) + op_index += 1 + + if config.hub_notes > 0 and rng.random() < config.edit_ratio: + hub = rng.randrange(config.hub_notes) + marker = f"bmk-w{writer:02d}-o{op_index:04d}-l0" + ops.append( + PlannedOp( + writer=writer, + op_index=op_index, + op_type="edit_hub", + title="", + identifier=f"hubs/hub-{hub}", + directory="hubs", + content="\n" + + _observation_line("update", f"writer {writer} appended to hub {hub}", marker), + markers=(marker,), + ) + ) + op_index += 1 + + if rng.random() < config.edit_ratio: + target = rng.randrange(note_index + 1) + marker = f"bmk-w{writer:02d}-o{op_index:04d}-l0" + ops.append( + PlannedOp( + writer=writer, + op_index=op_index, + op_type="edit_own", + title="", + identifier=f"notes/w{writer:02d}-n{target:04d}", + directory="notes", + content="\n" + + _observation_line( + "update", f"writer {writer} revisited note {target}", marker + ), + markers=(marker,), + ) + ) + op_index += 1 + return ops + + +# --- Execution --- + + +def classify_error(text: str) -> str: + lowered = text.lower() + if "deadlock" in lowered: + return "deadlock" + if "database is locked" in lowered or "database table is locked" in lowered: + return "sqlite_locked" + if "timed out" in lowered or "timeout" in lowered: + return "timeout" + if "conflict" in lowered or "version" in lowered or "stale" in lowered: + return "write_conflict" + return "other" + + +@dataclass +class WriterOutcome: + results: list[OpResult] = field(default_factory=list) + terminal_error: str | None = None + not_attempted: int = 0 + + +def _tool_call_for(op: PlannedOp, project_name: str) -> tuple[str, dict[str, Any]]: + if op.op_type in ("create", "create_hub"): + return "write_note", { + "title": op.title, + "directory": op.directory, + "content": op.content, + "project": project_name, + } + return "edit_note", { + "identifier": op.identifier, + "operation": "append", + "content": op.content, + "project": project_name, + } + + +def _execute_op(client: WarmMcpClient, op: PlannedOp, project_name: str) -> tuple[OpResult, bool]: + """Run one op; returns (result, terminal) — terminal means the session is unusable. + + A timed-out call leaves the request in flight on the single-slot session, + so timeouts are terminal; MCP-level tool errors are recorded and the + session keeps going (those errors ARE the measurement on v0.22.1). + """ + tool, arguments = _tool_call_for(op, project_name) + started_at = utc_now_iso() + start = time.perf_counter() + try: + result = client.call_tool(tool, arguments) + except TimeoutError: + latency_ms = (time.perf_counter() - start) * 1000 + error = f"tool call timed out after {latency_ms / 1000:.0f}s" + return ( + _op_result(op, started_at, latency_ms, ok=False, error=error), + True, + ) + except Exception as exc: + latency_ms = (time.perf_counter() - start) * 1000 + session_dead = isinstance(exc, RuntimeError) and "not running" in str(exc) + return ( + _op_result(op, started_at, latency_ms, ok=False, error=f"{type(exc).__name__}: {exc}"), + session_dead, + ) + latency_ms = (time.perf_counter() - start) * 1000 + if result.isError: + error = "Unknown MCP tool error" + for item in result.content: + text = getattr(item, "text", None) + if isinstance(text, str) and text.strip(): + error = text.strip() + break + return _op_result(op, started_at, latency_ms, ok=False, error=error), False + return _op_result(op, started_at, latency_ms, ok=True, error=None), False + + +def _op_result( + op: PlannedOp, started_at: str, latency_ms: float, *, ok: bool, error: str | None +) -> OpResult: + return OpResult( + writer=op.writer, + op_index=op.op_index, + op_type=op.op_type, + identifier=op.identifier, + started_at_utc=started_at, + latency_ms=round(latency_ms, 2), + ok=ok, + error=error, + error_kind=classify_error(error) if error else None, + markers=list(op.markers), + ) + + +def _run_writer( + *, + client: WarmMcpClient, + plan: list[PlannedOp], + project_name: str, + barrier: threading.Barrier, + deadline: float | None, + outcome: WriterOutcome, +) -> None: + barrier.wait() + for position, op in enumerate(plan): + if deadline is not None and time.monotonic() >= deadline: + outcome.not_attempted = len(plan) - position + return + result, terminal = _execute_op(client, op, project_name) + outcome.results.append(result) + if terminal: + outcome.terminal_error = result.error + outcome.not_attempted = len(plan) - position - 1 + return + + +# --- Integrity verification --- + + +@dataclass(frozen=True) +class ExpectedState: + """What the project must contain if every reported-success op converged.""" + + markers: frozenset[str] + ok_creates: int + hub_count: int + + +def _scan_markdown(project_dir: Path) -> tuple[int, Counter[str]]: + files = sorted(project_dir.rglob("*.md")) + markers: Counter[str] = Counter() + for path in files: + markers.update(MARKER_PATTERN.findall(path.read_text(encoding="utf-8"))) + return len(files), markers + + +def run_integrity_checks( + *, + db_path: Path, + project_name: str, + project_dir: Path, + expected: ExpectedState, +) -> IntegrityReport: + """Convergence verification against the on-disk files and the index DB. + + Fails loudly (raises) when the environment itself is broken — missing DB + file or missing project row — because that is a driver bug, not a + benchmark outcome. Product-level divergence is recorded in the report. + """ + if not db_path.exists(): + raise RuntimeError(f"Index database not found: {db_path}") + + md_files, found_markers = _scan_markdown(project_dir) + missing = sorted(expected.markers - set(found_markers)) + duplicated = sorted(marker for marker, count in found_markers.items() if count > 1) + + # Plain (not mode=ro) connection: the BM processes are stopped by now, and + # a read-only open can fail on a WAL database that still has -wal pages. + # This function issues SELECTs only. + connection = sqlite3.connect(db_path) + try: + row = connection.execute( + "SELECT id FROM project WHERE name = ?", (project_name,) + ).fetchone() + if row is None: + raise RuntimeError(f"Project '{project_name}' not found in {db_path}") + project_id = int(row[0]) + + entity_rows = int( + connection.execute( + "SELECT COUNT(*) FROM entity WHERE project_id = ?", (project_id,) + ).fetchone()[0] + ) + duplicate_permalinks = [ + str(r[0]) + for r in connection.execute( + "SELECT permalink FROM entity" + " WHERE project_id = ? AND permalink IS NOT NULL" + " GROUP BY permalink HAVING COUNT(*) > 1", + (project_id,), + ) + ] + observation_rows = int( + connection.execute( + "SELECT COUNT(*) FROM observation o JOIN entity e ON o.entity_id = e.id" + " WHERE e.project_id = ?", + (project_id,), + ).fetchone()[0] + ) + distinct_observation_tuples = int( + connection.execute( + "SELECT COUNT(*) FROM (SELECT DISTINCT o.entity_id, o.category, o.content" + " FROM observation o JOIN entity e ON o.entity_id = e.id" + " WHERE e.project_id = ?)", + (project_id,), + ).fetchone()[0] + ) + duplicate_observations = [ + f"entity={r[0]} [{r[1]}] x{r[3]}: {str(r[2])[:80]}" + for r in connection.execute( + "SELECT o.entity_id, o.category, o.content, COUNT(*)" + " FROM observation o JOIN entity e ON o.entity_id = e.id" + " WHERE e.project_id = ?" + " GROUP BY o.entity_id, o.category, o.content HAVING COUNT(*) > 1", + (project_id,), + ) + ] + duplicate_relation_tuples = int( + connection.execute( + "SELECT COUNT(*) FROM (" + " SELECT r.from_id, r.to_name, r.relation_type" + " FROM relation r JOIN entity e ON r.from_id = e.id" + " WHERE e.project_id = ?" + " GROUP BY r.from_id, r.to_name, r.relation_type HAVING COUNT(*) > 1)", + (project_id,), + ).fetchone()[0] + ) + finally: + connection.close() + + expected_files = expected.hub_count + expected.ok_creates + checks = [ + IntegrityCheck( + name="files_match_successful_creates", + passed=md_files == expected_files, + detail=f"{md_files} markdown files on disk, expected {expected_files}" + f" ({expected.hub_count} hubs + {expected.ok_creates} successful creates)", + ), + IntegrityCheck( + name="db_entities_match_files", + passed=entity_rows == md_files, + detail=f"{entity_rows} entity rows vs {md_files} markdown files", + ), + IntegrityCheck( + name="no_duplicate_permalinks", + passed=len(duplicate_permalinks) == 0, + detail=f"{len(duplicate_permalinks)} duplicated permalinks", + ), + IntegrityCheck( + name="no_duplicate_observation_tuples", + passed=len(duplicate_observations) == 0, + detail=f"{len(duplicate_observations)} duplicated (entity, category, content) tuples" + f" across {observation_rows} observation rows", + ), + IntegrityCheck( + name="no_duplicate_relation_tuples", + passed=duplicate_relation_tuples == 0, + detail=f"{duplicate_relation_tuples} duplicated (from, to, type) tuples", + ), + IntegrityCheck( + name="no_lost_writes", + passed=len(missing) == 0, + detail=f"{len(missing)} markers from successful ops missing on disk", + ), + IntegrityCheck( + name="no_doubled_writes", + passed=len(duplicated) == 0, + detail=f"{len(duplicated)} markers appear more than once on disk", + ), + ] + redundancy_pct = ( + 100.0 * (observation_rows - distinct_observation_tuples) / observation_rows + if observation_rows + else 0.0 + ) + return IntegrityReport( + checks=checks, + markdown_files=md_files, + entity_rows=entity_rows, + observation_rows=observation_rows, + distinct_observation_tuples=distinct_observation_tuples, + duplicate_observation_tuples=len(duplicate_observations), + observation_redundancy_pct=round(redundancy_pct, 2), + duplicate_permalinks=len(duplicate_permalinks), + duplicate_relation_tuples=duplicate_relation_tuples, + expected_markers=len(expected.markers), + found_markers=len(found_markers), + missing_markers=len(missing), + duplicated_markers=len(duplicated), + missing_marker_sample=missing[:10], + duplicate_permalink_sample=duplicate_permalinks[:10], + duplicate_observation_sample=duplicate_observations[:10], + converged=all(check.passed for check in checks), + ) + + +# --- Summaries and artifacts --- + + +def _percentile(sorted_values: list[float], quantile: float) -> float: + # Same nearest-rank convention as scoring/retrieval.py's p95. + if not sorted_values: + return 0.0 + index = max(0, min(len(sorted_values) - 1, math.ceil(len(sorted_values) * quantile) - 1)) + return sorted_values[index] + + +def summarize_op_type(results: list[OpResult]) -> OpTypeStats: + latencies = sorted(result.latency_ms for result in results) + ok = sum(1 for result in results if result.ok) + return OpTypeStats( + count=len(results), + ok=ok, + errors=len(results) - ok, + mean_ms=round(mean(latencies), 2) if latencies else 0.0, + p50_ms=round(_percentile(latencies, 0.50), 2), + p95_ms=round(_percentile(latencies, 0.95), 2), + max_ms=round(max(latencies), 2) if latencies else 0.0, + ) + + +def build_summary( + *, + config: ConcurrentWriteConfig, + results: list[OpResult], + outcomes: list[WriterOutcome], + concurrent_wall_seconds: float, + settle_seconds: float, + settle_mode: Literal["status-json", "fixed-delay"], + reindex_seconds: float | None, + integrity: IntegrityReport, +) -> ConcurrentWriteSummary: + ops_ok = sum(1 for result in results if result.ok) + error_kinds = Counter(result.error_kind for result in results if result.error_kind is not None) + per_op_type: dict[str, OpTypeStats] = {} + for op_type in ("create_hub", "create", "edit_hub", "edit_own"): + typed = [result for result in results if result.op_type == op_type] + if typed: + per_op_type[op_type] = summarize_op_type(typed) + notes_created_ok = sum(1 for r in results if r.op_type == "create" and r.ok) + return ConcurrentWriteSummary( + run_id=config.run_id, + concurrent_wall_seconds=round(concurrent_wall_seconds, 2), + settle_seconds=round(settle_seconds, 2), + settle_mode=settle_mode, + reindex_seconds=round(reindex_seconds, 2) if reindex_seconds is not None else None, + ops_total=len(results), + ops_ok=ops_ok, + ops_error=len(results) - ops_ok, + ops_not_attempted=sum(outcome.not_attempted for outcome in outcomes), + terminal_writer_failures=sum( + 1 for outcome in outcomes if outcome.terminal_error is not None + ), + error_kinds=dict(error_kinds), + per_op_type=per_op_type, + notes_created_ok=notes_created_ok, + creates_per_minute=round(notes_created_ok / (concurrent_wall_seconds / 60), 1) + if concurrent_wall_seconds > 0 + else 0.0, + ops_per_second=round(len(results) / concurrent_wall_seconds, 2) + if concurrent_wall_seconds > 0 + else 0.0, + integrity=integrity, + converged=integrity.converged, + ) + + +def build_summary_markdown( + manifest: ConcurrentWriteManifest, summary: ConcurrentWriteSummary +) -> str: + config = manifest.config + lines = [ + f"# Concurrent-Write Run `{manifest.run_id}`", + "", + "## Provenance", + "", + f"- Benchmark SHA: `{manifest.benchmark_git_sha}`", + f"- BM source: `{manifest.bm_source}`", + f"- BM resolved SHA: `{manifest.bm_resolved_sha or 'unknown'}`", + f"- BM version: `{manifest.bm_version or 'unknown'}`", + f"- Home: `{manifest.home_dir}`", + "", + "## Workload", + "", + f"- Writers: {config.writers} (one `bm mcp` session each)", + f"- Notes per writer: {config.notes_per_writer}", + f"- Edit ratio: {config.edit_ratio}, hub notes: {config.hub_notes}," + f" relation pool: {config.relation_pool}, seed: {config.seed}", + "", + "## Throughput", + "", + f"- Concurrent phase: {summary.concurrent_wall_seconds}s wall", + f"- Settle: {summary.settle_seconds}s ({summary.settle_mode})", + f"- Reindex: {f'{summary.reindex_seconds}s' if summary.reindex_seconds is not None else 'not measured'}", + f"- Ops: {summary.ops_total} total, {summary.ops_ok} ok, {summary.ops_error} errors," + f" {summary.ops_not_attempted} not attempted", + f"- Terminal writer failures: {summary.terminal_writer_failures}", + f"- Notes created: {summary.notes_created_ok}" + f" ({summary.creates_per_minute} notes/min, {summary.ops_per_second} ops/s overall)", + "", + "## Latency (ms)", + "", + "| Op | Count | OK | Errors | Mean | P50 | P95 | Max |", + "| --- | --- | --- | --- | --- | --- | --- | --- |", + ] + for op_type, stats in summary.per_op_type.items(): + lines.append( + f"| {op_type} | {stats.count} | {stats.ok} | {stats.errors} |" + f" {stats.mean_ms} | {stats.p50_ms} | {stats.p95_ms} | {stats.max_ms} |" + ) + if summary.error_kinds: + lines += ["", "## Errors", ""] + for kind, count in sorted(summary.error_kinds.items()): + lines.append(f"- {kind}: {count}") + integrity = summary.integrity + lines += [ + "", + f"## Convergence: {'CONVERGED' if summary.converged else 'DIVERGED'}", + "", + "| Check | Result | Detail |", + "| --- | --- | --- |", + ] + for check in integrity.checks: + lines.append(f"| {check.name} | {'pass' if check.passed else 'FAIL'} | {check.detail} |") + lines += [ + "", + f"- Observation redundancy: {integrity.observation_redundancy_pct}%" + f" ({integrity.observation_rows} rows, {integrity.distinct_observation_tuples} distinct)", + "", + "## Reproduce", + "", + "```bash", + f"uv run bm-bench run concurrent-write --run-id {manifest.run_id}" + f" --writers {config.writers} --notes-per-writer {config.notes_per_writer}" + f" --edit-ratio {config.edit_ratio} --hub-notes {config.hub_notes}" + f" --seed {config.seed} --bm-local-path {config.bm_local_path}", + "```", + ] + return "\n".join(lines).strip() + "\n" + + +def _write_json(path: Path, payload: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + + +def _write_jsonl(path: Path, rows: list[dict]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as file: + for row in rows: + file.write(json.dumps(row, sort_keys=True) + "\n") + + +def write_concurrent_artifacts( + *, + run_dir: Path, + manifest: ConcurrentWriteManifest, + results: list[OpResult], + summary: ConcurrentWriteSummary, +) -> None: + run_dir.mkdir(parents=True, exist_ok=True) + _write_json(run_dir / "manifest.json", manifest.model_dump(mode="json")) + _write_jsonl(run_dir / "per-op.jsonl", [row.model_dump(mode="json") for row in results]) + _write_json(run_dir / "concurrent-write-summary.json", summary.model_dump(mode="json")) + (run_dir / "summary.md").write_text(build_summary_markdown(manifest, summary), encoding="utf-8") + + +# --- Orchestration --- + + +def _isolated_env(home: Path) -> dict[str, str]: + """Benchmark-owned env: fresh config dir AND fresh default-project home. + + BASIC_MEMORY_HOME points inside the sandbox so the auto-created default + project indexes an empty directory instead of the operator's personal + notes (which would add noise to settle timing and DB contents). + """ + env = dict(os.environ) + env.pop("BASIC_MEMORY_CLOUD_MODE", None) + env["BASIC_MEMORY_CONFIG_DIR"] = str(home / "config") + env["BASIC_MEMORY_HOME"] = str(home / "default-home") + return env + + +def _bm_version(prefix: list[str], env: dict[str, str]) -> str | None: + try: + result = run_command(prefix + ["--version"], env=env) + except Exception: + return None + return result.stdout.strip() or None + + +def _settle_index( + *, + prefix: list[str], + env: dict[str, str], + project_name: str, + timeout_seconds: float, +) -> tuple[float, Literal["status-json", "fixed-delay"]]: + """Wait until the index reports no pending work; returns (seconds, mode).""" + start = time.monotonic() + probe = run_command(prefix + ["status", "--json", "--local"], check=False, env=env) + merged = ((probe.stdout or "") + "\n" + (probe.stderr or "")).lower() + if "no such option: --json" in merged: + # Old BM without --json: no readiness signal exists; give the watcher a + # fixed grace period and record the mode so the artifact is explicit. + time.sleep(10.0) + return time.monotonic() - start, "fixed-delay" + + deadline = start + timeout_seconds + delay = 0.25 + while True: + completed = run_command( + prefix + ["status", "--project", project_name, "--json", "--local"], env=env + ) + payload = json.loads(completed.stdout.strip() or "{}") + if isinstance(payload, dict) and status_json_is_ready(payload): + return time.monotonic() - start, "status-json" + if time.monotonic() >= deadline: + raise TimeoutError( + f"Index did not settle within {timeout_seconds}s for project '{project_name}'" + ) + time.sleep(delay) + delay = min(delay * 2, 2.0) + + +def run_concurrent_write(config: ConcurrentWriteConfig) -> Path: + """Execute the full driver: setup, concurrent phase, settle, verify, report.""" + bm_checkout = Path(config.bm_local_path).expanduser().resolve() + prefix = resolve_bm_command_prefix(str(bm_checkout)) + bm_resolved_sha = git_sha(bm_checkout) + if bm_resolved_sha is None: + raise ValueError("--bm-local-path must point to a Basic Memory git checkout") + config = config.model_copy(update={"bm_local_path": str(bm_checkout)}) + + home = Path("benchmarks/.bm-homes") / f"bm-write-{config.run_id}" + if home.exists(): + raise RuntimeError(f"Home already exists (re-running a run_id is not supported): {home}") + project_dir = home / "project" + project_dir.mkdir(parents=True) + (home / "default-home").mkdir() + env = _isolated_env(home) + + console.print(f"[bold]concurrent-write[/bold] run_id={config.run_id} home={home}") + run_command(prefix + ["project", "add", config.project_name, str(project_dir)], env=env) + bm_version = _bm_version(prefix, env) + + manifest = ConcurrentWriteManifest( + run_id=config.run_id, + created_at_utc=utc_now_iso(), + benchmark_git_sha=git_sha(Path(".")) or "unknown", + bm_source=config.bm_source, + bm_resolved_sha=bm_resolved_sha, + bm_local_path=config.bm_local_path, + bm_version=bm_version, + home_dir=str(home), + project_dir=str(project_dir), + project_name=config.project_name, + runtime=RuntimeInfo( + os=runtime_info()[0], + python_version=runtime_info()[1], + started_at_utc=utc_now_iso(), + ), + config=config, + ) + + mcp_command = prefix[0] + mcp_args = prefix[1:] + ["mcp"] + clients = [ + WarmMcpClient( + command=mcp_command, + args=mcp_args, + env=env, + request_timeout_seconds=config.op_timeout_seconds, + required_tool="write_note", + ) + for _ in range(config.writers) + ] + + results: list[OpResult] = [] + outcomes = [WriterOutcome() for _ in range(config.writers)] + try: + console.print(f"Starting {config.writers} warm `bm mcp` sessions...") + for client in clients: + client.start() + + # Setup phase: hubs are created sequentially through writer 0's session + # so the concurrent phase starts from a known shared state. + setup_results: list[OpResult] = [] + for op in build_hub_ops(config): + result, terminal = _execute_op(clients[0], op, config.project_name) + setup_results.append(result) + if not result.ok: + raise RuntimeError(f"Hub setup failed for {op.identifier}: {result.error}") + results.extend(setup_results) + + plans = [build_writer_plan(writer, config) for writer in range(config.writers)] + planned_ops = sum(len(plan) for plan in plans) + console.print( + f"Concurrent phase: {config.writers} writers, {planned_ops} planned ops" + f" ({config.notes_per_writer} creates each + edits)..." + ) + barrier = threading.Barrier(config.writers) + started = time.monotonic() + deadline = started + config.max_seconds if config.max_seconds is not None else None + threads = [ + threading.Thread( + target=_run_writer, + kwargs={ + "client": clients[writer], + "plan": plans[writer], + "project_name": config.project_name, + "barrier": barrier, + "deadline": deadline, + "outcome": outcomes[writer], + }, + name=f"bm-writer-{writer}", + ) + for writer in range(config.writers) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + concurrent_wall_seconds = time.monotonic() - started + for outcome in outcomes: + results.extend(outcome.results) + finally: + for client in clients: + client.stop() + + console.print("Waiting for index to settle...") + settle_seconds, settle_mode = _settle_index( + prefix=prefix, + env=env, + project_name=config.project_name, + timeout_seconds=config.settle_timeout_seconds, + ) + + ok_markers = frozenset(marker for result in results if result.ok for marker in result.markers) + ok_creates = sum(1 for result in results if result.op_type == "create" and result.ok) + ok_hubs = sum(1 for result in results if result.op_type == "create_hub" and result.ok) + integrity = run_integrity_checks( + db_path=home / "config" / "memory.db", + project_name=config.project_name, + project_dir=project_dir, + expected=ExpectedState(markers=ok_markers, ok_creates=ok_creates, hub_count=ok_hubs), + ) + + # Trigger: reindex is enabled as a separate timing measurement. + # Why: reindex mutates derived state and could hide or introduce the + # concurrency outcome. Outcome: the convergence verdict above remains the + # settled pre-reindex state; only wall time is measured afterward. + reindex_seconds: float | None = None + if config.measure_reindex: + console.print("Measuring full reindex wall time...") + reindex_start = time.monotonic() + run_command(prefix + ["reindex", "--search", "-p", config.project_name], env=env) + reindex_seconds = time.monotonic() - reindex_start + + summary = build_summary( + config=config, + results=results, + outcomes=outcomes, + concurrent_wall_seconds=concurrent_wall_seconds, + settle_seconds=settle_seconds, + settle_mode=settle_mode, + reindex_seconds=reindex_seconds, + integrity=integrity, + ) + + run_dir = Path(config.output_root) / config.run_id + write_concurrent_artifacts(run_dir=run_dir, manifest=manifest, results=results, summary=summary) + + verdict = "[green]CONVERGED[/green]" if summary.converged else "[red]DIVERGED[/red]" + console.print( + f"{verdict} — {summary.ops_ok}/{summary.ops_total} ops ok," + f" {summary.notes_created_ok} notes at {summary.creates_per_minute} notes/min," + f" settle {summary.settle_seconds}s" + ) + for check in integrity.checks: + status = "[green]pass[/green]" if check.passed else "[red]FAIL[/red]" + console.print(f" {status} {check.name}: {check.detail}") + return run_dir diff --git a/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py b/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py index c71016dd0..7c7ed879c 100644 --- a/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py +++ b/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py @@ -2,120 +2,27 @@ from __future__ import annotations -import asyncio import json import os import re import subprocess import tempfile -import threading import time -from concurrent.futures import Future -from dataclasses import dataclass from pathlib import Path -from queue import Queue from typing import Any, cast -import anyio -from mcp.client.session import ClientSession -from mcp.client.stdio import StdioServerParameters, stdio_client from mcp.types import CallToolResult +from basic_memory_benchmarks.bm_runtime import ( + WarmMcpClient, + resolve_bm_command_prefix, + status_json_is_ready, +) from basic_memory_benchmarks.models import RunConfig, SearchHit from basic_memory_benchmarks.providers.base import BenchmarkProvider from basic_memory_benchmarks.utils import run_command -@dataclass -class _McpToolRequest: - name: str - arguments: dict[str, Any] - response: Future[CallToolResult] - - -class _WarmMcpClient: - def __init__( - self, - *, - command: str = "bm", - args: list[str] | None = None, - env: dict[str, str] | None = None, - startup_timeout_seconds: float = 30.0, - request_timeout_seconds: float = 60.0, - ) -> None: - self._command = command - self._args = args or ["mcp"] - self._env = env - self._startup_timeout_seconds = startup_timeout_seconds - self._request_timeout_seconds = request_timeout_seconds - self._requests: Queue[_McpToolRequest | None] = Queue() - self._ready = threading.Event() - self._startup_error: Exception | None = None - self._thread: threading.Thread | None = None - - async def _serve(self) -> None: - params = StdioServerParameters(command=self._command, args=self._args, env=self._env) - async with stdio_client(params) as (read_stream, write_stream): - async with ClientSession(read_stream, write_stream) as session: - await session.initialize() - tools = await session.list_tools() - tool_names = {tool.name for tool in tools.tools} - if "search_notes" not in tool_names: - raise RuntimeError("bm mcp server does not expose 'search_notes'") - - self._ready.set() - - while True: - loop = asyncio.get_running_loop() - request = await loop.run_in_executor(None, self._requests.get) - if request is None: - break - try: - result = await session.call_tool(request.name, request.arguments) - except Exception as exc: - request.response.set_exception(exc) - else: - request.response.set_result(result) - - def _thread_main(self) -> None: - try: - anyio.run(self._serve) - except Exception as exc: - self._startup_error = exc - self._ready.set() - - def start(self) -> None: - if self._thread is not None and self._thread.is_alive(): - return - self._thread = threading.Thread( - target=self._thread_main, - name="bm-benchmark-mcp-client", - daemon=True, - ) - self._thread.start() - - if not self._ready.wait(timeout=self._startup_timeout_seconds): - raise TimeoutError("Timed out starting bm mcp session") - if self._startup_error is not None: - raise RuntimeError("Failed to start bm mcp session") from self._startup_error - - def call_tool(self, name: str, arguments: dict[str, Any]) -> CallToolResult: - if self._thread is None or not self._thread.is_alive(): - raise RuntimeError("bm mcp session is not running") - - response: Future[CallToolResult] = Future() - self._requests.put(_McpToolRequest(name=name, arguments=arguments, response=response)) - return response.result(timeout=self._request_timeout_seconds) - - def stop(self) -> None: - if self._thread is None: - return - if self._thread.is_alive(): - self._requests.put(None) - self._thread.join(timeout=self._startup_timeout_seconds) - self._thread = None - - class BasicMemoryLocalProvider(BenchmarkProvider): name = "bm-local" # One instance serves every group in a grouped run: the warm MCP session @@ -126,7 +33,7 @@ def __init__(self) -> None: # run_id -> resolved project name (grouped runs ingest many projects). self._resolved_project_names: dict[str, str] = {} self._status_json_supported: bool | None = None - self._mcp: _WarmMcpClient | None = None + self._mcp: WarmMcpClient | None = None self._bm_command_prefix: list[str] = ["bm"] self._bm_env: dict[str, str] | None = None self._config_dir: Path | None = None @@ -159,12 +66,7 @@ def _project_name(self, run_config: RunConfig) -> str: @staticmethod def _resolve_bm_command_prefix(run_config: RunConfig) -> list[str]: - if run_config.bm_local_path: - local_path = Path(run_config.bm_local_path) - if not local_path.exists(): - raise ValueError(f"--bm-local-path not found: {local_path}") - return ["uv", "run", "--project", str(local_path), "basic-memory"] - return ["bm"] + return resolve_bm_command_prefix(run_config.bm_local_path) def _run_bm( self, @@ -183,40 +85,7 @@ def _extract_existing_project_name(message: str) -> str | None: @staticmethod def _status_json_is_ready(payload: dict[str, Any]) -> bool: - total = payload.get("total") - if isinstance(total, int): - return total == 0 - - for list_key in ("new", "modified", "deleted", "skipped_files"): - value = payload.get(list_key) - if isinstance(value, list) and len(value) > 0: - return False - - for dict_key in ("moves", "checksums"): - value = payload.get(dict_key) - if isinstance(value, dict) and len(value) > 0: - return False - - status = payload.get("status") - if isinstance(status, str): - lowered = status.lower() - if "no changes" in lowered or "up to date" in lowered: - return True - if "sync" in lowered or "index" in lowered or "pending" in lowered: - return False - - for key in ("is_syncing", "is_indexing", "sync_in_progress", "index_in_progress"): - value = payload.get(key) - if isinstance(value, bool): - return not value - - for key in ("pending_files", "pending", "unindexed_files", "queued_files", "queue_size"): - value = payload.get(key) - if isinstance(value, int) and value != 0: - return False - - # If the schema is unknown and no busy signal exists, treat status as ready. - return True + return status_json_is_ready(payload) @staticmethod def _error_text_from_result(result: CallToolResult) -> str: @@ -334,7 +203,7 @@ def ingest(self, corpus_path: Path, run_config: RunConfig) -> None: if self._mcp is None: mcp_command = self._bm_command_prefix[0] mcp_args = self._bm_command_prefix[1:] + ["mcp"] - self._mcp = _WarmMcpClient(command=mcp_command, args=mcp_args, env=self._bm_env) + self._mcp = WarmMcpClient(command=mcp_command, args=mcp_args, env=self._bm_env) self._mcp.start() @staticmethod diff --git a/benchmarks/tests/test_concurrent_write.py b/benchmarks/tests/test_concurrent_write.py new file mode 100644 index 000000000..e16ead2eb --- /dev/null +++ b/benchmarks/tests/test_concurrent_write.py @@ -0,0 +1,420 @@ +"""Unit tests for the concurrent-writer driver (planning, stats, integrity).""" + +from __future__ import annotations + +import asyncio +import sqlite3 +import threading +from concurrent.futures import TimeoutError as FutureTimeoutError +from contextlib import asynccontextmanager +from pathlib import Path +from types import SimpleNamespace +from typing import Any, AsyncIterator + +import pytest + +import basic_memory_benchmarks.bm_runtime as bm_runtime +from basic_memory_benchmarks.bm_runtime import WarmMcpClient +from basic_memory_benchmarks.concurrent_write import ( + ConcurrentWriteConfig, + ExpectedState, + OpResult, + build_hub_ops, + build_summary_markdown, + build_writer_plan, + classify_error, + run_integrity_checks, + summarize_op_type, +) + + +def _config(**overrides: object) -> ConcurrentWriteConfig: + defaults: dict[str, object] = { + "run_id": "test-run", + "writers": 3, + "notes_per_writer": 10, + "edit_ratio": 0.5, + "hub_notes": 2, + "relation_pool": 4, + "seed": 42, + "bm_local_path": "/tmp/basic-memory", + } + defaults.update(overrides) + return ConcurrentWriteConfig.model_validate(defaults) + + +# --- Workload planning --- + + +def test_writer_plan_is_deterministic() -> None: + config = _config() + first = build_writer_plan(1, config) + second = build_writer_plan(1, config) + assert first == second + + +def test_writer_plans_differ_between_writers() -> None: + config = _config() + assert build_writer_plan(0, config) != build_writer_plan(1, config) + + +def test_markers_are_unique_across_writers_and_hubs() -> None: + config = _config() + markers: list[str] = [] + for op in build_hub_ops(config): + markers.extend(op.markers) + for writer in range(config.writers): + for op in build_writer_plan(writer, config): + markers.extend(op.markers) + assert len(markers) == len(set(markers)) + + +def test_plan_contains_expected_creates_and_valid_edit_targets() -> None: + config = _config() + plan = build_writer_plan(2, config) + creates = [op for op in plan if op.op_type == "create"] + assert len(creates) == config.notes_per_writer + + created_identifiers = {op.identifier for op in creates} + for op in plan: + if op.op_type == "edit_hub": + hub_index = int(op.identifier.rsplit("-", 1)[-1]) + assert 0 <= hub_index < config.hub_notes + if op.op_type == "edit_own": + # Writers only edit their own already-created notes. + assert op.identifier in created_identifiers + position = plan.index(op) + assert op.identifier in {p.identifier for p in plan[:position] if p.op_type == "create"} + + +def test_zero_hub_notes_produces_no_hub_ops() -> None: + config = _config(hub_notes=0) + assert build_hub_ops(config) == [] + for writer in range(config.writers): + assert all(op.op_type != "edit_hub" for op in build_writer_plan(writer, config)) + + +# --- MCP process lifecycle --- + + +def test_timed_out_mcp_call_is_cancelled_before_stop_returns( + monkeypatch: pytest.MonkeyPatch, +) -> None: + call_started = threading.Event() + transport_exited = threading.Event() + + @asynccontextmanager + async def fake_stdio_client(_params: object) -> AsyncIterator[tuple[object, object]]: + try: + yield object(), object() + finally: + transport_exited.set() + + class FakeClientSession: + def __init__(self, _read_stream: object, _write_stream: object) -> None: + pass + + async def __aenter__(self) -> FakeClientSession: + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + async def initialize(self) -> None: + return None + + async def list_tools(self) -> SimpleNamespace: + return SimpleNamespace(tools=[SimpleNamespace(name="write_note")]) + + async def call_tool(self, _name: str, _arguments: dict[str, Any]) -> Any: + call_started.set() + await asyncio.Event().wait() + + monkeypatch.setattr(bm_runtime, "stdio_client", fake_stdio_client) + monkeypatch.setattr(bm_runtime, "ClientSession", FakeClientSession) + + client = WarmMcpClient( + request_timeout_seconds=0.01, + startup_timeout_seconds=1.0, + required_tool="write_note", + ) + client.start() + with pytest.raises(FutureTimeoutError): + client.call_tool("write_note", {}) + assert call_started.wait(timeout=1.0) + + client.stop() + + assert transport_exited.wait(timeout=1.0) + with pytest.raises(RuntimeError, match="not running"): + client.call_tool("write_note", {}) + + +# --- Error classification and stats --- + + +@pytest.mark.parametrize( + ("text", "kind"), + [ + ("deadlock detected on relation", "deadlock"), + ("sqlite3.OperationalError: database is locked", "sqlite_locked"), + ("tool call timed out after 120s", "timeout"), + ("db_version conflict: stale write rejected", "write_conflict"), + ("something else entirely", "other"), + ], +) +def test_classify_error(text: str, kind: str) -> None: + assert classify_error(text) == kind + + +def test_summarize_op_type_latency_stats() -> None: + results = [ + OpResult( + writer=0, + op_index=i, + op_type="create", + identifier=f"notes/n{i}", + started_at_utc="2026-01-01T00:00:00Z", + latency_ms=float(latency), + ok=(i != 3), + error="boom" if i == 3 else None, + ) + for i, latency in enumerate([10, 20, 30, 40, 100]) + ] + stats = summarize_op_type(results) + assert stats.count == 5 + assert stats.ok == 4 + assert stats.errors == 1 + assert stats.mean_ms == 40.0 + assert stats.p50_ms == 30.0 + assert stats.max_ms == 100.0 + + +# --- Integrity verification --- + + +SCHEMA = """ +CREATE TABLE project (id INTEGER PRIMARY KEY, name TEXT); +CREATE TABLE entity (id INTEGER PRIMARY KEY, project_id INTEGER, permalink TEXT, file_path TEXT); +CREATE TABLE observation (id INTEGER PRIMARY KEY, entity_id INTEGER, category TEXT, content TEXT); +CREATE TABLE relation ( + id INTEGER PRIMARY KEY, from_id INTEGER, to_id INTEGER, to_name TEXT, relation_type TEXT +); +""" + + +def _make_db(path: Path, rows: dict[str, list[tuple]]) -> None: + connection = sqlite3.connect(path) + connection.executescript(SCHEMA) + for table, table_rows in rows.items(): + if not table_rows: + continue + placeholders = ",".join("?" for _ in table_rows[0]) + connection.executemany(f"INSERT INTO {table} VALUES ({placeholders})", table_rows) + connection.commit() + connection.close() + + +def _write_note_file(project_dir: Path, relative: str, markers: list[str]) -> None: + path = project_dir / relative + path.parent.mkdir(parents=True, exist_ok=True) + lines = [f"- [fact] generated content {marker}" for marker in markers] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def test_integrity_clean_state_converges(tmp_path: Path) -> None: + project_dir = tmp_path / "project" + _write_note_file(project_dir, "hubs/hub-0.md", ["bmk-setup-h00-l0"]) + _write_note_file(project_dir, "notes/w00-n0000.md", ["bmk-w00-o0000-l0"]) + db_path = tmp_path / "memory.db" + _make_db( + db_path, + { + "project": [(1, "bm-write-test")], + "entity": [ + (1, 1, "hubs/hub-0", "hubs/hub-0.md"), + (2, 1, "notes/w00-n0000", "notes/w00-n0000.md"), + ], + "observation": [ + (1, 1, "fact", "generated content bmk-setup-h00-l0"), + (2, 2, "fact", "generated content bmk-w00-o0000-l0"), + ], + "relation": [(1, 2, None, "topic-1", "relates_to")], + }, + ) + report = run_integrity_checks( + db_path=db_path, + project_name="bm-write-test", + project_dir=project_dir, + expected=ExpectedState( + markers=frozenset({"bmk-setup-h00-l0", "bmk-w00-o0000-l0"}), + ok_creates=1, + hub_count=1, + ), + ) + assert report.converged + assert report.markdown_files == 2 + assert report.entity_rows == 2 + assert report.observation_redundancy_pct == 0.0 + + +def test_integrity_flags_duplicate_observation_tuples(tmp_path: Path) -> None: + project_dir = tmp_path / "project" + _write_note_file(project_dir, "notes/w00-n0000.md", ["bmk-w00-o0000-l0"]) + db_path = tmp_path / "memory.db" + _make_db( + db_path, + { + "project": [(1, "bm-write-test")], + "entity": [(1, 1, "notes/w00-n0000", "notes/w00-n0000.md")], + # The #1214 shape: the same observation indexed twice. + "observation": [ + (1, 1, "fact", "generated content bmk-w00-o0000-l0"), + (2, 1, "fact", "generated content bmk-w00-o0000-l0"), + ], + }, + ) + report = run_integrity_checks( + db_path=db_path, + project_name="bm-write-test", + project_dir=project_dir, + expected=ExpectedState(markers=frozenset({"bmk-w00-o0000-l0"}), ok_creates=1, hub_count=0), + ) + assert not report.converged + assert report.duplicate_observation_tuples == 1 + assert report.observation_redundancy_pct == 50.0 + failed = {check.name for check in report.checks if not check.passed} + assert failed == {"no_duplicate_observation_tuples"} + + +def test_integrity_flags_lost_write_and_duplicate_permalink(tmp_path: Path) -> None: + project_dir = tmp_path / "project" + # Marker bmk-w00-o0001-l0 was reported ok but never landed on disk. + _write_note_file(project_dir, "notes/w00-n0000.md", ["bmk-w00-o0000-l0"]) + db_path = tmp_path / "memory.db" + _make_db( + db_path, + { + "project": [(1, "bm-write-test")], + "entity": [ + (1, 1, "notes/w00-n0000", "notes/w00-n0000.md"), + # Duplicate permalink row. + (2, 1, "notes/w00-n0000", "notes/w00-n0000 (copy).md"), + ], + "observation": [(1, 1, "fact", "generated content bmk-w00-o0000-l0")], + }, + ) + report = run_integrity_checks( + db_path=db_path, + project_name="bm-write-test", + project_dir=project_dir, + expected=ExpectedState( + markers=frozenset({"bmk-w00-o0000-l0", "bmk-w00-o0001-l0"}), + ok_creates=2, + hub_count=0, + ), + ) + assert not report.converged + assert report.missing_markers == 1 + assert report.missing_marker_sample == ["bmk-w00-o0001-l0"] + assert report.duplicate_permalinks == 1 + failed = {check.name for check in report.checks if not check.passed} + assert "no_lost_writes" in failed + assert "no_duplicate_permalinks" in failed + assert "files_match_successful_creates" in failed + + +def test_integrity_missing_db_fails_loudly(tmp_path: Path) -> None: + with pytest.raises(RuntimeError, match="Index database not found"): + run_integrity_checks( + db_path=tmp_path / "missing.db", + project_name="bm-write-test", + project_dir=tmp_path, + expected=ExpectedState(markers=frozenset(), ok_creates=0, hub_count=0), + ) + + +def test_integrity_missing_project_fails_loudly(tmp_path: Path) -> None: + db_path = tmp_path / "memory.db" + _make_db(db_path, {"project": [(1, "some-other-project")]}) + with pytest.raises(RuntimeError, match="not found"): + run_integrity_checks( + db_path=db_path, + project_name="bm-write-test", + project_dir=tmp_path, + expected=ExpectedState(markers=frozenset(), ok_creates=0, hub_count=0), + ) + + +# --- Summary rendering --- + + +def test_summary_markdown_reports_convergence(tmp_path: Path) -> None: + from basic_memory_benchmarks.concurrent_write import ( + ConcurrentWriteManifest, + WriterOutcome, + build_summary, + ) + from basic_memory_benchmarks.models import RuntimeInfo + + config = _config() + project_dir = tmp_path / "project" + _write_note_file(project_dir, "notes/w00-n0000.md", ["bmk-w00-o0000-l0"]) + db_path = tmp_path / "memory.db" + _make_db( + db_path, + { + "project": [(1, config.project_name)], + "entity": [(1, 1, "notes/w00-n0000", "notes/w00-n0000.md")], + "observation": [(1, 1, "fact", "generated content bmk-w00-o0000-l0")], + }, + ) + integrity = run_integrity_checks( + db_path=db_path, + project_name=config.project_name, + project_dir=project_dir, + expected=ExpectedState(markers=frozenset({"bmk-w00-o0000-l0"}), ok_creates=1, hub_count=0), + ) + results = [ + OpResult( + writer=0, + op_index=0, + op_type="create", + identifier="notes/w00-n0000", + started_at_utc="2026-01-01T00:00:00Z", + latency_ms=12.0, + ok=True, + markers=["bmk-w00-o0000-l0"], + ) + ] + summary = build_summary( + config=config, + results=results, + outcomes=[WriterOutcome(results=results)], + concurrent_wall_seconds=1.5, + settle_seconds=0.5, + settle_mode="status-json", + reindex_seconds=2.0, + integrity=integrity, + ) + manifest = ConcurrentWriteManifest( + run_id=config.run_id, + created_at_utc="2026-01-01T00:00:00Z", + benchmark_git_sha="abc123", + bm_source="local-checkout", + bm_resolved_sha="def456", + bm_local_path=config.bm_local_path, + home_dir=str(tmp_path), + project_dir=str(project_dir), + project_name=config.project_name, + runtime=RuntimeInfo( + os="test", python_version="3.12", started_at_utc="2026-01-01T00:00:00Z" + ), + config=config, + ) + markdown = build_summary_markdown(manifest, summary) + assert "## Convergence: CONVERGED" in markdown + assert "no_duplicate_observation_tuples" in markdown + assert "bm-bench run concurrent-write" in markdown + assert summary.converged + assert summary.notes_created_ok == 1 From 365bf175bb58134cacdda0de2adcbe719812e04f Mon Sep 17 00:00:00 2001 From: phernandez Date: Sun, 23 Aug 2026 11:57:22 -0500 Subject: [PATCH 2/3] fix(benchmarks): harden concurrent-write verdicts Signed-off-by: phernandez --- .../src/basic_memory_benchmarks/bm_runtime.py | 32 +++++--- .../concurrent_write.py | 79 +++++++++++++++---- .../providers/bm_local.py | 12 +-- .../tests/providers/test_bm_local_provider.py | 5 ++ benchmarks/tests/test_concurrent_write.py | 74 +++++++++++++++++ 5 files changed, 169 insertions(+), 33 deletions(-) diff --git a/benchmarks/src/basic_memory_benchmarks/bm_runtime.py b/benchmarks/src/basic_memory_benchmarks/bm_runtime.py index 0e6762d8b..8038d1937 100644 --- a/benchmarks/src/basic_memory_benchmarks/bm_runtime.py +++ b/benchmarks/src/basic_memory_benchmarks/bm_runtime.py @@ -184,25 +184,30 @@ def resolve_bm_command_prefix(bm_local_path: str | None) -> list[str]: return ["bm"] -def status_json_is_ready(payload: dict[str, Any]) -> bool: +def status_json_is_ready(payload: dict[str, Any]) -> bool | None: """Interpret `bm status --json` output across BM versions. - The schema varies by version; every known busy signal is checked, and an - unknown schema with no busy signal counts as ready. + Returns ``True`` when a known schema is idle, ``False`` when it is busy, + and ``None`` when the payload has no supported readiness signal. """ total = payload.get("total") if isinstance(total, int): return total == 0 + recognized_signal = False for list_key in ("new", "modified", "deleted", "skipped_files"): value = payload.get(list_key) - if isinstance(value, list) and len(value) > 0: - return False + if isinstance(value, list): + recognized_signal = True + if value: + return False for dict_key in ("moves", "checksums"): value = payload.get(dict_key) - if isinstance(value, dict) and len(value) > 0: - return False + if isinstance(value, dict): + recognized_signal = True + if value: + return False status = payload.get("status") if isinstance(status, str): @@ -215,12 +220,15 @@ def status_json_is_ready(payload: dict[str, Any]) -> bool: for key in ("is_syncing", "is_indexing", "sync_in_progress", "index_in_progress"): value = payload.get(key) if isinstance(value, bool): - return not value + recognized_signal = True + if value: + return False for key in ("pending_files", "pending", "unindexed_files", "queued_files", "queue_size"): value = payload.get(key) - if isinstance(value, int) and value != 0: - return False + if isinstance(value, int): + recognized_signal = True + if value != 0: + return False - # If the schema is unknown and no busy signal exists, treat status as ready. - return True + return True if recognized_signal else None diff --git a/benchmarks/src/basic_memory_benchmarks/concurrent_write.py b/benchmarks/src/basic_memory_benchmarks/concurrent_write.py index 6db1dc7f6..f273bd47d 100644 --- a/benchmarks/src/basic_memory_benchmarks/concurrent_write.py +++ b/benchmarks/src/basic_memory_benchmarks/concurrent_write.py @@ -39,6 +39,7 @@ from statistics import mean from typing import Any, Literal +from mcp.types import CallToolResult from pydantic import BaseModel, Field from rich.console import Console @@ -55,6 +56,7 @@ OpType = Literal["create_hub", "create", "edit_hub", "edit_own"] MARKER_PATTERN = re.compile(r"bmk-[a-z0-9-]+") +FALLBACK_SETTLE_SECONDS = 10.0 # --- Configuration and artifact models --- @@ -341,15 +343,54 @@ def _tool_call_for(op: PlannedOp, project_name: str) -> tuple[str, dict[str, Any "directory": op.directory, "content": op.content, "project": project_name, + "output_format": "json", } return "edit_note", { "identifier": op.identifier, "operation": "append", "content": op.content, "project": project_name, + "output_format": "json", } +def _tool_result_payload(result: CallToolResult) -> dict[str, Any]: + structured = result.structuredContent + if isinstance(structured, dict): + wrapped = structured.get("result") + if isinstance(wrapped, dict): + return wrapped + return structured + + for item in result.content: + text = getattr(item, "text", None) + if not isinstance(text, str): + continue + try: + parsed = json.loads(text) + except json.JSONDecodeError: + continue + if isinstance(parsed, dict): + return parsed + return {} + + +def tool_result_error(result: CallToolResult) -> str | None: + """Return the MCP or tool-level error, including JSON responses with HTTP failures.""" + if result.isError: + for item in result.content: + text = getattr(item, "text", None) + if isinstance(text, str) and text.strip(): + return text.strip() + return "Unknown MCP tool error" + + payload = _tool_result_payload(result) + if not payload: + return "MCP tool returned no JSON payload" + error = payload.get("error") + return str(error) if error else None + + def _execute_op(client: WarmMcpClient, op: PlannedOp, project_name: str) -> tuple[OpResult, bool]: """Run one op; returns (result, terminal) — terminal means the session is unusable. @@ -377,13 +418,8 @@ def _execute_op(client: WarmMcpClient, op: PlannedOp, project_name: str) -> tupl session_dead, ) latency_ms = (time.perf_counter() - start) * 1000 - if result.isError: - error = "Unknown MCP tool error" - for item in result.content: - text = getattr(item, "text", None) - if isinstance(text, str) and text.strip(): - error = text.strip() - break + error = tool_result_error(result) + if error is not None: return _op_result(op, started_at, latency_ms, ok=False, error=error), False return _op_result(op, started_at, latency_ms, ok=True, error=None), False @@ -634,6 +670,9 @@ def build_summary( integrity: IntegrityReport, ) -> ConcurrentWriteSummary: ops_ok = sum(1 for result in results if result.ok) + ops_error = len(results) - ops_ok + ops_not_attempted = sum(outcome.not_attempted for outcome in outcomes) + terminal_writer_failures = sum(1 for outcome in outcomes if outcome.terminal_error is not None) error_kinds = Counter(result.error_kind for result in results if result.error_kind is not None) per_op_type: dict[str, OpTypeStats] = {} for op_type in ("create_hub", "create", "edit_hub", "edit_own"): @@ -649,11 +688,9 @@ def build_summary( reindex_seconds=round(reindex_seconds, 2) if reindex_seconds is not None else None, ops_total=len(results), ops_ok=ops_ok, - ops_error=len(results) - ops_ok, - ops_not_attempted=sum(outcome.not_attempted for outcome in outcomes), - terminal_writer_failures=sum( - 1 for outcome in outcomes if outcome.terminal_error is not None - ), + ops_error=ops_error, + ops_not_attempted=ops_not_attempted, + terminal_writer_failures=terminal_writer_failures, error_kinds=dict(error_kinds), per_op_type=per_op_type, notes_created_ok=notes_created_ok, @@ -664,7 +701,12 @@ def build_summary( if concurrent_wall_seconds > 0 else 0.0, integrity=integrity, - converged=integrity.converged, + converged=( + integrity.converged + and ops_error == 0 + and ops_not_attempted == 0 + and terminal_writer_failures == 0 + ), ) @@ -807,7 +849,7 @@ def _settle_index( if "no such option: --json" in merged: # Old BM without --json: no readiness signal exists; give the watcher a # fixed grace period and record the mode so the artifact is explicit. - time.sleep(10.0) + time.sleep(FALLBACK_SETTLE_SECONDS) return time.monotonic() - start, "fixed-delay" deadline = start + timeout_seconds @@ -817,8 +859,13 @@ def _settle_index( prefix + ["status", "--project", project_name, "--json", "--local"], env=env ) payload = json.loads(completed.stdout.strip() or "{}") - if isinstance(payload, dict) and status_json_is_ready(payload): - return time.monotonic() - start, "status-json" + if isinstance(payload, dict): + readiness = status_json_is_ready(payload) + if readiness is None: + time.sleep(FALLBACK_SETTLE_SECONDS) + return time.monotonic() - start, "fixed-delay" + if readiness: + return time.monotonic() - start, "status-json" if time.monotonic() >= deadline: raise TimeoutError( f"Index did not settle within {timeout_seconds}s for project '{project_name}'" diff --git a/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py b/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py index 7c7ed879c..7ce4bb13a 100644 --- a/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py +++ b/benchmarks/src/basic_memory_benchmarks/providers/bm_local.py @@ -84,7 +84,7 @@ def _extract_existing_project_name(message: str) -> str | None: return None @staticmethod - def _status_json_is_ready(payload: dict[str, Any]) -> bool: + def _status_json_is_ready(payload: dict[str, Any]) -> bool | None: return status_json_is_ready(payload) @staticmethod @@ -149,10 +149,12 @@ def _wait_for_index_ready(self, project_name: str) -> None: ] ) payload = json.loads(completed.stdout.strip() or "{}") - if isinstance(payload, dict) and self._status_json_is_ready( - cast(dict[str, Any], payload) - ): - return + if isinstance(payload, dict): + readiness = self._status_json_is_ready(cast(dict[str, Any], payload)) + if readiness is None: + return + if readiness: + return if time.monotonic() >= deadline: raise TimeoutError( diff --git a/benchmarks/tests/providers/test_bm_local_provider.py b/benchmarks/tests/providers/test_bm_local_provider.py index 4ed3fe5b7..4a1f05884 100644 --- a/benchmarks/tests/providers/test_bm_local_provider.py +++ b/benchmarks/tests/providers/test_bm_local_provider.py @@ -104,6 +104,11 @@ def test_status_json_is_not_ready_when_indexing() -> None: assert not BasicMemoryLocalProvider._status_json_is_ready(payload) +def test_status_json_unknown_schema_is_unsupported() -> None: + payload = {"total_files": 1, "observed_files": [{"path": "note.md"}]} + assert BasicMemoryLocalProvider._status_json_is_ready(payload) is None + + def test_resolve_bm_command_prefix_default_uses_bm() -> None: run_config = RunConfig( run_id="r1", diff --git a/benchmarks/tests/test_concurrent_write.py b/benchmarks/tests/test_concurrent_write.py index e16ead2eb..8832ede64 100644 --- a/benchmarks/tests/test_concurrent_write.py +++ b/benchmarks/tests/test_concurrent_write.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import json import sqlite3 import threading from concurrent.futures import TimeoutError as FutureTimeoutError @@ -12,8 +13,10 @@ from typing import Any, AsyncIterator import pytest +from mcp.types import CallToolResult, TextContent import basic_memory_benchmarks.bm_runtime as bm_runtime +import basic_memory_benchmarks.concurrent_write as concurrent_write from basic_memory_benchmarks.bm_runtime import WarmMcpClient from basic_memory_benchmarks.concurrent_write import ( ConcurrentWriteConfig, @@ -25,6 +28,7 @@ classify_error, run_integrity_checks, summarize_op_type, + tool_result_error, ) @@ -190,6 +194,53 @@ def test_summarize_op_type_latency_stats() -> None: assert stats.max_ms == 100.0 +def test_tool_result_error_reads_structured_tool_failure() -> None: + result = CallToolResult( + content=[TextContent(type="text", text='{"error":"deadlock detected"}')], + structuredContent={"result": {"error": "deadlock detected"}}, + isError=False, + ) + assert tool_result_error(result) == "deadlock detected" + + +def test_tool_result_error_accepts_structured_success() -> None: + result = CallToolResult( + content=[TextContent(type="text", text='{"title":"note"}')], + structuredContent={"result": {"title": "note"}}, + isError=False, + ) + assert tool_result_error(result) is None + + +def test_concurrent_ops_request_json_tool_responses() -> None: + op = build_writer_plan(0, _config(notes_per_writer=1))[0] + _tool, arguments = concurrent_write._tool_call_for(op, "project") + assert arguments["output_format"] == "json" + + +def test_unknown_status_schema_uses_fixed_settle_delay( + monkeypatch: pytest.MonkeyPatch, +) -> None: + status = json.dumps({"total_files": 2, "observed_files": [{"path": "note.md"}]}) + monkeypatch.setattr( + concurrent_write, + "run_command", + lambda *_args, **_kwargs: SimpleNamespace(stdout=status, stderr=""), + ) + delays: list[float] = [] + monkeypatch.setattr(concurrent_write.time, "sleep", delays.append) + + _seconds, mode = concurrent_write._settle_index( + prefix=["bm"], + env={}, + project_name="project", + timeout_seconds=1.0, + ) + + assert mode == "fixed-delay" + assert delays == [concurrent_write.FALLBACK_SETTLE_SECONDS] + + # --- Integrity verification --- @@ -418,3 +469,26 @@ def test_summary_markdown_reports_convergence(tmp_path: Path) -> None: assert "bm-bench run concurrent-write" in markdown assert summary.converged assert summary.notes_created_ok == 1 + + failed_result = OpResult( + writer=0, + op_index=1, + op_type="edit_own", + identifier="notes/w00-n0000", + started_at_utc="2026-01-01T00:00:01Z", + latency_ms=15.0, + ok=False, + error="deadlock detected", + error_kind="deadlock", + ) + failed_summary = build_summary( + config=config, + results=[*results, failed_result], + outcomes=[WriterOutcome(results=[*results, failed_result])], + concurrent_wall_seconds=1.5, + settle_seconds=0.5, + settle_mode="status-json", + reindex_seconds=None, + integrity=integrity, + ) + assert not failed_summary.converged From b60830c250563b09b879a87e1f18d286f35440cd Mon Sep 17 00:00:00 2001 From: phernandez Date: Sun, 23 Aug 2026 12:06:17 -0500 Subject: [PATCH 3/3] fix(benchmarks): preserve measurement provenance Signed-off-by: phernandez --- benchmarks/README.md | 5 +- .../concurrent_write.py | 32 ++++++++++-- benchmarks/tests/test_concurrent_write.py | 52 +++++++++++++++++++ 3 files changed, 82 insertions(+), 7 deletions(-) diff --git a/benchmarks/README.md b/benchmarks/README.md index 73c0d59ed..9b67ba830 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -178,8 +178,9 @@ create and edit notes in one shared Basic Memory project, with overlapping relation targets and shared hub notes that every writer appends to (the multi-agent shape from basic-memory#1213/#1214). -The driver requires a local git checkout so every result records the exact -Basic Memory commit under test: +The driver requires a clean local git checkout so every result records the exact +Basic Memory commit under test. Commit or stash tracked and untracked changes in +the target checkout before running it: ```bash # Run from the Core benchmarks directory. diff --git a/benchmarks/src/basic_memory_benchmarks/concurrent_write.py b/benchmarks/src/basic_memory_benchmarks/concurrent_write.py index f273bd47d..3d0a774c3 100644 --- a/benchmarks/src/basic_memory_benchmarks/concurrent_write.py +++ b/benchmarks/src/basic_memory_benchmarks/concurrent_write.py @@ -658,6 +658,29 @@ def summarize_op_type(results: list[OpResult]) -> OpTypeStats: ) +def resolve_clean_checkout_sha(checkout: Path) -> str: + """Return the exact target SHA, rejecting bytes that the SHA cannot identify.""" + resolved_sha = git_sha(checkout) + if resolved_sha is None: + raise ValueError("--bm-local-path must point to a Basic Memory git checkout") + + status = run_command( + ["git", "-C", str(checkout), "status", "--porcelain=v1", "--untracked-files=all"] + ) + dirty_paths = [line for line in status.stdout.splitlines() if line.strip()] + # Trigger: tracked or untracked bytes differ from the recorded commit. + # Why: the run must be reproducible from bm_resolved_sha alone. + # Outcome: abort before creating the isolated benchmark home or artifacts. + if dirty_paths: + sample = ", ".join(dirty_paths[:5]) + suffix = " ..." if len(dirty_paths) > 5 else "" + raise ValueError( + "--bm-local-path must be clean so bm_resolved_sha identifies the executed bytes; " + f"dirty paths: {sample}{suffix}" + ) + return resolved_sha + + def build_summary( *, config: ConcurrentWriteConfig, @@ -679,6 +702,7 @@ def build_summary( typed = [result for result in results if result.op_type == op_type] if typed: per_op_type[op_type] = summarize_op_type(typed) + concurrent_results = [result for result in results if result.op_type != "create_hub"] notes_created_ok = sum(1 for r in results if r.op_type == "create" and r.ok) return ConcurrentWriteSummary( run_id=config.run_id, @@ -697,7 +721,7 @@ def build_summary( creates_per_minute=round(notes_created_ok / (concurrent_wall_seconds / 60), 1) if concurrent_wall_seconds > 0 else 0.0, - ops_per_second=round(len(results) / concurrent_wall_seconds, 2) + ops_per_second=round(len(concurrent_results) / concurrent_wall_seconds, 2) if concurrent_wall_seconds > 0 else 0.0, integrity=integrity, @@ -741,7 +765,7 @@ def build_summary_markdown( f" {summary.ops_not_attempted} not attempted", f"- Terminal writer failures: {summary.terminal_writer_failures}", f"- Notes created: {summary.notes_created_ok}" - f" ({summary.creates_per_minute} notes/min, {summary.ops_per_second} ops/s overall)", + f" ({summary.creates_per_minute} notes/min, {summary.ops_per_second} concurrent ops/s)", "", "## Latency (ms)", "", @@ -878,9 +902,7 @@ def run_concurrent_write(config: ConcurrentWriteConfig) -> Path: """Execute the full driver: setup, concurrent phase, settle, verify, report.""" bm_checkout = Path(config.bm_local_path).expanduser().resolve() prefix = resolve_bm_command_prefix(str(bm_checkout)) - bm_resolved_sha = git_sha(bm_checkout) - if bm_resolved_sha is None: - raise ValueError("--bm-local-path must point to a Basic Memory git checkout") + bm_resolved_sha = resolve_clean_checkout_sha(bm_checkout) config = config.model_copy(update={"bm_local_path": str(bm_checkout)}) home = Path("benchmarks/.bm-homes") / f"bm-write-{config.run_id}" diff --git a/benchmarks/tests/test_concurrent_write.py b/benchmarks/tests/test_concurrent_write.py index 8832ede64..a21f73e21 100644 --- a/benchmarks/tests/test_concurrent_write.py +++ b/benchmarks/tests/test_concurrent_write.py @@ -27,6 +27,7 @@ build_writer_plan, classify_error, run_integrity_checks, + resolve_clean_checkout_sha, summarize_op_type, tool_result_error, ) @@ -241,6 +242,33 @@ def test_unknown_status_schema_uses_fixed_settle_delay( assert delays == [concurrent_write.FALLBACK_SETTLE_SECONDS] +def test_resolve_clean_checkout_sha_rejects_dirty_target( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr(concurrent_write, "git_sha", lambda _path: "abc123") + monkeypatch.setattr( + concurrent_write, + "run_command", + lambda *_args, **_kwargs: SimpleNamespace(stdout=" M src/basic_memory/app.py\n?? local.py\n"), + ) + + with pytest.raises(ValueError, match="bm_resolved_sha.*dirty paths"): + resolve_clean_checkout_sha(tmp_path) + + +def test_resolve_clean_checkout_sha_accepts_clean_target( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr(concurrent_write, "git_sha", lambda _path: "abc123") + monkeypatch.setattr( + concurrent_write, + "run_command", + lambda *_args, **_kwargs: SimpleNamespace(stdout=""), + ) + + assert resolve_clean_checkout_sha(tmp_path) == "abc123" + + # --- Integrity verification --- @@ -469,6 +497,30 @@ def test_summary_markdown_reports_convergence(tmp_path: Path) -> None: assert "bm-bench run concurrent-write" in markdown assert summary.converged assert summary.notes_created_ok == 1 + assert summary.ops_per_second == 0.67 + + setup_result = OpResult( + writer=-1, + op_index=0, + op_type="create_hub", + identifier="hubs/hub-0", + started_at_utc="2026-01-01T00:00:00Z", + latency_ms=500.0, + ok=True, + markers=["bmk-setup-h00-l0"], + ) + with_setup_summary = build_summary( + config=config, + results=[setup_result, *results], + outcomes=[WriterOutcome(results=results)], + concurrent_wall_seconds=1.5, + settle_seconds=0.5, + settle_mode="status-json", + reindex_seconds=None, + integrity=integrity, + ) + assert with_setup_summary.ops_total == 2 + assert with_setup_summary.ops_per_second == 0.67 failed_result = OpResult( writer=0,