feat: concurrent-writer driver with convergence verification (bm#1248)

README.md

@@ -169,6 +169,39 @@ 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 rather than retrieval quality: N
+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).
+
+```bash
+# Small-scale smoke (4 writers x 25 notes, strict convergence gate)
+just bench-write-smoke
+
+# Load shape, report-only (divergence is a valid benchmark result)
+just bench-write-load writers=8 notes=200
+
+# Direct invocation with a local BM checkout
+uv run bm-bench run concurrent-write \
+  --writers 4 --notes-per-writer 25 \
+  --bm-local-path /path/to/basic-memory
+```
+
+Per run (`benchmarks/runs/<run-id>/`): `manifest.json`, `per-op.jsonl`
+(latency + error per operation), `concurrent-write-summary.json`, `summary.md`.
+After the concurrent phase settles, the driver verifies convergence directly
+against the on-disk files and the run's isolated SQLite index: file/entity/row
+counts agree, no duplicate permalinks, no duplicate observation or relation
+tuples, and every observation line written by a reported-success op is present
+exactly once (unique `bmk-*` markers detect both lost and 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`.

justfile

@@ -210,6 +210,21 @@ 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-write-smoke:
+    uv run bm-bench run concurrent-write \
+      --writers 4 --notes-per-writer 25 \
+      {{bm_local_path_flag}} \
+      --strict
+
+# Load shape for the v0.22.1-vs-v0.23 comparison; report-only (divergence IS the result)
+bench-write-load writers="8" notes="200":
+    uv run bm-bench run concurrent-write \
+      --writers {{writers}} --notes-per-writer {{notes}} \
+      {{bm_local_path_flag}}
+
 # --- Artifacts and comparison ---
 
 bench-latest-run:

src/basic_memory_benchmarks/bm_runtime.py

@@ -0,0 +1,174 @@
+"""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 <path> 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 Queue
+from typing import Any
+
+import anyio
+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
+
+    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 self._required_tool not in tool_names:
+                    raise RuntimeError(f"bm mcp server does not expose '{self._required_tool}'")
+
+                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
+
+
+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

src/basic_memory_benchmarks/cli.py

@@ -19,6 +19,10 @@
     LONGMEMEVAL_S_URL,
     fetch_longmemeval_dataset,
 )
+from basic_memory_benchmarks.concurrent_write import (
+    ConcurrentWriteConfig,
+    run_concurrent_write,
+)
 from basic_memory_benchmarks.models import DatasetProvenance, RunConfig
 from basic_memory_benchmarks.reporting.compare import (
     compare_provider_metric,
@@ -260,6 +264,66 @@ 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("github:basicmachines-co/basic-memory@main", "--bm-source"),
+    bm_local_path: str | None = typer.Option(None, "--bm-local-path"),
+    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 (divergence is a valid benchmark result, so default is report-only)",
+    ),
+) -> None:
+    """Concurrency benchmark: N MCP writers against one 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=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"),