diff --git a/.gitignore b/.gitignore
index d7c33a50d..49372175e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -63,6 +63,7 @@ claude-output
 .mcpregistry_*
 /.testmondata
 .benchmarks/
+/.scratch/
 
 # Consolidated package build artifacts
 /integrations/openclaw/node_modules/
diff --git a/benchmarks/docs/read-load-benchmark.md b/benchmarks/docs/read-load-benchmark.md
new file mode 100644
index 000000000..b6f5f1129
--- /dev/null
+++ b/benchmarks/docs/read-load-benchmark.md
@@ -0,0 +1,64 @@
+# Read-path load benchmark
+
+This benchmark measures repeated, direct-permalink `read_note` calls through the real
+`bm mcp` stdio server. It compares an authoritative warm baseline with the same workload after
+the standalone Redis read cache is warmed.
+
+## Workload
+
+- deterministic 1, 16, and 64 KiB Markdown notes;
+- 32 distinct notes per size;
+- 128 measured reads at concurrency 1, 8, 32, and 64;
+- corpus materialization, indexing, connection setup, and cache warmup outside measurement;
+- isolated Basic Memory config, database, project, and home directories for every run;
+- JSONL output with p50, p95, p99, throughput, response bandwidth, errors, and workload metadata;
+- a `manifest.json` beside each result with benchmark and Basic Memory SHAs, provider versions,
+  selected Redis and database server versions, target and harness Python versions, dirty-worktree
+  state, synthetic-corpus checksum, and runtime configuration.
+
+Every inherited `BASIC_MEMORY_*` setting is removed before the harness adds its explicit isolated
+configuration. The harness explicitly disables Basic Memory auto-update so measurement cannot
+start background package checks or upgrades. The cached run sets `BASIC_MEMORY_REDIS_URL` only in
+the spawned MCP process and sizes `BASIC_MEMORY_REDIS_MAX_CONNECTIONS` to the largest requested
+workload or seed concurrency. This prevents high-concurrency rows from exhausting the default pool
+and silently measuring the cache's authoritative-read fallback. Output records whether Redis was
+enabled and the non-secret pool size without recording the URL, because URLs may contain
+credentials.
+
+The Basic Memory SHA comes from the `basic_memory` module imported by the Python environment behind
+`--bm-command`, not from the console script's parent directory. Publishable per-ref environments
+must therefore use an editable install linked to the source checkout; the harness fails rather
+than attributing a wheel or unrelated enclosing repository to the wrong SHA.
+
+## Run a paired comparison
+
+Use the same Redis server for the cached repetitions, but use a distinct scratch directory for
+every run. The server must be ready before starting the benchmark; its startup time is not part
+of the measurement.
+
+```bash
+just bench-read-cache redis://127.0.0.1:6379/0 run-01
+```
+
+The recipe writes each side's `results.jsonl` and `manifest.json` under
+`.scratch/read-load-{authoritative,redis-warm}-run-01/`, then prints the Markdown comparison. Give
+each repetition a distinct run ID so its corpus, results, and provenance remain available. Use
+`just bench-read-load <label> [redis_url]` when running one side independently.
+When invoking the script with a custom `--output` outside `--scratch`, `manifest.json` follows the
+JSONL file into that output directory so retained results never lose their provenance.
+The output path must be new unless `--truncate` is explicit; otherwise the harness fails before
+starting runtime resources so one manifest cannot describe JSONL rows appended by multiple runs.
+Each custom output directory is one run-artifact boundary: a second output filename is rejected
+when that directory already contains `manifest.json`, preserving the first result's provenance.
+Custom output and manifest paths must also stay outside the benchmark's runtime-owned `config`,
+`project`, and `main-home` directories. Otherwise Basic Memory could index a result while the run
+is active and invalidate the Redis entries being measured.
+
+Use `just bench-read-smoke [redis_url]` for a tiny real-MCP run at concurrency 64 that verifies
+Redis pool sizing plus result and manifest generation before committing to the full matrix.
+
+Latency is evidence, not a CI threshold. Use at least six paired repetitions before making a
+performance claim, alternate run order, and discard runs with material host contention. The
+manifest must report clean benchmark and Basic Memory worktrees for a publishable comparison. The
+real-Redis integration suite remains the correctness gate for cache identity and invalidation.
+Run that gate with `just test-read-cache`.
diff --git a/benchmarks/scripts/read_load_bench.py b/benchmarks/scripts/read_load_bench.py
new file mode 100644
index 000000000..ad2465e7d
--- /dev/null
+++ b/benchmarks/scripts/read_load_bench.py
@@ -0,0 +1,924 @@
+"""Read-path load benchmark for Basic Memory.
+
+Drives direct-permalink `read_note` calls over a warm `bm mcp` stdio session
+and measures caller-perceived latency, throughput, response bandwidth, and
+content correctness across note sizes and concurrency levels.
+
+The benchmark compares installed Basic Memory builds without importing their
+internals. Point `--bm-command` at a per-ref virtual environment; the harness,
+corpus, request order, and MCP contract remain fixed.
+
+Output is one JSONL record per size/concurrency pair in the generic benchmark
+format: `{"benchmark", "metrics", "timestamp_utc"}`.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import hashlib
+import json
+import os
+import platform
+import random
+import shlex
+import shutil
+import sqlite3
+import subprocess
+import sys
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Protocol
+from urllib.parse import urlparse
+
+from mcp.client.session import ClientSession
+from mcp.client.stdio import StdioServerParameters, stdio_client
+from mcp.types import CallToolResult
+from sqlalchemy.ext.asyncio import create_async_engine
+
+_CORPUS_SEARCH_TOKEN = "readloadbenchmarkmarker"
+
+
+@dataclass(frozen=True, slots=True)
+class ReadTarget:
+    """One materialized note and the marker that proves a correct response."""
+
+    identifier: str
+    marker: str
+    requested_content_bytes: int
+
+
+@dataclass(frozen=True, slots=True)
+class BasicMemoryRuntime:
+    """Provenance from the Python interpreter that runs the target BM command."""
+
+    module_path: Path
+    python_version: str
+
+
+class PostgresContainer(Protocol):
+    """Narrow testcontainer lifecycle used by the optional Postgres backend."""
+
+    def start(self) -> object: ...
+
+    def stop(self) -> None: ...
+
+    def get_connection_url(
+        self,
+        host: str | None = None,
+        driver: str | None = None,
+    ) -> str: ...
+
+
+# --- Synthetic corpus ------------------------------------------------------
+
+
+def size_label(size_bytes: int) -> str:
+    """Return a stable filesystem-safe label for one payload size."""
+    if size_bytes % 1024 == 0:
+        return f"{size_bytes // 1024}kib"
+    return f"{size_bytes}b"
+
+
+def synthetic_read_note(size_bytes: int, index: int) -> tuple[str, str, str]:
+    """Build deterministic ASCII Markdown of exactly ``size_bytes`` bytes."""
+    label = size_label(size_bytes)
+    title = f"read-load-{label}-note-{index:05d}"
+    marker = f"read-load-marker-{label}-{index:05d}"
+    prefix = (
+        f"# {title}\n\n"
+        f"{_CORPUS_SEARCH_TOKEN} {marker}\n\n"
+        "This note exercises direct content reads through the public MCP read_note tool.\n\n"
+    )
+    if len(prefix) > size_bytes:
+        raise ValueError(f"size {size_bytes} is too small for benchmark metadata")
+    filler = "database content read benchmark payload "
+    repeats = ((size_bytes - len(prefix)) // len(filler)) + 1
+    content = (prefix + filler * repeats)[:size_bytes]
+    assert len(content.encode("utf-8")) == size_bytes
+    return title, marker, content
+
+
+# --- Metric and result helpers --------------------------------------------
+
+
+def percentile(values: list[float], pct: float) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    if len(ordered) == 1:
+        return ordered[0]
+    rank = (pct / 100.0) * (len(ordered) - 1)
+    low = int(rank)
+    high = min(low + 1, len(ordered) - 1)
+    fraction = rank - low
+    return ordered[low] + (ordered[high] - ordered[low]) * fraction
+
+
+def emit(output_path: Path | None, record: dict[str, object]) -> None:
+    line = json.dumps(record)
+    print(line, flush=True)
+    if output_path is not None:
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        with output_path.open("a", encoding="utf-8") as handle:
+            handle.write(line + "\n")
+
+
+def checked_output(command: list[str], *, env: dict[str, str] | None = None) -> str:
+    """Run a provenance command and require one non-empty output value."""
+    try:
+        result = subprocess.run(
+            command,
+            check=True,
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+    except (OSError, subprocess.CalledProcessError) as error:
+        raise RuntimeError(f"provenance command failed: {command[0]}") from error
+    output = result.stdout.strip()
+    if not output:
+        raise RuntimeError(f"provenance command returned no output: {command[0]}")
+    return output
+
+
+def git_root(path: Path) -> Path:
+    """Find the repository that owns a benchmark executable or source file."""
+    start = path if path.is_dir() else path.parent
+    for candidate in (start, *start.parents):
+        if (candidate / ".git").exists():
+            return candidate
+    raise RuntimeError(f"could not resolve a Git repository for {path}")
+
+
+def git_sha(repo_root: Path) -> str:
+    return checked_output(["git", "-C", str(repo_root), "rev-parse", "HEAD"])
+
+
+def git_is_dirty(repo_root: Path) -> bool:
+    """Record whether the resolved SHA has uncommitted differences."""
+    try:
+        result = subprocess.run(
+            ["git", "-C", str(repo_root), "status", "--porcelain"],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+    except (OSError, subprocess.CalledProcessError) as error:
+        raise RuntimeError("could not inspect benchmark Git state") from error
+    return bool(result.stdout.strip())
+
+
+def git_source(repo_root: Path) -> str:
+    """Return a credential-free repository source identifier."""
+    try:
+        remote = checked_output(["git", "-C", str(repo_root), "remote", "get-url", "origin"])
+    except RuntimeError:
+        return f"local:{repo_root.name}"
+
+    if remote.startswith("git@") and ":" in remote:
+        host, path = remote.split("@", 1)[1].split(":", 1)
+        return f"git:{host}/{path.removesuffix('.git')}"
+    if "://" in remote:
+        parsed = urlparse(remote)
+        if parsed.hostname:
+            return f"git:{parsed.hostname}{parsed.path.removesuffix('.git')}"
+    return f"local:{repo_root.name}"
+
+
+def basic_memory_runtime(command_path: Path, env: dict[str, str]) -> BasicMemoryRuntime:
+    """Resolve module and Python provenance from the target BM environment."""
+    python_command: list[str] | None = None
+    for name in ("python", "python3", "python.exe"):
+        interpreter = command_path.parent / name
+        if interpreter.is_file():
+            python_command = [str(interpreter)]
+            break
+
+    if python_command is None:
+        try:
+            with command_path.open(encoding="utf-8") as handle:
+                first_line = handle.readline().strip()
+        except (OSError, UnicodeDecodeError) as error:
+            raise RuntimeError(
+                f"could not resolve the Python environment for {command_path}"
+            ) from error
+        if not first_line.startswith("#!"):
+            raise RuntimeError(f"Basic Memory command has no Python shebang: {command_path}")
+        python_command = shlex.split(first_line.removeprefix("#!"))
+        if not python_command:
+            raise RuntimeError(f"Basic Memory command has an empty shebang: {command_path}")
+
+    probe = (
+        "import basic_memory, json, platform; "
+        'print(json.dumps({"module_path": basic_memory.__file__, '
+        '"python_version": platform.python_version()}))'
+    )
+    runtime_output = checked_output(
+        [
+            *python_command,
+            "-c",
+            probe,
+        ],
+        env=env,
+    )
+    try:
+        runtime = json.loads(runtime_output)
+    except json.JSONDecodeError as error:
+        raise RuntimeError("Basic Memory returned invalid runtime provenance") from error
+    if not isinstance(runtime, dict):
+        raise TypeError("Basic Memory returned invalid runtime provenance")
+    module_output = runtime.get("module_path")
+    python_version = runtime.get("python_version")
+    if not isinstance(module_output, str) or not isinstance(python_version, str):
+        raise TypeError("Basic Memory returned incomplete runtime provenance")
+
+    module_path = Path(module_output).resolve()
+    if not module_path.is_file():
+        raise RuntimeError(f"Basic Memory reported an invalid module path: {module_output}")
+    return BasicMemoryRuntime(module_path=module_path, python_version=python_version)
+
+
+def synthetic_corpus_checksum(*, sizes: list[int], notes_per_size: int) -> str:
+    """Hash the exact deterministic corpus with unambiguous value boundaries."""
+    digest = hashlib.sha256()
+    for size_bytes in sizes:
+        for index in range(notes_per_size):
+            for value in synthetic_read_note(size_bytes, index):
+                encoded = value.encode("utf-8")
+                digest.update(len(encoded).to_bytes(8, "big"))
+                digest.update(encoded)
+    return digest.hexdigest()
+
+
+async def redis_server_version(redis_url: str) -> str:
+    """Read Redis server provenance without persisting its credential-bearing URL."""
+    from redis.asyncio import Redis
+
+    connection_url = redis_url if "://" in redis_url else f"redis://{redis_url}"
+    client = Redis.from_url(connection_url, decode_responses=True)
+    try:
+        server = await client.info(section="server")
+    finally:
+        await client.aclose()
+    version = server.get("redis_version")
+    if not isinstance(version, str) or not version:
+        raise RuntimeError("Redis did not report its server version")
+    return version
+
+
+async def database_server_version(*, backend: str, database_url: str | None) -> str:
+    """Read the selected database engine version without retaining its URL."""
+    if backend == "sqlite":
+        return sqlite3.sqlite_version
+    if database_url is None:
+        raise ValueError("Postgres provenance requires a database URL")
+
+    engine = create_async_engine(database_url)
+    try:
+        async with engine.connect() as connection:
+            result = await connection.exec_driver_sql("SHOW server_version")
+            return str(result.scalar_one())
+    finally:
+        await engine.dispose()
+
+
+def benchmark_manifest_path(*, scratch: Path, output_path: Path | None) -> Path:
+    """Keep provenance beside retained results, or in scratch for stdout-only runs."""
+    return (output_path.parent if output_path is not None else scratch) / "manifest.json"
+
+
+def write_manifest(
+    *,
+    scratch: Path,
+    output_path: Path | None,
+    args: argparse.Namespace,
+    env: dict[str, str],
+    sizes: list[int],
+    concurrency_levels: list[int],
+    database_version: str,
+    redis_version: str | None,
+) -> None:
+    """Write the provenance required to reproduce and interpret one run."""
+    command_path = Path(shutil.which(args.bm_command) or "").resolve()
+    if not command_path.is_file():
+        raise RuntimeError(f"could not resolve Basic Memory command: {args.bm_command}")
+    bm_runtime = basic_memory_runtime(command_path, env)
+    bm_module_path = bm_runtime.module_path
+    bm_repo_root = git_root(bm_module_path)
+    package_root = (bm_repo_root / "src" / "basic_memory").resolve()
+    if not bm_module_path.is_relative_to(package_root):
+        raise RuntimeError(
+            "installed Basic Memory is not linked to the resolved source checkout; "
+            "use a per-ref editable environment"
+        )
+    benchmark_repo_root = git_root(Path(__file__).resolve())
+    try:
+        command_source = str(command_path.relative_to(bm_repo_root))
+    except ValueError:
+        command_source = command_path.name
+
+    provider_versions = {
+        "basic-memory": {
+            "command": command_source,
+            "module": str(bm_module_path.relative_to(bm_repo_root)),
+            "python_version": bm_runtime.python_version,
+            "version": checked_output([str(command_path), "--version"], env=env),
+        }
+    }
+    if redis_version is not None:
+        provider_versions["redis"] = {"version": redis_version}
+    database_provider = "postgresql" if args.backend == "postgres" else "sqlite"
+    provider_versions[database_provider] = {"version": database_version}
+
+    manifest = {
+        "schema_version": 1,
+        "run_id": args.label,
+        "created_at_utc": datetime.now(UTC).isoformat(),
+        "benchmark_source": git_source(benchmark_repo_root),
+        "benchmark_git_sha": git_sha(benchmark_repo_root),
+        "benchmark_git_dirty": git_is_dirty(benchmark_repo_root),
+        "bm_source": git_source(bm_repo_root),
+        "bm_resolved_sha": git_sha(bm_repo_root),
+        "bm_git_dirty": git_is_dirty(bm_repo_root),
+        "provider_versions": provider_versions,
+        "dataset": {
+            "source": "synthetic:read-load-v1",
+            "checksum_sha256": synthetic_corpus_checksum(
+                sizes=sizes,
+                notes_per_size=args.notes_per_size,
+            ),
+            "sizes_bytes": sizes,
+            "notes_per_size": args.notes_per_size,
+        },
+        "runtime": {
+            "os": platform.platform(),
+            "harness_python_version": sys.version.split()[0],
+            "backend": args.backend,
+            "redis_read_cache_enabled": redis_version is not None,
+        },
+        "config": {
+            "reads_per_scenario": args.reads,
+            "concurrency_levels": concurrency_levels,
+            "seed_concurrency": args.seed_concurrency,
+            "redis_max_connections": (
+                int(env["BASIC_MEMORY_REDIS_MAX_CONNECTIONS"])
+                if redis_version is not None
+                else None
+            ),
+            "seed": args.seed,
+            "ready_timeout_seconds": args.ready_timeout,
+            "quiesce_seconds": args.quiesce_seconds,
+        },
+    }
+    manifest_path = benchmark_manifest_path(scratch=scratch, output_path=output_path)
+    manifest_path.parent.mkdir(parents=True, exist_ok=True)
+    manifest_path.write_text(
+        json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+
+
+def result_payload(result: CallToolResult) -> dict[str, object]:
+    structured = result.structuredContent
+    if isinstance(structured, dict):
+        wrapped = structured.get("result")
+        payload = wrapped if isinstance(wrapped, dict) else structured
+        return {str(key): value for key, value in payload.items()}
+    for item in result.content:
+        text = getattr(item, "text", None)
+        if not isinstance(text, str) or not text.strip():
+            continue
+        try:
+            parsed = json.loads(text)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(parsed, dict):
+            return {str(key): value for key, value in parsed.items()}
+    return {}
+
+
+def result_text(result: CallToolResult) -> str | None:
+    """Return text from a successful text-mode tool response."""
+    if result.isError:
+        return None
+    text_parts = [
+        text for item in result.content if isinstance((text := getattr(item, "text", None)), str)
+    ]
+    return "\n".join(text_parts) if text_parts else None
+
+
+# --- Isolated runtime ------------------------------------------------------
+
+
+def isolated_env(
+    config_dir: Path,
+    *,
+    redis_url: str | None,
+    redis_max_connections: int | None,
+) -> dict[str, str]:
+    """Create a local runtime environment without inherited BM configuration."""
+    env = {key: value for key, value in os.environ.items() if not key.startswith("BASIC_MEMORY_")}
+    env["BASIC_MEMORY_AUTO_UPDATE"] = "false"
+    env["BASIC_MEMORY_CONFIG_DIR"] = str(config_dir)
+    env["BASIC_MEMORY_SEMANTIC_SEARCH_ENABLED"] = "false"
+    env["BASIC_MEMORY_LOG_LEVEL"] = "WARNING"
+    env["LOGFIRE_IGNORE_NO_CONFIG"] = "1"
+    if redis_url is not None:
+        if redis_max_connections is None or redis_max_connections <= 0:
+            raise ValueError("Redis runs require a positive connection-pool size")
+        env["BASIC_MEMORY_REDIS_URL"] = redis_url
+        env["BASIC_MEMORY_REDIS_MAX_CONNECTIONS"] = str(redis_max_connections)
+    return env
+
+
+def start_postgres() -> PostgresContainer:
+    """Start a throwaway pgvector testcontainer for a Postgres run."""
+    from testcontainers.postgres import PostgresContainer as TestPostgresContainer
+
+    container = TestPostgresContainer("pgvector/pgvector:pg16")
+    try:
+        container.start()
+    except Exception:
+        container.stop()
+        raise
+    return container
+
+
+def asyncpg_url(container: PostgresContainer) -> str:
+    """Normalize a testcontainer URL to SQLAlchemy's asyncpg scheme."""
+    url = container.get_connection_url()
+    for sync_driver in ("postgresql+psycopg2", "postgresql+psycopg", "postgresql"):
+        prefix = sync_driver + "://"
+        if url.startswith(prefix):
+            return "postgresql+asyncpg://" + url[len(prefix) :]
+    return url
+
+
+def note_content_rows(db_path: Path) -> int:
+    """Count SQLite note_content rows; older releases may not have the table."""
+    if not db_path.exists():
+        return 0
+    try:
+        connection = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=1.0)
+        try:
+            table = connection.execute(
+                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'note_content'"
+            ).fetchone()
+            if table is None:
+                return 0
+            row = connection.execute("SELECT COUNT(*) FROM note_content").fetchone()
+            return int(row[0]) if row is not None else 0
+        finally:
+            connection.close()
+    except sqlite3.Error:
+        return 0
+
+
+async def poll_until(
+    predicate: Callable[[], Awaitable[bool]],
+    *,
+    timeout_seconds: float,
+    interval_seconds: float = 0.1,
+) -> float | None:
+    """Return elapsed milliseconds when an async predicate becomes true."""
+    started = time.perf_counter()
+    delay = interval_seconds
+    while True:
+        if await predicate():
+            return (time.perf_counter() - started) * 1000.0
+        if time.perf_counter() - started >= timeout_seconds:
+            return None
+        await asyncio.sleep(delay)
+        delay = min(delay * 1.5, 1.0)
+
+
+# --- Corpus preparation ----------------------------------------------------
+
+
+async def write_target(
+    session: ClientSession,
+    *,
+    project: str,
+    size_bytes: int,
+    index: int,
+) -> ReadTarget:
+    title, marker, content = synthetic_read_note(size_bytes, index)
+    directory = f"read-load/{size_label(size_bytes)}"
+    result = await session.call_tool(
+        "write_note",
+        {
+            "project": project,
+            "title": title,
+            "directory": directory,
+            "content": content,
+            "output_format": "json",
+        },
+    )
+    if result.isError:
+        raise RuntimeError(f"write_note failed while seeding {title}")
+    payload = result_payload(result)
+    identifier = payload.get("permalink")
+    if not isinstance(identifier, str) or not identifier:
+        identifier = f"{directory}/{title}"
+    return ReadTarget(
+        identifier=identifier,
+        marker=marker,
+        requested_content_bytes=size_bytes,
+    )
+
+
+async def seed_corpus(
+    session: ClientSession,
+    *,
+    project: str,
+    sizes: list[int],
+    notes_per_size: int,
+    concurrency: int,
+) -> list[ReadTarget]:
+    """Create the deterministic corpus with bounded concurrent writes."""
+    semaphore = asyncio.Semaphore(concurrency)
+
+    async def write_one(size_bytes: int, index: int) -> ReadTarget:
+        async with semaphore:
+            return await write_target(
+                session,
+                project=project,
+                size_bytes=size_bytes,
+                index=index,
+            )
+
+    return list(
+        await asyncio.gather(
+            *(
+                write_one(size_bytes, index)
+                for size_bytes in sizes
+                for index in range(notes_per_size)
+            )
+        )
+    )
+
+
+async def searchable_count(session: ClientSession, project: str) -> int:
+    result = await session.call_tool(
+        "search_notes",
+        {
+            "project": project,
+            "query": _CORPUS_SEARCH_TOKEN,
+            "search_type": "text",
+            "page": 1,
+            "page_size": 1,
+            "output_format": "json",
+        },
+    )
+    if result.isError:
+        return 0
+    payload = result_payload(result)
+    total = payload.get("total")
+    if isinstance(total, int):
+        return total
+    results = payload.get("results")
+    return len(results) if isinstance(results, list) else 0
+
+
+async def warm_targets(session: ClientSession, project: str, targets: list[ReadTarget]) -> None:
+    """Validate each direct permalink once and warm database/filesystem caches."""
+    for target in targets:
+        result = await session.call_tool(
+            "read_note",
+            {"project": project, "identifier": target.identifier},
+        )
+        content = result_text(result)
+        if content is None or target.marker not in content:
+            raise RuntimeError(f"warm read failed validation for {target.identifier}")
+
+
+# --- Measured workload -----------------------------------------------------
+
+
+async def read_burst(
+    session: ClientSession,
+    *,
+    project: str,
+    targets: list[ReadTarget],
+    count: int,
+    concurrency: int,
+    seed: int,
+) -> tuple[list[float], int, int, int]:
+    """Read deterministic targets with at most ``concurrency`` calls in flight."""
+    request_targets = [targets[index % len(targets)] for index in range(count)]
+    random.Random(seed).shuffle(request_targets)
+    semaphore = asyncio.Semaphore(concurrency)
+    latencies: list[float] = []
+    errors = 0
+    validation_failures = 0
+    response_bytes = 0
+
+    async def read_one(target: ReadTarget) -> None:
+        nonlocal errors, response_bytes, validation_failures
+        async with semaphore:
+            started = time.perf_counter()
+            try:
+                result = await session.call_tool(
+                    "read_note",
+                    {"project": project, "identifier": target.identifier},
+                )
+            # Each request is an observation. Transport/protocol failures count as benchmark
+            # errors so the remaining deterministic burst can complete and report its rate.
+            except Exception:  # noqa: BLE001
+                errors += 1
+                latencies.append((time.perf_counter() - started) * 1000.0)
+                return
+            latencies.append((time.perf_counter() - started) * 1000.0)
+            content = result_text(result)
+            if content is None:
+                errors += 1
+                return
+            response_bytes += len(content.encode("utf-8"))
+            if target.marker not in content:
+                validation_failures += 1
+
+    await asyncio.gather(*(read_one(target) for target in request_targets))
+    return latencies, errors, validation_failures, response_bytes
+
+
+async def run(args: argparse.Namespace) -> int:
+    sizes = [int(value) for value in args.sizes.split(",") if value.strip()]
+    concurrency_levels = [int(value) for value in args.concurrency.split(",") if value.strip()]
+    if not sizes or min(sizes) <= 0:
+        raise ValueError("--sizes must contain positive byte counts")
+    if len(sizes) != len(set(sizes)):
+        raise ValueError("--sizes must not contain duplicates")
+    if not concurrency_levels or min(concurrency_levels) <= 0:
+        raise ValueError("--concurrency must contain positive integers")
+    if len(concurrency_levels) != len(set(concurrency_levels)):
+        raise ValueError("--concurrency must not contain duplicates")
+    for option, value in (
+        ("--notes-per-size", args.notes_per_size),
+        ("--reads", args.reads),
+        ("--seed-concurrency", args.seed_concurrency),
+    ):
+        if value <= 0:
+            raise ValueError(f"{option} must be a positive integer")
+
+    scratch = Path(args.scratch).resolve()
+    output_path = Path(args.output).resolve() if args.output else None
+    manifest_path = benchmark_manifest_path(scratch=scratch, output_path=output_path)
+    config_dir = (scratch / "config").resolve()
+    project_dir = (scratch / "project").resolve()
+    main_home = (scratch / "main-home").resolve()
+    runtime_dirs = (config_dir, project_dir, main_home)
+    if output_path == manifest_path:
+        raise ValueError("--output filename must not be manifest.json")
+    if output_path is not None and any(
+        artifact_path.is_relative_to(runtime_dir)
+        for artifact_path in (output_path, manifest_path)
+        for runtime_dir in runtime_dirs
+    ):
+        raise ValueError("--output and manifest must stay outside benchmark runtime directories")
+    if output_path is not None and output_path.exists() and not args.truncate:
+        raise ValueError("--output already exists; pass --truncate to replace its run artifacts")
+    if output_path is not None and manifest_path.exists() and not output_path.exists():
+        raise ValueError("--output directory already contains manifest.json for another run")
+    if output_path is not None and args.truncate:
+        output_path.unlink(missing_ok=True)
+        manifest_path.unlink(missing_ok=True)
+
+    for path in (config_dir, project_dir, main_home):
+        if path.exists():
+            shutil.rmtree(path)
+        path.mkdir(parents=True, exist_ok=True)
+
+    pg_container: PostgresContainer | None = None
+    try:
+        pg_container = (
+            start_postgres() if args.backend == "postgres" and not args.database_url else None
+        )
+        database_url = args.database_url or (asyncpg_url(pg_container) if pg_container else None)
+
+        redis_url = args.redis_url.strip() if args.redis_url else None
+        redis_max_connections = (
+            max(max(concurrency_levels), args.seed_concurrency) if redis_url is not None else None
+        )
+        project = "readload"
+        env = isolated_env(
+            config_dir,
+            redis_url=redis_url,
+            redis_max_connections=redis_max_connections,
+        )
+        env["BASIC_MEMORY_HOME"] = str(main_home)
+        if args.backend == "postgres":
+            if database_url is None:
+                raise ValueError("Postgres backend requires a database URL")
+            env["BASIC_MEMORY_DATABASE_BACKEND"] = "postgres"
+            env["BASIC_MEMORY_DATABASE_URL"] = database_url
+
+        database_version = await database_server_version(
+            backend=args.backend,
+            database_url=database_url,
+        )
+        redis_version = await redis_server_version(redis_url) if redis_url is not None else None
+        write_manifest(
+            scratch=scratch,
+            output_path=output_path,
+            args=args,
+            env=env,
+            sizes=sizes,
+            concurrency_levels=concurrency_levels,
+            database_version=database_version,
+            redis_version=redis_version,
+        )
+        params = StdioServerParameters(command=args.bm_command, args=["mcp"], env=env)
+        had_failures = False
+
+        async with (
+            stdio_client(params) as (read_stream, write_stream),
+            ClientSession(read_stream, write_stream) as session,
+        ):
+            await session.initialize()
+            created = await session.call_tool(
+                "create_memory_project",
+                {"project_name": project, "project_path": str(project_dir)},
+            )
+            if created.isError:
+                raise RuntimeError("could not create benchmark project")
+
+            targets = await seed_corpus(
+                session,
+                project=project,
+                sizes=sizes,
+                notes_per_size=args.notes_per_size,
+                concurrency=args.seed_concurrency,
+            )
+            expected_notes = len(targets)
+
+            async def files_ready() -> bool:
+                return len(list(project_dir.rglob("*.md"))) >= expected_notes
+
+            async def search_ready() -> bool:
+                return await searchable_count(session, project) >= expected_notes
+
+            materialized_ms = await poll_until(
+                files_ready,
+                timeout_seconds=args.ready_timeout,
+            )
+            searchable_ms = await poll_until(
+                search_ready,
+                timeout_seconds=args.ready_timeout,
+            )
+            if materialized_ms is None or searchable_ms is None:
+                raise RuntimeError(
+                    "corpus did not finish materializing and indexing before measurement"
+                )
+
+            await asyncio.sleep(args.quiesce_seconds)
+
+            sqlite_note_content_rows = (
+                note_content_rows(config_dir / "memory.db") if args.backend == "sqlite" else -1
+            )
+
+            for size_bytes in sizes:
+                size_targets = [
+                    target for target in targets if target.requested_content_bytes == size_bytes
+                ]
+                for concurrency in concurrency_levels:
+                    # Response entries expire after 60 seconds and reads do not extend their TTL.
+                    # Rewarming outside the timer keeps every row on the advertised cache state.
+                    await warm_targets(session, project, size_targets)
+                    wall_started = time.perf_counter()
+                    latencies, errors, validation_failures, response_bytes = await read_burst(
+                        session,
+                        project=project,
+                        targets=size_targets,
+                        count=args.reads,
+                        concurrency=concurrency,
+                        seed=args.seed + size_bytes + concurrency,
+                    )
+                    wall_seconds = time.perf_counter() - wall_started
+                    successful_reads = args.reads - errors
+                    throughput = args.reads / wall_seconds if wall_seconds > 0 else 0.0
+                    mib_per_second = (
+                        response_bytes / (1024 * 1024) / wall_seconds if wall_seconds > 0 else 0.0
+                    )
+                    had_failures = had_failures or errors > 0 or validation_failures > 0
+                    record: dict[str, object] = {
+                        "benchmark": (f"read-load size={size_label(size_bytes)} c={concurrency}"),
+                        "timestamp_utc": datetime.now(UTC).isoformat(),
+                        "label": args.label,
+                        "metadata": {
+                            "backend": args.backend,
+                            "redis_read_cache_enabled": redis_url is not None,
+                            "semantic_search_enabled": False,
+                            "warm_direct_permalink_reads": True,
+                            "corpus_notes": expected_notes,
+                            "notes_per_size": args.notes_per_size,
+                            "materialization_ready_ms": round(materialized_ms, 1),
+                            "search_ready_ms": round(searchable_ms, 1),
+                            "sqlite_note_content_rows": sqlite_note_content_rows,
+                        },
+                        "metrics": {
+                            "concurrency": concurrency,
+                            "requested_content_bytes": size_bytes,
+                            "reads_requested": args.reads,
+                            "successful_reads": successful_reads,
+                            "read_latency_p50_ms": round(percentile(latencies, 50), 3),
+                            "read_latency_p95_ms": round(percentile(latencies, 95), 3),
+                            "read_latency_p99_ms": round(percentile(latencies, 99), 3),
+                            "read_latency_max_ms": (round(max(latencies), 3) if latencies else 0.0),
+                            "read_throughput_per_sec": round(throughput, 3),
+                            "response_mib_per_sec": round(mib_per_second, 3),
+                            "average_response_bytes": (
+                                round(response_bytes / successful_reads, 1)
+                                if successful_reads > 0
+                                else 0.0
+                            ),
+                            "read_error_rate": round(errors / args.reads, 4),
+                            "validation_failure_rate": round(validation_failures / args.reads, 4),
+                        },
+                    }
+                    emit(output_path, record)
+    finally:
+        if pg_container is not None:
+            pg_container.stop()
+
+    return 2 if had_failures else 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--bm-command",
+        required=True,
+        help="Path to the basic-memory executable in the per-ref virtual environment",
+    )
+    parser.add_argument("--label", default="ref", help="Ref label recorded in each row")
+    parser.add_argument(
+        "--sizes",
+        default="1024,16384,65536",
+        help="Comma-separated generated Markdown sizes in bytes",
+    )
+    parser.add_argument(
+        "--notes-per-size",
+        type=int,
+        default=32,
+        help="Distinct corpus notes generated for each size",
+    )
+    parser.add_argument("--reads", type=int, default=128, help="Reads per size/concurrency row")
+    parser.add_argument(
+        "--concurrency",
+        default="1,8,32,64",
+        help="Comma-separated in-flight read limits",
+    )
+    parser.add_argument(
+        "--seed-concurrency",
+        type=int,
+        default=8,
+        help="Bounded write concurrency used only to prepare the corpus",
+    )
+    parser.add_argument("--seed", type=int, default=1021, help="Deterministic request-order seed")
+    parser.add_argument(
+        "--backend",
+        choices=("sqlite", "postgres"),
+        default="sqlite",
+        help="Database backend used by the Basic Memory server",
+    )
+    parser.add_argument(
+        "--database-url",
+        default=None,
+        help="Postgres URL; when omitted, a throwaway pgvector testcontainer is started",
+    )
+    parser.add_argument(
+        "--redis-url",
+        default=None,
+        help="Enable standalone MCP read caching with this Redis URL or bare hostname",
+    )
+    parser.add_argument(
+        "--scratch",
+        default=".scratch/read-load",
+        help="Scratch directory for isolated config and project files",
+    )
+    parser.add_argument("--output", default=None, help="JSONL output path (also printed)")
+    parser.add_argument(
+        "--truncate",
+        action="store_true",
+        help="Replace an existing output and its manifest before writing",
+    )
+    parser.add_argument(
+        "--ready-timeout",
+        type=float,
+        default=120.0,
+        help="Seconds to wait for corpus materialization and FTS indexing",
+    )
+    parser.add_argument(
+        "--quiesce-seconds",
+        type=float,
+        default=1.0,
+        help="Quiet interval between readiness and unmeasured cache warmup",
+    )
+    return asyncio.run(run(parser.parse_args()))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/docs/REDIS_READ_CACHE_PLAN.md b/docs/REDIS_READ_CACHE_PLAN.md
index 3207d4803..8e4610843 100644
--- a/docs/REDIS_READ_CACHE_PLAN.md
+++ b/docs/REDIS_READ_CACHE_PLAN.md
@@ -13,8 +13,8 @@ Redis instance.
   requirement.
 - Make cache semantics, typed serialization, TTLs, and invalidation part of Basic Memory.
 - Let a host such as Basic Memory Cloud inject an existing Redis client and opaque namespace.
-- Preserve independent clients, key prefixes, metrics, and failure behavior for read caching and
-  rate limiting.
+- Preserve independent key prefixes, metrics, ownership, and failure behavior for read caching
+  and rate limiting.
 
 ## Ownership Boundary
 
@@ -36,13 +36,13 @@ Cloud owns:
 - capacity, eviction, and availability decisions when Redis is shared.
 
 Physical topology is deliberately outside the Basic Memory cache contract. Cloud may point the
-read cache and limiter at separate Redis instances or at separately configured clients on one
-instance.
+read cache and limiter at separate Redis instances or clients, or reuse one client and connection
+pool when both concerns share an instance.
 
 ### Tenant isolation on shared Redis
 
-Cloud does not create a Redis instance or connection pool per tenant. It reuses one long-lived,
-Basic Memory-specific Redis client and constructs a lightweight `RedisReadCache` adapter from
+Cloud does not create a Redis instance or connection pool per tenant. It reuses one long-lived
+Redis client and connection pool and constructs a lightweight `RedisReadCache` adapter from
 trusted request or worker context:
 
 - `namespace`: the stable tenant/workspace UUID;
@@ -61,8 +61,7 @@ invalidates the exact scope populated by the request path.
 
 ### Cloud host requirements
 
-Basic Memory Cloud must satisfy all of these requirements before enabling cached reads for a
-tenant:
+Basic Memory Cloud must satisfy all of these requirements when composing cached reads:
 
 1. Derive one stable namespace from authenticated Cloud context. The preferred value is the
    tenant UUID or workspace UUID already owned by Cloud. Project UUID alone is not a tenant
@@ -70,11 +69,13 @@ tenant:
 1. Complete authorization and tenant database/schema selection before cache lookup. The
    namespace prevents key collisions; it does not replace Cloud's access-control boundary.
 1. Override the low-level `get_read_cache` dependency at the Cloud composition root. Construct
-   `RedisReadCache(client=shared_basic_memory_client, namespace=trusted_namespace)` as a
-   lightweight request-scoped adapter; reuse the long-lived client and connection pool.
-   FastAPI route dependencies then create lightweight `ModelReadCache` facades from that shared
-   backend and bind Basic Memory's response type, TTL, and payload-size policy. Cloud should not
-   duplicate those policy constants or construct Redis clients per response model.
+   `RedisReadCache(client=shared_redis_client, namespace=trusted_namespace)` as a lightweight
+   request-scoped adapter; reuse the long-lived client and connection pool. A single Cloud
+   composition function owns this choice so the cache can move to a separate client or instance
+   later without changing Basic Memory routes or workers. FastAPI route dependencies then create
+   lightweight `ModelReadCache` facades from that shared backend and bind Basic Memory's response
+   type, TTL, and payload-size policy. Cloud should not duplicate those policy constants or
+   construct Redis clients per response model.
 1. Pass the trusted tenant/workspace identity through internal queue payloads, or include enough
    trusted identifiers for workers to derive the exact same namespace. Never copy a namespace
    from a public request field.
@@ -83,6 +84,10 @@ tenant:
    watcher-detected paired moves, relation-resolution workers, and startup recovery or
    reconciliation. Request-path invalidation alone is insufficient because a worker can update a
    cached entity after the request returns.
+1. Invalidate after semantic vector publication, including partial-failure paths. Vector sync can
+   run after the mutation and file-index generation bumps, so VECTOR/HYBRID responses filled while
+   embeddings are stale must not survive the later derived-state update. Queue-backed Cloud vector
+   workers use the same trusted tenant namespace and canonical project UUID as request reads.
 1. Treat each asynchronous state transition as a separate freshness boundary. Invalidate after
    the accepted-note transaction commits, immediately after terminal materialization/status
    publication before indexing begins, again after indexing, and again after relation resolution
@@ -142,9 +147,10 @@ tenant:
 1. Keep `bm:read:v1` separate from rate-limit and Cloud control-plane prefixes, metrics,
    timeouts, and failure policies. The clients may target one Redis deployment, but a read-cache
    timeout must bypass while a rate-limit decision keeps its Cloud-owned security behavior.
-1. Enable reads only after request and worker invalidation use the same namespace in the target
-   environment. Roll out by tenant cohort, watch hit/bypass/invalidation outcomes and database
-   queries, then remove overlapping Cloud gateway response caches only after parity.
+1. Activate request reads and worker invalidation together in the same Cloud release, using the
+   same namespace function. The cache is fail-open and TTL-bounded, so do not add tenant cohorts,
+   shadow reads, or a second rollout switch; use integration coverage and cache telemetry to
+   verify the direct activation.
 1. Coordinate rolling deployments around the `bm:read:v1` payload/key contract. Bump the prefix
    for incompatible serialized response changes so mixed application versions never interpret
    one another's payloads with different schemas.
@@ -233,12 +239,12 @@ Phase one:
 | Directory tree              |  60 seconds | Full hierarchy; two MiB measured payload cap        |
 | Paginated directory listing |  60 seconds | Include path, depth, glob, page, and page-size keys |
 
-Phase two, after measuring phase one:
+Additional measured surfaces:
 
 | Operation                   |   Initial TTL | Constraints                                    |
 | --------------------------- | ------------: | ---------------------------------------------- |
-| Search                      |    30 seconds | Canonicalize the complete query and pagination |
-| Context and recent activity | 15-30 seconds | Normalize or bound time-relative inputs        |
+| Search                      |    30 seconds | Implemented; complete query plus pagination key |
+| Context and recent activity | 15-30 seconds | Future; normalize or bound time-relative inputs |
 
 Do not initially cache failures, missing entities, graph/orphan responses, large or arbitrary
 binary resources, schema inference, writes, or Cloud control-plane data.
@@ -337,10 +343,21 @@ an optional package extra. A host may instead supply a compatible, already-owned
 
 The Core `ApiContainer` carries `ReadCache | None` and defaults to `None`. A managed host
 activates caching by injecting or dependency-overriding a namespace-bound implementation and owns
-that client's lifecycle; Cloud therefore reuses its long-lived Basic Memory cache client. Local
-CLI, MCP in-process ASGI routing, and the standalone API simply skip cache work in the first
-rollout. A later standalone Redis setting can create and close a client in the FastAPI lifespan
-without changing the cache contract.
+that client's lifecycle; Cloud therefore reuses its long-lived Basic Memory cache client.
+
+Standalone `bm mcp` activates the cache when `BASIC_MEMORY_REDIS_URL` is set. A bare hostname such
+as `redis` is normalized to `redis://redis`; complete `redis://` and `rediss://` URLs are preserved.
+The optional `BASIC_MEMORY_REDIS_MAX_CONNECTIONS` setting tunes that process-owned connection pool
+and defaults to 20; it does not activate caching by itself. The MCP lifespan owns one async client
+and shares its namespace-bound cache with both in-process FastAPI requests and watcher/index
+invalidation, then closes the client after those paths shut down. After project reconciliation and
+before accepting requests, every standalone MCP lifespan replaces each active project's persisted
+generation. This prevents entries from an earlier process from hiding a source-of-truth file edit
+made while MCP was stopped. If Redis cannot complete every startup invalidation, caching is disabled
+for that lifespan and requests use the authoritative path; a later Redis recovery cannot revive the
+untrusted generation. When the URL is absent, the dependency remains `None` and callers take only
+the authoritative path. Cloud runtime mode ignores this standalone composition because Cloud
+injects a separately owned cache using its trusted tenant namespace and capacity policy.
 
 `get_read_cache` is the host override point and returns `ReadCache | None`. Core-owned,
 route-specific FastAPI providers call `create_model_read_cache` to return a correctly typed facade
@@ -351,6 +368,15 @@ The FastAPI Redis SDK is not the foundational dependency for this work. The cach
 also participate in portable indexing and hosted storage-event invalidation, and Basic Memory's
 local ASGI transport does not run FastAPI lifespan.
 
+The end-to-end MCP benchmark is documented in
+[`benchmarks/docs/read-load-benchmark.md`](../benchmarks/docs/read-load-benchmark.md). It removes
+ambient Redis configuration for the authoritative run and sets `BASIC_MEMORY_REDIS_URL` only for
+the warmed-cache run, so both cases exercise the same `bm mcp` process boundary. The harness also
+sets `BASIC_MEMORY_REDIS_MAX_CONNECTIONS` to the largest requested workload or seed concurrency and
+records that non-secret value in the manifest. Without this capacity guarantee, Redis pool
+exhaustion would correctly bypass to authoritative reads but mislabel the resulting benchmark row
+as a warmed-cache measurement.
+
 ## Cloud Production Calibration
 
 A bounded production snapshot from 2026-07-30 00:00-19:03 UTC changed directory reads from a
@@ -399,9 +425,10 @@ absorb that authorization-aware composition. Cloud should optimize that active-p
 reconciliation and its own project-list cache independently.
 
 Directory caching in Basic Memory is intended to replace overlapping route families after Cloud
-reaches namespace, invalidation, and observability parity. During rollout, the inner cache can
-also share tenant-project directory results across already-authorized users while Cloud's current
-outer key remains user-specific. Do not keep both response-cache layers as the final design.
+reaches namespace, invalidation, and observability parity. During the transition away from the
+overlapping outer cache, the inner cache can also share tenant-project directory results across
+already-authorized users while Cloud's current outer key remains user-specific. Do not keep both
+response-cache layers as the final design.
 
 ## Failure Behavior
 
@@ -438,7 +465,7 @@ Redis URL.
 Run the focused suite with:
 
 ```bash
-LOGFIRE_IGNORE_NO_CONFIG=1 uv run pytest -p pytest_mock --no-cov -q test-int/read_cache
+just test-read-cache
 ```
 
 `BASIC_MEMORY_TEST_REDIS_URL` selects an externally managed test server. Otherwise the fixture
@@ -457,6 +484,8 @@ The real-Redis suite must prove:
 - generation metadata expires for inactive projects, including legacy persistent keys, and stores
   keep it alive at least as long as their response data;
 - Redis restart or unavailability produces explicit bypass behavior;
+- a fresh standalone MCP lifespan replaces persisted project generations before serving, so
+  offline file edits cannot remain hidden behind cache entries from the previous process;
 - no invalidation operation touches keys outside the Basic Memory prefix;
 - payload size limits;
 - repeated API entity reads use the real cached representation;
@@ -520,13 +549,15 @@ semantics themselves are asserted only against the real Redis integration fixtur
 - Wire project invalidation through accepted writes and indexing paths.
 - Add full-stack API, repeated `read_note`, and directory refresh integration coverage.
 
-### 3. Cloud rollout
+### 3. Cloud integration
 
 - Inject a Basic Memory-specific Redis client and tenant namespace.
 - Derive that namespace from trusted request and worker context with one canonical function.
 - Invalidate after accepted-note commit, immediately after terminal materialization/status
   publication before indexing, again after indexing, and after relation-resolution workers using
   the same tenant namespace as the request path.
+- Invalidate after semantic vector publication, including partial failures, because vector sync is
+  derived work that can complete after the note/index freshness boundaries.
 - Invalidate watcher-detected paired moves at move completion, and invalidate any recovery or
   reconciliation attempt that can publish terminal state before releasing the serving barrier or
   resuming tenant traffic.
@@ -564,15 +595,14 @@ semantics themselves are asserted only against the real Redis integration fixtur
   can change while every cache identity remains stable.
 - Put each startup recovery phase inside a cancellation-safe invalidation scope before beginning
   the next phase.
-- Enable reads for a tenant only after every request, worker, partial-index, direct-index,
-  read-repair, import, accepted-delete, move, and recovery boundary has namespace and invalidation
-  parity.
-- Start with shadow telemetry or a limited tenant cohort.
-- Compare hit rate, Redis latency, database query volume, and end-to-end tool latency.
+- Land request reads and the worker invalidation boundaries in one Cloud change, using the same
+  tenant namespace function everywhere.
+- Do not add shadow reads, tenant cohorts, or a cache-specific rollout switch. Observe hit rate,
+  Redis latency, database query volume, and end-to-end tool latency after direct activation.
 
 ### 4. Expand from evidence
 
-- Add search and graph-context reads when measured reuse supports them.
+- Add graph-context and recent-activity reads when measured reuse supports them.
 - Refine project-wide invalidation only if unrelated writes materially reduce the entity hit
   rate.
 
diff --git a/justfile b/justfile
index 524eddb26..13842d03d 100644
--- a/justfile
+++ b/justfile
@@ -103,6 +103,11 @@ testmon-status:
 test-smoke:
     BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{PYTEST_FLAGS}} -m smoke test-int/mcp/test_smoke_integration.py
 
+# Run the complete semantic read-cache suite against a real Redis server.
+# BASIC_MEMORY_TEST_REDIS_URL may select an existing server; otherwise testcontainers owns one.
+test-read-cache:
+    BASIC_MEMORY_ENV=test LOGFIRE_IGNORE_NO_CONFIG=1 uv run pytest -p pytest_mock --no-cov -q test-int/read_cache
+
 # Fast static check: auto-fix lint, format, and typecheck, but do not run tests.
 fast-check:
     just fix
@@ -403,6 +408,46 @@ semantic-report *args:
 benchmark-compare baseline candidate *args:
     uv run python test-int/compare_search_benchmarks.py "{{baseline}}" "{{candidate}}" --format table {{args}}
 
+# Check the standalone read-load driver in the root environment that supplies Redis/testcontainers.
+bench-read-check:
+    uv run ruff check benchmarks/scripts/read_load_bench.py
+    uv run pyright benchmarks/scripts/read_load_bench.py
+
+# Run one MCP read-load sweep. An empty Redis URL is the authoritative baseline.
+bench-read-load label="authoritative" redis_url="":
+    uv run python benchmarks/scripts/read_load_bench.py \
+        --bm-command .venv/bin/basic-memory \
+        --label "{{label}}" \
+        --redis-url "{{redis_url}}" \
+        --scratch ".scratch/read-load-{{label}}" \
+        --output ".scratch/read-load-{{label}}/results.jsonl" \
+        --truncate
+
+# Exercise MCP startup, Redis pool capacity, warmup, result output, and provenance with a tiny corpus.
+bench-read-smoke redis_url="redis://127.0.0.1:6379/0":
+    uv run python benchmarks/scripts/read_load_bench.py \
+        --bm-command .venv/bin/basic-memory \
+        --label "smoke" \
+        --redis-url "{{redis_url}}" \
+        --scratch ".scratch/read-load-smoke" \
+        --output ".scratch/read-load-smoke/results.jsonl" \
+        --truncate \
+        --sizes 1024 \
+        --notes-per-size 2 \
+        --reads 64 \
+        --concurrency 64 \
+        --seed-concurrency 1 \
+        --quiesce-seconds 0.1
+
+# Run adjacent authoritative and warmed-Redis sweeps, then print their comparison.
+bench-read-cache redis_url="redis://127.0.0.1:6379/0" run_id="latest":
+    just bench-read-load "authoritative-{{run_id}}"
+    just bench-read-load "redis-warm-{{run_id}}" "{{redis_url}}"
+    uv run python test-int/compare_search_benchmarks.py \
+        ".scratch/read-load-authoritative-{{run_id}}/results.jsonl" \
+        ".scratch/read-load-redis-warm-{{run_id}}/results.jsonl" \
+        --format markdown
+
 # Run all tests including Windows, Postgres, and Benchmarks (for CI/comprehensive testing)
 # Use this before releasing to ensure everything works across all backends and platforms
 test-all:
diff --git a/src/basic_memory/api/container.py b/src/basic_memory/api/container.py
index c0c585835..ad1eb4923 100644
--- a/src/basic_memory/api/container.py
+++ b/src/basic_memory/api/container.py
@@ -10,6 +10,8 @@
 - Factories for services are provided, not singletons
 """
 
+from collections.abc import Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
@@ -41,8 +43,8 @@ class ApiContainer:
     session_maker: async_sessionmaker[AsyncSession] | None = None
 
     # --- Optional semantic read cache ---
-    # Hosts inject a namespace-bound implementation; local and off-lifespan
-    # ASGI requests deliberately stay dependency-free by default.
+    # Hosts inject a namespace-bound implementation. Standalone MCP may install
+    # one for its in-process ASGI path; unconfigured callers receive None.
     read_cache: ReadCache | None = None
 
     @classmethod
@@ -155,3 +157,15 @@ def set_container(container: ApiContainer) -> None:
     """Set the API container (called by lifespan)."""
     global _container
     _container = container
+
+
+@contextmanager
+def installed_container(container: ApiContainer) -> Iterator[None]:
+    """Temporarily expose a container to off-lifespan ASGI requests."""
+    global _container
+    previous = _container
+    _container = container
+    try:
+        yield
+    finally:
+        _container = previous
diff --git a/src/basic_memory/api/v2/routers/search_router.py b/src/basic_memory/api/v2/routers/search_router.py
index 83ff1e2bd..1e903d9cb 100644
--- a/src/basic_memory/api/v2/routers/search_router.py
+++ b/src/basic_memory/api/v2/routers/search_router.py
@@ -5,11 +5,30 @@
 """
 
 import asyncio
+import json
+from contextlib import nullcontext
+from typing import Annotated
 
-from fastapi import APIRouter, HTTPException, Path
+from fastapi import APIRouter, Depends, HTTPException, Path, Response
 
 import logfire
 from basic_memory.api.v2.utils import to_search_results
+from basic_memory.deps import (
+    EntityServiceV2ExternalDep,
+    ProjectExternalIdPathDep,
+    ReadCacheDep,
+    SearchReindexSchedulerDep,
+    SearchServiceV2ExternalDep,
+    create_model_read_cache,
+)
+from basic_memory.read_cache import (
+    ModelReadCache,
+    ReadCacheKey,
+    ReadCacheOperation,
+    ReadCacheScope,
+    read_cache_request_digest,
+)
+from basic_memory.read_cache.policy import SEARCH_READ_CACHE_TTL_SECONDS
 from basic_memory.repository.semantic_errors import (
     RerankProviderContractError,
     RerankTransientError,
@@ -17,26 +36,59 @@
     SemanticSearchDisabledError,
 )
 from basic_memory.schemas.search import SearchQuery, SearchResponse, SearchRetrievalMode
-from basic_memory.deps import (
-    SearchServiceV2ExternalDep,
-    EntityServiceV2ExternalDep,
-    SearchReindexSchedulerDep,
-    ProjectExternalIdPathDep,
-)
 
-# Note: No prefix here - it's added during registration as /v2/{project_id}/search
+# App registration mounts this router at /v2/projects/{project_id}.
 router = APIRouter(tags=["search"])
 
 
+def get_search_read_cache(
+    read_cache: ReadCacheDep,
+) -> ModelReadCache[SearchResponse] | None:
+    """Bind search responses to the optional cache backend and short query TTL."""
+    return create_model_read_cache(
+        read_cache,
+        SearchResponse,
+        ttl_seconds=SEARCH_READ_CACHE_TTL_SECONDS,
+    )
+
+
+SearchReadCacheDep = Annotated[
+    ModelReadCache[SearchResponse] | None,
+    Depends(get_search_read_cache),
+]
+
+
+def _search_request_digest(query: SearchQuery, *, page: int, page_size: int) -> str:
+    """Return one structural identity for POST and QUERY search requests."""
+    canonical_request = json.dumps(
+        {
+            "query": query.model_dump(mode="json"),
+            "page": page,
+            "page_size": page_size,
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return read_cache_request_digest(canonical_request)
+
+
+@router.api_route(
+    "/search/",
+    methods=["QUERY"],
+    response_model=SearchResponse,
+    include_in_schema=False,
+)
 @router.post("/search/", response_model=SearchResponse)
 async def search(
     query: SearchQuery,
     search_service: SearchServiceV2ExternalDep,
     entity_service: EntityServiceV2ExternalDep,
+    read_cache: SearchReadCacheDep,
+    response: Response,
     project_id: str = Path(..., description="Project external UUID"),
     page: int = 1,
     page_size: int = 10,
-):
+) -> SearchResponse:
     """Search across all knowledge and documents in a project.
 
     V2 uses external_id UUIDs for stable API references.
@@ -52,6 +104,7 @@ async def search(
     Returns:
         SearchResponse with paginated search results
     """
+    response.headers["Accept-Query"] = "application/json"
     with logfire.span(
         "api.request.search",
         entrypoint="api",
@@ -70,81 +123,99 @@ async def search(
             query.note_types or query.entity_types or query.categories or query.metadata_filters
         ),
     ):
-        offset = (page - 1) * page_size
-        exact_count_available = query.retrieval_mode == SearchRetrievalMode.FTS
-        try:
+        cache_key = ReadCacheKey(
+            project_id=project_id,
+            operation=ReadCacheOperation.search,
+            request_digest=_search_request_digest(query, page=page, page_size=page_size),
+        )
+        cache_scope = (
+            read_cache.read(key=cache_key)
+            if read_cache is not None
+            else nullcontext(ReadCacheScope[SearchResponse]())
+        )
+        async with cache_scope as cached:
+            if cached.value is not None:
+                return cached.value
+
+            offset = (page - 1) * page_size
+            exact_count_available = query.retrieval_mode == SearchRetrievalMode.FTS
+            try:
+                with logfire.span(
+                    "api.search.search.execute_query",
+                    domain="search",
+                    action="search",
+                    phase="execute_query",
+                    page=page,
+                    page_size=page_size,
+                ):
+                    if exact_count_available:
+                        results, total = await asyncio.gather(
+                            search_service.search(query, limit=page_size, offset=offset),
+                            search_service.count(query),
+                        )
+                    else:
+                        results = await search_service.search(
+                            query, limit=page_size + 1, offset=offset
+                        )
+                        total = 0
+            except SemanticSearchDisabledError as exc:
+                raise HTTPException(status_code=400, detail=str(exc)) from exc
+            except SemanticDependenciesMissingError as exc:
+                raise HTTPException(status_code=400, detail=str(exc)) from exc
+            except RerankTransientError as exc:
+                # Returning raw retrieval order would make pagination inconsistent with
+                # earlier reranked pages. Preserve ordering semantics and make the outage
+                # explicitly retryable instead.
+                raise HTTPException(status_code=503, detail=str(exc)) from exc
+            except RerankProviderContractError as exc:
+                # Upstream reranker returned a malformed response — an upstream fault, not a
+                # client error and not a transient outage (those map to a retryable 503).
+                raise HTTPException(status_code=502, detail=str(exc)) from exc
+            except ValueError as exc:
+                raise HTTPException(status_code=400, detail=str(exc)) from exc
+
             with logfire.span(
-                "api.search.search.execute_query",
+                "api.search.search.paginate_results",
                 domain="search",
                 action="search",
-                phase="execute_query",
-                page=page,
-                page_size=page_size,
+                phase="paginate_results",
+                result_count=len(results),
             ):
                 if exact_count_available:
-                    results, total = await asyncio.gather(
-                        search_service.search(query, limit=page_size, offset=offset),
-                        search_service.count(query),
-                    )
+                    has_more = offset + len(results) < total
                 else:
-                    results = await search_service.search(query, limit=page_size + 1, offset=offset)
-                    total = 0
-        except SemanticSearchDisabledError as exc:
-            raise HTTPException(status_code=400, detail=str(exc)) from exc
-        except SemanticDependenciesMissingError as exc:
-            raise HTTPException(status_code=400, detail=str(exc)) from exc
-        except RerankTransientError as exc:
-            # Returning raw retrieval order would make pagination inconsistent with
-            # earlier reranked pages. Preserve ordering semantics and make the outage
-            # explicitly retryable instead.
-            raise HTTPException(status_code=503, detail=str(exc)) from exc
-        except RerankProviderContractError as exc:
-            # Upstream reranker returned a malformed response — an upstream fault, not a
-            # client error and not a transient outage (those map to a retryable 503).
-            raise HTTPException(status_code=502, detail=str(exc)) from exc
-        except ValueError as exc:
-            raise HTTPException(status_code=400, detail=str(exc)) from exc
-
-        with logfire.span(
-            "api.search.search.paginate_results",
-            domain="search",
-            action="search",
-            phase="paginate_results",
-            result_count=len(results),
-        ):
-            if exact_count_available:
-                has_more = offset + len(results) < total
-            else:
-                # Trigger: semantic modes would need another vector/hybrid retrieval to count.
-                # Why: search requests should not pay for a second semantic pass.
-                # Outcome: preserve probe pagination, leave total at 0, and mark it unknown.
-                has_more = len(results) > page_size
-                if has_more:
-                    results = results[:page_size]
-
-        with logfire.span(
-            "api.search.search.hydrate_results",
-            domain="search",
-            action="search",
-            phase="hydrate_results",
-            result_count=len(results),
-        ):
-            search_results = await to_search_results(entity_service, results)
-        with logfire.span(
-            "api.search.search.build_response",
-            domain="search",
-            action="search",
-            phase="build_response",
-            result_count=len(search_results),
-        ):
-            return SearchResponse(
-                results=search_results,
-                current_page=page,
-                page_size=page_size,
-                total=total,
-                total_is_exact=exact_count_available,
-                has_more=has_more,
-            )
+                    # Trigger: semantic modes would need another vector/hybrid retrieval to count.
+                    # Why: search requests should not pay for a second semantic pass.
+                    # Outcome: preserve probe pagination, leave total at 0, and mark it unknown.
+                    has_more = len(results) > page_size
+                    if has_more:
+                        results = results[:page_size]
+
+            with logfire.span(
+                "api.search.search.hydrate_results",
+                domain="search",
+                action="search",
+                phase="hydrate_results",
+                result_count=len(results),
+            ):
+                search_results = await to_search_results(entity_service, results)
+            with logfire.span(
+                "api.search.search.build_response",
+                domain="search",
+                action="search",
+                phase="build_response",
+                result_count=len(search_results),
+            ):
+                result = SearchResponse(
+                    results=search_results,
+                    current_page=page,
+                    page_size=page_size,
+                    total=total,
+                    total_is_exact=exact_count_available,
+                    has_more=has_more,
+                )
+                cached.value = result
+                return result
 
 
 @router.post("/search/reindex")
diff --git a/src/basic_memory/config_models.py b/src/basic_memory/config_models.py
index 8689d4847..2211f3620 100644
--- a/src/basic_memory/config_models.py
+++ b/src/basic_memory/config_models.py
@@ -238,6 +238,16 @@ def __init__(self, **data: Any) -> None: ...
         description="Database connection URL. For Postgres, use postgresql+asyncpg://user:pass@host:port/db. If not set, SQLite will use default path.",
     )
 
+    redis_url: Optional[str] = Field(
+        default=None,
+        description="Optional Redis URL for standalone MCP read caching. Bare hostnames are accepted.",
+    )
+    redis_max_connections: int = Field(
+        default=20,
+        description="Maximum Redis connections used by standalone MCP read caching.",
+        gt=0,
+    )
+
     # Semantic search configuration
     semantic_search_enabled: bool = Field(
         default_factory=_default_semantic_search_enabled,
diff --git a/src/basic_memory/deps/read_cache.py b/src/basic_memory/deps/read_cache.py
index ec81b341e..505391891 100644
--- a/src/basic_memory/deps/read_cache.py
+++ b/src/basic_memory/deps/read_cache.py
@@ -34,6 +34,7 @@ def create_model_read_cache[ModelT: BaseModel](
     read_cache: ReadCache | None,
     model_type: type[ModelT],
     *,
+    ttl_seconds: int = READ_CACHE_TTL_SECONDS,
     max_payload_bytes: int = READ_CACHE_MAX_PAYLOAD_BYTES,
 ) -> ModelReadCache[ModelT] | None:
     """Bind one response model to the host cache and Basic Memory's read policy."""
@@ -42,6 +43,6 @@ def create_model_read_cache[ModelT: BaseModel](
     return ModelReadCache(
         backend=read_cache,
         model_type=model_type,
-        ttl_seconds=READ_CACHE_TTL_SECONDS,
+        ttl_seconds=ttl_seconds,
         max_payload_bytes=max_payload_bytes,
     )
diff --git a/src/basic_memory/deps/services.py b/src/basic_memory/deps/services.py
index 1cf33e030..6e8e62198 100644
--- a/src/basic_memory/deps/services.py
+++ b/src/basic_memory/deps/services.py
@@ -397,11 +397,17 @@ async def get_project_index_observer(
 
 
 async def get_entity_vector_sync_scheduler(
+    project_external_id: Annotated[
+        str, FastAPIPath(alias="project_id", description="Project external UUID")
+    ],
     search_service: SearchServiceV2ExternalDep,
     app_config: AppConfigDep,
+    read_cache: ReadCacheDep,
 ) -> EntityVectorSyncScheduler:
     return LocalEntityVectorSyncScheduler(
         search_service=search_service,
+        project_external_id=project_external_id,
+        read_cache=read_cache,
         test_mode=app_config.is_test_env,
     )
 
diff --git a/src/basic_memory/index/local_schedulers.py b/src/basic_memory/index/local_schedulers.py
index f71dda676..928e96834 100644
--- a/src/basic_memory/index/local_schedulers.py
+++ b/src/basic_memory/index/local_schedulers.py
@@ -9,7 +9,6 @@
 """
 
 import asyncio
-from contextlib import nullcontext
 from dataclasses import dataclass
 from typing import Any, Coroutine
 
@@ -89,15 +88,28 @@ async def drain_background_tasks() -> None:
 @dataclass(frozen=True, slots=True)
 class LocalEntityVectorSyncScheduler:
     search_service: EntityVectorSync
+    project_external_id: str
+    read_cache: ReadCacheInvalidator | None
     test_mode: bool
 
     def schedule_entity_vector_sync(self, *, entity_id: int, project_id: int) -> None:
         _ = project_id
         _schedule_background_coroutine(
-            self.search_service.sync_entity_vectors(entity_id),
+            self._run_entity_vector_sync(entity_id),
             test_mode=self.test_mode,
         )
 
+    async def _run_entity_vector_sync(self, entity_id: int) -> None:
+        if self.read_cache is None:
+            await self.search_service.sync_entity_vectors(entity_id)
+            return
+
+        # Vector publication happens after the mutation/index generation bump and can change
+        # VECTOR/HYBRID results. Advance the generation after success or partial failure so a
+        # result cached while vectors were stale cannot survive this derived-state update.
+        async with invalidate_cache(self.read_cache, self.project_external_id):
+            await self.search_service.sync_entity_vectors(entity_id)
+
 
 # Process-lifetime single-flight state: project ids with an index run already
 # scheduled or in flight. Every POST .../index and the startup scan previously
@@ -168,15 +180,14 @@ def schedule_search_reindex(self, *, project_id: int) -> None:
         )
 
     async def _run_search_reindex(self) -> None:
+        if self.read_cache is None:
+            await self.search_service.reindex_all()
+            return
+
         # A rebuild publishes search rows incrementally after dropping the old
         # index. Invalidate after success or partial failure so fuzzy resolutions
         # cached during that window cannot survive the rebuild.
-        invalidation_scope = (
-            invalidate_cache(self.read_cache, self.project_external_id)
-            if self.read_cache is not None
-            else nullcontext()
-        )
-        async with invalidation_scope:
+        async with invalidate_cache(self.read_cache, self.project_external_id):
             await self.search_service.reindex_all()
 
 
@@ -241,13 +252,11 @@ async def _resolve_after_debounce(self, project_id: int) -> None:
             # Relation resolution commits entity changes after the index pass.
             # A second bump closes the window in which an intermediate entity
             # response could have populated the current generation.
-            invalidation_scope = (
-                invalidate_cache(self.read_cache, self.project_external_id)
-                if self.read_cache is not None
-                else nullcontext()
-            )
-            async with invalidation_scope:
+            if self.read_cache is None:
                 await resolve_project_relations(self.relation_runtime)
+            else:
+                async with invalidate_cache(self.read_cache, self.project_external_id):
+                    await resolve_project_relations(self.relation_runtime)
         finally:
             rerun = project_id in _dirty_relation_resolution
             _dirty_relation_resolution.discard(project_id)
diff --git a/src/basic_memory/mcp/clients/search.py b/src/basic_memory/mcp/clients/search.py
index 6fd3ab047..0d04fda2c 100644
--- a/src/basic_memory/mcp/clients/search.py
+++ b/src/basic_memory/mcp/clients/search.py
@@ -60,7 +60,7 @@ async def search(
         Raises:
             ToolError: If the request fails
         """
-        from basic_memory.mcp.tools.utils import call_post
+        from basic_memory.mcp.tools.utils import call_query
 
         with logfire.span(
             "mcp.client.search.search",
@@ -69,7 +69,7 @@ async def search(
             page=page,
             page_size=page_size,
         ):
-            response = await call_post(
+            response = await call_query(
                 self.http_client,
                 f"{self._base_path}/",
                 json=query,
diff --git a/src/basic_memory/mcp/container.py b/src/basic_memory/mcp/container.py
index f559922ae..0ed8814c8 100644
--- a/src/basic_memory/mcp/container.py
+++ b/src/basic_memory/mcp/container.py
@@ -14,6 +14,7 @@
 from typing import TYPE_CHECKING
 
 from basic_memory.config import BasicMemoryConfig, ConfigManager
+from basic_memory.read_cache import ReadCache
 from basic_memory.runtime.mode import RuntimeMode, resolve_runtime_mode
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -30,6 +31,7 @@ class McpContainer:
 
     config: BasicMemoryConfig
     mode: RuntimeMode
+    read_cache: ReadCache | None = None
 
     @classmethod
     def create(cls) -> "McpContainer":
@@ -83,6 +85,7 @@ def create_watch_coordinator(self) -> "WatchCoordinator":
             config=self.config,
             should_watch=self.should_watch_files,
             skip_reason=self.watch_skip_reason,
+            read_cache=self.read_cache,
         )
 
 
diff --git a/src/basic_memory/mcp/server.py b/src/basic_memory/mcp/server.py
index d1d098c14..85773177c 100644
--- a/src/basic_memory/mcp/server.py
+++ b/src/basic_memory/mcp/server.py
@@ -11,12 +11,16 @@
 from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
 
 from basic_memory import db
+from basic_memory.api.container import ApiContainer, installed_container
 from basic_memory.cli.auth import CLIAuth
 from basic_memory.index.note_content_materialization import drain_pending_materializations
 from basic_memory.db import scoped_session
 from basic_memory.index.local_schedulers import drain_background_tasks
 from basic_memory.mcp.client_info import MCPClientInfoMiddleware
 from basic_memory.mcp.container import McpContainer, set_container
+from basic_memory.read_cache import ReadCache, ReadCacheUnavailable
+from basic_memory.read_cache.lifecycle import open_redis_read_cache
+from basic_memory.repository import ProjectRepository
 from basic_memory.services.initialization import initialize_app
 import logfire
 
@@ -51,6 +55,31 @@ async def _log_embedding_status(session_maker: async_sessionmaker[AsyncSession])
         logger.debug(f"Could not check embedding status at startup: {exc}")
 
 
+async def _invalidate_persisted_read_cache(
+    read_cache: ReadCache,
+    session_maker: async_sessionmaker[AsyncSession],
+) -> ReadCache | None:
+    """Invalidate cache generations that may predate offline file changes."""
+    project_repository = ProjectRepository()
+    async with scoped_session(session_maker) as session:
+        active_projects = await project_repository.get_active_projects(session)
+
+    try:
+        for project in active_projects:
+            await read_cache.invalidate_project(str(project.external_id))
+    except ReadCacheUnavailable as error:
+        # Trigger: Redis cannot invalidate every persisted project generation at startup.
+        # Why: reusing any prior generation could hide an offline file edit until TTL expiry.
+        # Outcome: disable caching for this lifespan and serve only authoritative reads.
+        logger.error(
+            "Standalone Redis read cache disabled because startup invalidation was unavailable",
+            error=str(error),
+        )
+        return None
+
+    return read_cache
+
+
 @asynccontextmanager
 async def lifespan(app: FastMCP):
     """Lifecycle manager for the MCP server.
@@ -63,95 +92,119 @@ async def lifespan(app: FastMCP):
     # --- Composition Root ---
     # Create container and read config (single point of config access)
     container = McpContainer.create()
-    set_container(container)
-
     config = container.config
-    with logfire.span(
-        "mcp.lifecycle.startup",
-        entrypoint="mcp",
-        mode=container.mode.name.lower(),
-        default_project=config.default_project,
-    ):
-        logger.info(f"Starting Basic Memory MCP server (mode={container.mode.name})")
-        logger.info(
-            f"Config: database_backend={config.database_backend.value}, "
-            f"semantic_search_enabled={config.semantic_search_enabled}, "
-            f"default_project={config.default_project}"
-        )
-        if config.semantic_search_enabled:
-            logger.info(
-                f"Semantic search: provider={config.semantic_embedding_provider}, "
-                f"model={config.semantic_embedding_model}, "
-                f"dimensions={config.semantic_embedding_dimensions or 'auto'}, "
-                f"batch_size={config.semantic_embedding_batch_size}, "
-                f"document_prefix_set={bool(config.semantic_embedding_document_prefix)}, "
-                f"query_prefix_set={bool(config.semantic_embedding_query_prefix)}"
-            )
-
-        # Log configured projects with their routing mode
-        for name, entry in config.projects.items():
-            default = " (default)" if name == config.default_project else ""
-            logger.info(f"Project: {name} -> {entry.path} [mode={entry.mode.value}]{default}")
-
-        # Check cloud auth status (local file check, no network call)
-        auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
-        tokens = auth.load_tokens()
-        if tokens is not None:
-            if not auth.is_token_valid(tokens):
-                expires_at = tokens.get("expires_at", 0)
-                expired_ago = int(time.time() - expires_at)
-                logger.warning(
-                    f"Cloud token expired {expired_ago}s ago - may need 'bm cloud login'"
+    standalone_redis_url = None if container.mode.is_cloud else config.redis_url
+
+    async with open_redis_read_cache(
+        standalone_redis_url,
+        max_connections=config.redis_max_connections,
+    ) as read_cache:
+        container.read_cache = read_cache
+        set_container(container)
+        api_container = ApiContainer(config=config, mode=container.mode, read_cache=read_cache)
+
+        # Local MCP requests use FastAPI through ASGITransport without entering its lifespan.
+        # Installing the same container keeps request reads and watcher invalidation on one cache.
+        with installed_container(api_container):
+            with logfire.span(
+                "mcp.lifecycle.startup",
+                entrypoint="mcp",
+                mode=container.mode.name.lower(),
+                default_project=config.default_project,
+            ):
+                logger.info(f"Starting Basic Memory MCP server (mode={container.mode.name})")
+                logger.info(
+                    f"Config: database_backend={config.database_backend.value}, "
+                    f"semantic_search_enabled={config.semantic_search_enabled}, "
+                    f"default_project={config.default_project}"
                 )
-            else:
-                logger.info("Cloud: authenticated (OAuth token valid)")
-
-        if config.cloud_api_key:
-            logger.info("Cloud: API key configured")
-
-        # Track if we created the engine (vs test fixtures providing it)
-        # This prevents disposing an engine provided by test fixtures when
-        # multiple Client connections are made in the same test
-        engine_was_none = db._engine is None
-
-        # Initialize app (runs migrations, reconciles projects)
-        await initialize_app(container.config)
-
-        # Log embedding status so it's easy to spot in the logs
-        if config.semantic_search_enabled and db._session_maker is not None:
-            await _log_embedding_status(db._session_maker)
-
-        # Create and start local watch coordinator (lifecycle centralized in coordinator)
-        watch_coordinator = container.create_watch_coordinator()
-        await watch_coordinator.start()
-
-    try:
-        yield
-    finally:
-        # Shutdown - coordinator handles clean task cancellation
-        with logfire.span(
-            "mcp.lifecycle.shutdown",
-            entrypoint="mcp",
-            mode=container.mode.name.lower(),
-        ):
-            logger.debug("Shutting down Basic Memory MCP server")
-
-            await watch_coordinator.stop()
-
-            # A local note write returns 202 before its markdown file is written;
-            # shutdown can land while that materialization (and the vector sync /
-            # relation resolution it schedules) is still queued in the in-process
-            # pool. Drain both queues before the engine closes so an accepted
-            # write is never lost — mirrors the API lifespan shutdown.
-            await drain_pending_materializations()
-            await drain_background_tasks()
-
-            # Only shutdown DB if we created it (not if test fixture provided it)
-            if engine_was_none:
-                await db.shutdown_db()
-                logger.debug("Database connections closed")
-            else:  # pragma: no cover
-                logger.debug("Skipping DB shutdown - engine provided externally")
+                if config.semantic_search_enabled:
+                    logger.info(
+                        f"Semantic search: provider={config.semantic_embedding_provider}, "
+                        f"model={config.semantic_embedding_model}, "
+                        f"dimensions={config.semantic_embedding_dimensions or 'auto'}, "
+                        f"batch_size={config.semantic_embedding_batch_size}, "
+                        f"document_prefix_set={bool(config.semantic_embedding_document_prefix)}, "
+                        f"query_prefix_set={bool(config.semantic_embedding_query_prefix)}"
+                    )
+
+                # Log configured projects with their routing mode
+                for name, entry in config.projects.items():
+                    default = " (default)" if name == config.default_project else ""
+                    logger.info(
+                        f"Project: {name} -> {entry.path} [mode={entry.mode.value}]{default}"
+                    )
+
+                # Check cloud auth status (local file check, no network call)
+                auth = CLIAuth(client_id=config.cloud_client_id, authkit_domain=config.cloud_domain)
+                tokens = auth.load_tokens()
+                if tokens is not None:
+                    if not auth.is_token_valid(tokens):
+                        expires_at = tokens.get("expires_at", 0)
+                        expired_ago = int(time.time() - expires_at)
+                        logger.warning(
+                            f"Cloud token expired {expired_ago}s ago - may need 'bm cloud login'"
+                        )
+                    else:
+                        logger.info("Cloud: authenticated (OAuth token valid)")
+
+                if config.cloud_api_key:
+                    logger.info("Cloud: API key configured")
+
+                # Track if we created the engine (vs test fixtures providing it)
+                # This prevents disposing an engine provided by test fixtures when
+                # multiple Client connections are made in the same test
+                engine_was_none = db._engine is None
+
+                # Initialize app (runs migrations, reconciles projects)
+                await initialize_app(container.config)
+
+                if read_cache is not None:
+                    assert db._session_maker is not None, (
+                        "Database session maker missing after MCP initialization"
+                    )
+                    read_cache = await _invalidate_persisted_read_cache(
+                        read_cache,
+                        db._session_maker,
+                    )
+                    container.read_cache = read_cache
+                    api_container.read_cache = read_cache
+
+                # Log embedding status so it's easy to spot in the logs
+                if config.semantic_search_enabled and db._session_maker is not None:
+                    await _log_embedding_status(db._session_maker)
+
+                # Create and start local watch coordinator (lifecycle centralized in coordinator)
+                watch_coordinator = container.create_watch_coordinator()
+                await watch_coordinator.start()
+
+            try:
+                yield
+            finally:
+                # Shutdown - coordinator handles clean task cancellation
+                with logfire.span(
+                    "mcp.lifecycle.shutdown",
+                    entrypoint="mcp",
+                    mode=container.mode.name.lower(),
+                ):
+                    logger.debug("Shutting down Basic Memory MCP server")
+
+                    await watch_coordinator.stop()
+
+                    # A local note write returns 202 before its markdown file is written;
+                    # shutdown can land while that materialization (and the vector sync /
+                    # relation resolution it schedules) is still queued in the in-process
+                    # pool. Drain both queues before the engine closes so an accepted
+                    # write is never lost — mirrors the API lifespan shutdown.
+                    await drain_pending_materializations()
+                    await drain_background_tasks()
+
+                    # Only shutdown DB if we created it (not if test fixture provided it)
+                    if engine_was_none:
+                        await db.shutdown_db()
+                        logger.debug("Database connections closed")
+                    else:  # pragma: no cover
+                        logger.debug("Skipping DB shutdown - engine provided externally")
 
 
 # A newly-connected model only sees the server `instructions` for free — everything else
diff --git a/src/basic_memory/mcp/tools/utils.py b/src/basic_memory/mcp/tools/utils.py
index 8c5078b26..87a11a7be 100644
--- a/src/basic_memory/mcp/tools/utils.py
+++ b/src/basic_memory/mcp/tools/utils.py
@@ -656,6 +656,88 @@ async def call_post(
         raise
 
 
+async def call_query(
+    client: AsyncClient,
+    url: URL | str,
+    *,
+    client_name: str | None = None,
+    operation: str | None = None,
+    path_template: str | None = None,
+    content: RequestContent | None = None,
+    data: RequestData | None = None,
+    files: RequestFiles | None = None,
+    json: typing.Any | None = None,
+    params: QueryParamTypes | None = None,
+    headers: HeaderTypes | None = None,
+    cookies: CookieTypes | None = None,
+    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
+    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    extensions: RequestExtensions | None = None,
+) -> Response:
+    """Make a safe, idempotent QUERY request and handle errors appropriately."""
+    logger.debug(f"Calling QUERY '{url}'")
+    error_message = None
+    request_span: logfire.LogfireSpan | None = None
+
+    try:
+        with logfire.span(
+            "mcp.http.request",
+            method="QUERY",
+            client_name=client_name,
+            operation=operation,
+            path_template=path_template,
+            phase="request",
+            has_query=bool(params),
+            has_body=any(value is not None for value in (content, data, files, json)),
+        ) as request_span:
+            response = await client.request(
+                "QUERY",
+                url=url,
+                content=content,
+                data=data,
+                files=files,
+                json=json,
+                params=params,
+                headers=_request_headers(headers),
+                cookies=cookies,
+                auth=auth,
+                follow_redirects=follow_redirects,
+                timeout=timeout,
+                extensions=extensions,
+            )
+            request_span.set_attributes(_response_span_attrs(response))
+
+        if response.is_success:
+            return response
+
+        status_code = response.status_code
+        response_data = _extract_response_data(response)
+        error_message = _resolve_error_message(status_code, url, "QUERY", response_data)
+
+        if 400 <= status_code < 500:
+            if status_code == 429:  # pragma: no cover
+                logger.warning(f"Rate limit exceeded: QUERY {url}: {error_message}")
+            else:
+                logger.info(f"Client error: QUERY {url}: {error_message}")
+        else:
+            logger.error(f"Server error: QUERY {url}: {error_message}")
+
+        response.raise_for_status()
+        return response  # pragma: no cover
+
+    except HTTPStatusError as e:
+        raise ToolError(error_message) from e
+    except TransportError as e:
+        if request_span is not None:
+            request_span.set_attributes(_transport_error_span_attrs(e))
+        raise ToolError(_transport_error_message(e, url, "QUERY")) from e
+    except Exception as e:
+        if request_span is not None:
+            request_span.set_attributes(_transport_error_span_attrs(e))
+        raise
+
+
 async def resolve_entity_id(client: AsyncClient, project_external_id: str, identifier: str) -> str:
     """Resolve a string identifier to an entity external_id using the v2 API.
 
diff --git a/src/basic_memory/read_cache/contract.py b/src/basic_memory/read_cache/contract.py
index 1dc99dad4..ecc1f318f 100644
--- a/src/basic_memory/read_cache/contract.py
+++ b/src/basic_memory/read_cache/contract.py
@@ -15,6 +15,7 @@ class ReadCacheOperation(StrEnum):
     entity = "entity"
     resolve = "resolve"
     resource = "resource"
+    search = "search"
 
 
 class ReadCacheStoreStatus(StrEnum):
diff --git a/src/basic_memory/read_cache/lifecycle.py b/src/basic_memory/read_cache/lifecycle.py
new file mode 100644
index 000000000..95eb7f5ae
--- /dev/null
+++ b/src/basic_memory/read_cache/lifecycle.py
@@ -0,0 +1,46 @@
+"""Lifecycle helpers for the optional standalone Redis read cache."""
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from loguru import logger
+
+from basic_memory.read_cache.contract import ReadCache
+
+
+STANDALONE_CACHE_NAMESPACE = "standalone"
+
+
+def normalize_redis_url(redis_url: str) -> str:
+    """Accept either a complete Redis URL or a bare Redis hostname."""
+    normalized = redis_url.strip()
+    if "://" in normalized:
+        return normalized
+    return f"redis://{normalized}"
+
+
+@asynccontextmanager
+async def open_redis_read_cache(
+    redis_url: str | None,
+    *,
+    max_connections: int = 20,
+) -> AsyncIterator[ReadCache | None]:
+    """Open one process-owned Redis cache when standalone caching is configured."""
+    if max_connections <= 0:
+        raise ValueError("Redis max_connections must be positive")
+    if redis_url is None or not redis_url.strip():
+        yield None
+        return
+
+    # Importing the adapter only when configured preserves Redis as an optional dependency.
+    from basic_memory.read_cache.redis import RedisReadCache, create_redis_read_cache_client
+
+    client = create_redis_read_cache_client(
+        normalize_redis_url(redis_url),
+        max_connections=max_connections,
+    )
+    logger.info("Standalone Redis read cache enabled")
+    try:
+        yield RedisReadCache(client=client, namespace=STANDALONE_CACHE_NAMESPACE)
+    finally:
+        await client.aclose()
diff --git a/src/basic_memory/read_cache/policy.py b/src/basic_memory/read_cache/policy.py
index b269120cf..133c208e3 100644
--- a/src/basic_memory/read_cache/policy.py
+++ b/src/basic_memory/read_cache/policy.py
@@ -3,3 +3,4 @@
 READ_CACHE_TTL_SECONDS = 60
 READ_CACHE_MAX_PAYLOAD_BYTES = 1024 * 1024
 DIRECTORY_READ_CACHE_MAX_PAYLOAD_BYTES = 2 * 1024 * 1024
+SEARCH_READ_CACHE_TTL_SECONDS = 30
diff --git a/src/basic_memory/redaction.py b/src/basic_memory/redaction.py
index a874c1174..bc081c2d8 100644
--- a/src/basic_memory/redaction.py
+++ b/src/basic_memory/redaction.py
@@ -24,6 +24,7 @@
     {
         "database_url",
         "milvus_uri",
+        "redis_url",
         "reranker_api_base",
         "semantic_embedding_api_base",
     }
@@ -65,8 +66,21 @@ def redact_url(url: str) -> str:
 
     Replaces any credentials with *** so the host/path remain visible for
     diagnostics (e.g. ``postgresql://***@localhost/mydb``).  If the value
-    cannot be parsed as a URL it is returned unchanged.
+    cannot be parsed as a URL it is returned unchanged. Scheme-less authorities
+    are handled explicitly because ``urlparse`` otherwise treats their userinfo
+    as a path and can expose credentials from supported bare Redis values.
     """
+    if "://" not in url:
+        base, query_separator, query = url.partition("?")
+        redacted_query = _redact_query_secrets(query)
+        redacted_base = base
+        if "@" in base:
+            _, _, authority = base.rpartition("@")
+            redacted_base = f"***@{authority}"
+        if redacted_base == base and redacted_query == query:
+            return url
+        return f"{redacted_base}?{redacted_query}" if query_separator else redacted_base
+
     try:
         parsed = urlparse(url)
     except ValueError:
diff --git a/test-int/read_cache/test_mcp_read_cache.py b/test-int/read_cache/test_mcp_read_cache.py
new file mode 100644
index 000000000..4735ccafb
--- /dev/null
+++ b/test-int/read_cache/test_mcp_read_cache.py
@@ -0,0 +1,152 @@
+"""Real Redis coverage for the standalone MCP cache lifecycle."""
+
+from pathlib import Path
+
+import pytest
+from fastmcp import Client
+from redis.asyncio import Redis
+
+from basic_memory.api.container import resolve_container
+from basic_memory.mcp.container import get_container as get_mcp_container
+from basic_memory.read_cache.keys import (
+    DEFAULT_READ_CACHE_PREFIX,
+    redis_read_cache_generation_key,
+)
+from basic_memory.read_cache.lifecycle import STANDALONE_CACHE_NAMESPACE
+
+
+def _reset_config_cache() -> None:
+    from basic_memory import config as config_module
+
+    config_module._CONFIG_CACHE = None
+    config_module._CONFIG_MTIME = None
+    config_module._CONFIG_SIZE = None
+
+
+@pytest.mark.asyncio
+async def test_mcp_uses_configured_redis_for_requests_and_invalidation(
+    redis_url,
+    mcp_server,
+    app,
+    test_project,
+    monkeypatch,
+) -> None:
+    """The MCP lifespan shares one real cache with FastAPI and its watcher."""
+    monkeypatch.setenv("BASIC_MEMORY_REDIS_URL", redis_url)
+    # Fixtures load config before the test body can set its process-start environment.
+    # Reset the process cache so the MCP startup observes the same state as a fresh `bm mcp`.
+    _reset_config_cache()
+
+    redis = Redis.from_url(redis_url, decode_responses=False)
+    generation_key = redis_read_cache_generation_key(
+        prefix=DEFAULT_READ_CACHE_PREFIX,
+        namespace=STANDALONE_CACHE_NAMESPACE,
+        project_id=str(test_project.external_id),
+    )
+    project_key_pattern = f"{generation_key.removesuffix(':generation')}:*"
+
+    try:
+        existing_keys = [key async for key in redis.scan_iter(match=project_key_pattern)]
+        if existing_keys:
+            await redis.delete(*existing_keys)
+
+        async with Client(mcp_server) as client:
+            mcp_container = get_mcp_container()
+            assert mcp_container.read_cache is not None
+            assert resolve_container().read_cache is mcp_container.read_cache
+
+            created = await client.call_tool(
+                "write_note",
+                {
+                    "project": test_project.name,
+                    "title": "Standalone Redis Cache",
+                    "directory": "cache",
+                    "content": "# Standalone Redis Cache\n\nA cached MCP read.",
+                },
+            )
+            assert created.is_error is False
+
+            for _ in range(2):
+                result = await client.call_tool(
+                    "read_note",
+                    {
+                        "project": test_project.name,
+                        "identifier": "cache/standalone-redis-cache",
+                        "output_format": "json",
+                    },
+                )
+                assert result.is_error is False
+
+            cache_keys = [key async for key in redis.scan_iter(match=project_key_pattern)]
+            assert generation_key.encode() in cache_keys
+            assert len(cache_keys) > 1
+    finally:
+        cache_keys = [key async for key in redis.scan_iter(match=project_key_pattern)]
+        if cache_keys:
+            await redis.delete(*cache_keys)
+        await redis.aclose()
+
+
+@pytest.mark.asyncio
+async def test_mcp_restart_invalidates_persisted_project_generation(
+    redis_url,
+    mcp_server,
+    test_project,
+    monkeypatch,
+) -> None:
+    """A fresh MCP lifespan cannot reuse cache entries created before offline edits."""
+    monkeypatch.setenv("BASIC_MEMORY_REDIS_URL", redis_url)
+    _reset_config_cache()
+
+    redis = Redis.from_url(redis_url, decode_responses=False)
+    generation_key = redis_read_cache_generation_key(
+        prefix=DEFAULT_READ_CACHE_PREFIX,
+        namespace=STANDALONE_CACHE_NAMESPACE,
+        project_id=str(test_project.external_id),
+    )
+    project_key_pattern = f"{generation_key.removesuffix(':generation')}:*"
+
+    try:
+        existing_keys = [key async for key in redis.scan_iter(match=project_key_pattern)]
+        if existing_keys:
+            await redis.delete(*existing_keys)
+
+        async with Client(mcp_server) as client:
+            created = await client.call_tool(
+                "write_note",
+                {
+                    "project": test_project.name,
+                    "title": "Offline Restart",
+                    "directory": "cache",
+                    "content": "# Offline Restart\n\nBefore restart.",
+                },
+            )
+            assert created.is_error is False
+            read = await client.call_tool(
+                "read_note",
+                {
+                    "project": test_project.name,
+                    "identifier": "cache/offline-restart",
+                    "output_format": "json",
+                },
+            )
+            assert read.is_error is False
+            generation_before_restart = await redis.get(generation_key)
+            assert generation_before_restart is not None
+
+        note_path = Path(test_project.path) / "cache" / "Offline Restart.md"
+        assert note_path.is_file()
+        note_path.write_text(
+            "# Offline Restart\n\nEdited while MCP was stopped.\n", encoding="utf-8"
+        )
+
+        async with Client(mcp_server):
+            generation_after_restart = await redis.get(generation_key)
+
+        assert generation_after_restart is not None
+        assert generation_after_restart != generation_before_restart
+    finally:
+        cache_keys = [key async for key in redis.scan_iter(match=project_key_pattern)]
+        if cache_keys:
+            await redis.delete(*cache_keys)
+        await redis.aclose()
diff --git a/test-int/read_cache/test_search_read_cache.py b/test-int/read_cache/test_search_read_cache.py
new file mode 100644
index 000000000..c4270b56e
--- /dev/null
+++ b/test-int/read_cache/test_search_read_cache.py
@@ -0,0 +1,193 @@
+"""Real Redis coverage for semantic search response caching."""
+
+from typing import Any, Protocol
+
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient
+from redis.asyncio import Redis
+
+from basic_memory.deps.read_cache import get_read_cache
+from basic_memory.deps.services import get_search_service_v2_external
+from basic_memory.index.local_schedulers import LocalEntityVectorSyncScheduler
+from basic_memory.models import Project
+from basic_memory.read_cache import ReadCacheKey, ReadCacheOperation
+from basic_memory.read_cache.redis import RedisReadCache
+
+
+class RedisCacheHarness(Protocol):
+    """Structural type for the real Redis fixture."""
+
+    cache: RedisReadCache
+    client: Redis
+    namespace: str
+    prefix: str
+
+
+pytestmark = pytest.mark.redis
+
+
+class CountingSearchService:
+    """Authoritative empty search whose calls expose cache behavior."""
+
+    def __init__(self) -> None:
+        self.search_calls = 0
+        self.count_calls = 0
+
+    async def search(self, query: Any, *, limit: int, offset: int) -> list[Any]:
+        del query, limit, offset
+        self.search_calls += 1
+        return []
+
+    async def count(self, query: Any) -> int:
+        del query
+        self.count_calls += 1
+        return 0
+
+
+class PartiallyFailingVectorSync:
+    """Vector publisher that fails after derived state may have changed."""
+
+    def __init__(self) -> None:
+        self.synced_entity_ids: list[int] = []
+
+    async def sync_entity_vectors(self, entity_id: int) -> None:
+        self.synced_entity_ids.append(entity_id)
+        raise RuntimeError("vector publication failed")
+
+
+@pytest.mark.asyncio
+async def test_post_and_query_share_canonical_real_redis_entry(
+    app: FastAPI,
+    client: AsyncClient,
+    test_project: Project,
+    redis_cache: RedisCacheHarness,
+) -> None:
+    """Method aliases and JSON map order resolve to one project-scoped search value."""
+    search_service = CountingSearchService()
+    app.dependency_overrides[get_read_cache] = lambda: redis_cache.cache
+    app.dependency_overrides[get_search_service_v2_external] = lambda: search_service
+
+    project_external_id = str(test_project.external_id)
+    search_url = f"/v2/projects/{project_external_id}/search/"
+    pagination = {"page": 1, "page_size": 10}
+    post_response = await client.post(
+        search_url,
+        params=pagination,
+        json={
+            "text": "canonical cache",
+            "metadata_filters": {
+                "status": "open",
+                "nested": {"z": 2, "a": 1},
+            },
+        },
+    )
+    query_response = await client.request(
+        "QUERY",
+        search_url,
+        params=pagination,
+        json={
+            "metadata_filters": {
+                "nested": {"a": 1, "z": 2},
+                "status": "open",
+            },
+            "text": "canonical cache",
+        },
+    )
+
+    assert post_response.status_code == 200
+    assert query_response.status_code == 200
+    assert query_response.json() == post_response.json()
+    assert query_response.headers["Accept-Query"] == "application/json"
+    assert search_service.search_calls == 1
+    assert search_service.count_calls == 1
+
+    search_keys = [
+        key async for key in redis_cache.client.scan_iter(match=f"{redis_cache.prefix}:*:search:*")
+    ]
+    assert len(search_keys) == 1
+    search_ttl = await redis_cache.client.ttl(search_keys[0])
+    assert 0 < search_ttl <= 30
+
+    await redis_cache.cache.invalidate_project(project_external_id)
+    refreshed_response = await client.request(
+        "QUERY",
+        search_url,
+        params=pagination,
+        json={
+            "text": "canonical cache",
+            "metadata_filters": {
+                "status": "open",
+                "nested": {"z": 2, "a": 1},
+            },
+        },
+    )
+
+    assert refreshed_response.status_code == 200
+    assert search_service.search_calls == 2
+    assert search_service.count_calls == 2
+
+
+@pytest.mark.asyncio
+async def test_search_pagination_uses_distinct_real_redis_entries(
+    app: FastAPI,
+    client: AsyncClient,
+    test_project: Project,
+    redis_cache: RedisCacheHarness,
+) -> None:
+    """Pagination changes result identity even when the query document is unchanged."""
+    search_service = CountingSearchService()
+    app.dependency_overrides[get_read_cache] = lambda: redis_cache.cache
+    app.dependency_overrides[get_search_service_v2_external] = lambda: search_service
+
+    search_url = f"/v2/projects/{test_project.external_id}/search/"
+    first_page = await client.request(
+        "QUERY",
+        search_url,
+        params={"page": 1, "page_size": 10},
+        json={"text": "pagination"},
+    )
+    second_page = await client.request(
+        "QUERY",
+        search_url,
+        params={"page": 2, "page_size": 10},
+        json={"text": "pagination"},
+    )
+
+    assert first_page.status_code == 200
+    assert second_page.status_code == 200
+    assert search_service.search_calls == 2
+    assert search_service.count_calls == 2
+    search_keys = [
+        key async for key in redis_cache.client.scan_iter(match=f"{redis_cache.prefix}:*:search:*")
+    ]
+    assert len(search_keys) == 2
+
+
+@pytest.mark.asyncio
+async def test_partial_vector_publication_invalidates_real_redis_generation(
+    test_project: Project,
+    redis_cache: RedisCacheHarness,
+) -> None:
+    """A failed background vector sync cannot preserve cached vector or hybrid results."""
+    project_external_id = str(test_project.external_id)
+    cache_key = ReadCacheKey(
+        project_id=project_external_id,
+        operation=ReadCacheOperation.search,
+        request_digest="0" * 64,
+    )
+    generation_before = (await redis_cache.cache.lookup(cache_key)).generation
+    vector_sync = PartiallyFailingVectorSync()
+    scheduler = LocalEntityVectorSyncScheduler(
+        search_service=vector_sync,
+        project_external_id=project_external_id,
+        read_cache=redis_cache.cache,
+        test_mode=False,
+    )
+
+    with pytest.raises(RuntimeError, match="vector publication failed"):
+        await scheduler._run_entity_vector_sync(entity_id=42)
+
+    generation_after = (await redis_cache.cache.lookup(cache_key)).generation
+    assert vector_sync.synced_entity_ids == [42]
+    assert generation_after != generation_before
diff --git a/tests/api/test_container.py b/tests/api/test_container.py
index b7f22e33a..9a36d37f2 100644
--- a/tests/api/test_container.py
+++ b/tests/api/test_container.py
@@ -1,6 +1,6 @@
 """Tests for the API container composition root."""
 
-from basic_memory.api.container import ApiContainer
+from basic_memory.api.container import ApiContainer, installed_container, resolve_container
 from basic_memory.runtime.mode import RuntimeMode
 
 
@@ -32,3 +32,17 @@ def test_should_not_watch_files_in_cloud_mode(self, app_config):
         container = ApiContainer(config=app_config, mode=RuntimeMode.CLOUD)
         assert container.should_watch_files is False
         assert container.watch_skip_reason == "Cloud mode enabled"
+
+
+def test_installed_container_restores_previous_container(app_config, monkeypatch):
+    """Off-lifespan hosts can expose dependencies without replacing API ownership."""
+    import basic_memory.api.container as container_module
+
+    previous = ApiContainer(config=app_config, mode=RuntimeMode.LOCAL)
+    current = ApiContainer(config=app_config, mode=RuntimeMode.TEST)
+    monkeypatch.setattr(container_module, "_container", previous)
+
+    with installed_container(current):
+        assert resolve_container() is current
+
+    assert resolve_container() is previous
diff --git a/tests/api/v2/test_search_router.py b/tests/api/v2/test_search_router.py
index 8d885296c..95ea7342c 100644
--- a/tests/api/v2/test_search_router.py
+++ b/tests/api/v2/test_search_router.py
@@ -72,6 +72,25 @@ async def test_search_entities(
     assert "page_size" in data
 
 
+@pytest.mark.asyncio
+async def test_query_search_uses_same_contract_and_advertises_media_type(
+    client: AsyncClient,
+    app,
+    v2_project_url: str,
+):
+    """QUERY is canonical while POST remains the documented compatibility route."""
+    response = await client.request(
+        "QUERY",
+        f"{v2_project_url}/search/",
+        json={"text": "safe search"},
+    )
+
+    assert response.status_code == 200
+    assert response.headers["Accept-Query"] == "application/json"
+    search_path = app.openapi()["paths"]["/v2/projects/{project_id}/search/"]
+    assert set(search_path) == {"post"}
+
+
 @pytest.mark.asyncio
 async def test_search_with_pagination(
     client: AsyncClient,
diff --git a/tests/api/v2/test_search_router_telemetry.py b/tests/api/v2/test_search_router_telemetry.py
index f4d2e7e54..a3c33a0f1 100644
--- a/tests/api/v2/test_search_router_telemetry.py
+++ b/tests/api/v2/test_search_router_telemetry.py
@@ -8,6 +8,7 @@
 
 import logfire
 import pytest
+from fastapi import Response
 
 from basic_memory.schemas.search import SearchQuery
 
@@ -37,16 +38,20 @@ async def fake_to_search_results(entity_service, results):
     monkeypatch.setattr(logfire, "span", fake_span)
     monkeypatch.setattr(router, "to_search_results", fake_to_search_results)
 
-    response = await router.search(
-        SearchQuery(text="hello world"),
-        FakeSearchService(),
-        object(),
-        project_id=123,
+    http_response = Response()
+    search_response = await router.search(
+        query=SearchQuery(text="hello world"),
+        search_service=FakeSearchService(),
+        entity_service=object(),
+        read_cache=None,
+        response=http_response,
+        project_id="11111111-1111-1111-1111-111111111111",
         page=2,
         page_size=5,
     )
 
-    assert response.current_page == 2
+    assert search_response.current_page == 2
+    assert http_response.headers["Accept-Query"] == "application/json"
     # The root span fires with these attrs; additional nested spans may fire as well.
     assert operations[0] == (
         "api.request.search",
diff --git a/tests/cli/test_config_command.py b/tests/cli/test_config_command.py
index eb39ab2c8..c909f02bc 100644
--- a/tests/cli/test_config_command.py
+++ b/tests/cli/test_config_command.py
@@ -327,6 +327,41 @@ def test_config_list_redacts_reranker_credentials(runner, write_config):
     )
 
 
+@pytest.mark.parametrize(
+    ("redis_url", "expected"),
+    [
+        (
+            "redis://cache-user:cache-password@redis.internal:6379/0"
+            "?password=query-secret&timeout=30",
+            "redis://***@redis.internal:6379/0?password=%2A%2A%2A&timeout=30",
+        ),
+        (
+            "cache-user:cache-password@redis.internal:6379/0?password=query-secret&timeout=30",
+            "***@redis.internal:6379/0?password=%2A%2A%2A&timeout=30",
+        ),
+    ],
+)
+def test_config_get_and_list_redact_redis_credentials(
+    runner,
+    write_config,
+    redis_url: str,
+    expected: str,
+):
+    write_config(_base_config(redis_url=redis_url))
+
+    get_result = runner.invoke(app, ["config", "get", "redis_url"])
+    list_result = runner.invoke(app, ["config", "list", "--json"])
+
+    assert get_result.exit_code == 0, get_result.output
+    assert list_result.exit_code == 0, list_result.output
+    for secret in ("cache-user", "cache-password", "query-secret"):
+        assert secret not in get_result.output
+        assert secret not in list_result.output
+    rows = {row["key"]: row for row in json.loads(list_result.output)}
+    assert get_result.output.split("redis_url = ", 1)[1].strip() == expected
+    assert rows["redis_url"]["value"] == expected
+
+
 def test_config_set_never_prints_reranker_api_key(runner, write_config):
     config_file = write_config(_base_config())
 
diff --git a/tests/index/test_local_schedulers.py b/tests/index/test_local_schedulers.py
index c60872471..8c17a03e4 100644
--- a/tests/index/test_local_schedulers.py
+++ b/tests/index/test_local_schedulers.py
@@ -1,7 +1,7 @@
 """Typed scheduler tests for derived async work."""
 
 import asyncio
-from typing import cast
+from typing import cast, override
 
 import pytest
 
@@ -57,15 +57,44 @@ async def invalidate_project(self, project_id: str) -> ReadCacheInvalidationStat
 async def test_entity_vector_scheduler_maps_to_search_service():
     """Entity vector scheduling should call the semantic vector sync method."""
     search_service = StubSearchService()
+    read_cache = RecordingReadCache()
 
     scheduler = LocalEntityVectorSyncScheduler(
         search_service=search_service,
+        project_external_id=PROJECT_EXTERNAL_ID,
+        read_cache=read_cache,
         test_mode=False,
     )
     scheduler.schedule_entity_vector_sync(entity_id=7, project_id=13)
     await asyncio.sleep(0.05)
 
     assert search_service.vector_synced == [7]
+    assert read_cache.invalidated_project_ids == [PROJECT_EXTERNAL_ID]
+
+
+@pytest.mark.asyncio
+async def test_entity_vector_scheduler_invalidates_after_partial_failure():
+    """A failed vector publication may still have changed derived search state."""
+
+    class FailingSearchService(StubSearchService):
+        @override
+        async def sync_entity_vectors(self, entity_id: int) -> None:
+            self.vector_synced.append(entity_id)
+            raise RuntimeError("vector publication failed")
+
+    search_service = FailingSearchService()
+    read_cache = RecordingReadCache()
+    scheduler = LocalEntityVectorSyncScheduler(
+        search_service=search_service,
+        project_external_id=PROJECT_EXTERNAL_ID,
+        read_cache=read_cache,
+        test_mode=False,
+    )
+
+    with pytest.raises(RuntimeError, match="vector publication failed"):
+        await scheduler._run_entity_vector_sync(7)
+
+    assert read_cache.invalidated_project_ids == [PROJECT_EXTERNAL_ID]
 
 
 def _clear_project_index_scheduler_state() -> None:
@@ -359,6 +388,8 @@ async def test_drain_background_tasks_awaits_scheduled_work():
 
     scheduler = LocalEntityVectorSyncScheduler(
         search_service=search_service,
+        project_external_id=PROJECT_EXTERNAL_ID,
+        read_cache=None,
         test_mode=False,
     )
     scheduler.schedule_entity_vector_sync(entity_id=7, project_id=13)
diff --git a/tests/mcp/clients/test_clients.py b/tests/mcp/clients/test_clients.py
index e126b1b96..f44651918 100644
--- a/tests/mcp/clients/test_clients.py
+++ b/tests/mcp/clients/test_clients.py
@@ -241,12 +241,12 @@ async def test_search(self, monkeypatch):
             "page_size": 10,
         }
 
-        async def mock_call_post(client, url, **kwargs):
+        async def mock_call_query(client, url, **kwargs):
             assert "/v2/projects/proj-123/search/" in url
             assert kwargs.get("params") == {"page": 1, "page_size": 10}
             return mock_response
 
-        monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", mock_call_post)
+        monkeypatch.setattr("basic_memory.mcp.tools.utils.call_query", mock_call_query)
 
         mock_http = MagicMock()
         client = SearchClient(mock_http, "proj-123")
@@ -276,10 +276,10 @@ async def test_search_infers_unknown_total_for_legacy_semantic_response(
             "has_more": False,
         }
 
-        async def mock_call_post(client, url, **kwargs):
+        async def mock_call_query(client, url, **kwargs):
             return mock_response
 
-        monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", mock_call_post)
+        monkeypatch.setattr("basic_memory.mcp.tools.utils.call_query", mock_call_query)
 
         client = SearchClient(MagicMock(), "proj-123")
         result = await client.search({"text": "query", "retrieval_mode": retrieval_mode})
@@ -301,10 +301,10 @@ async def test_search_preserves_explicit_total_exactness(self, monkeypatch):
             "has_more": False,
         }
 
-        async def mock_call_post(client, url, **kwargs):
+        async def mock_call_query(client, url, **kwargs):
             return mock_response
 
-        monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", mock_call_post)
+        monkeypatch.setattr("basic_memory.mcp.tools.utils.call_query", mock_call_query)
 
         client = SearchClient(MagicMock(), "proj-123")
         result = await client.search(
diff --git a/tests/mcp/test_client_telemetry.py b/tests/mcp/test_client_telemetry.py
index df6a90ddd..53443ab82 100644
--- a/tests/mcp/test_client_telemetry.py
+++ b/tests/mcp/test_client_telemetry.py
@@ -74,7 +74,7 @@ async def test_search_client_emits_client_and_http_spans(monkeypatch) -> None:
     monkeypatch.setattr(logfire, "span", fake_span)
 
     async def handler(request: httpx.Request) -> httpx.Response:
-        assert request.method == "POST"
+        assert request.method == "QUERY"
         return httpx.Response(
             200,
             json={
@@ -96,7 +96,7 @@ async def handler(request: httpx.Request) -> httpx.Response:
         "mcp.http.request",
     ]
     assert spans[1][1] == {
-        "method": "POST",
+        "method": "QUERY",
         "client_name": "search",
         "operation": "search",
         "path_template": "/v2/projects/{project_id}/search/",
diff --git a/tests/mcp/test_mcp_container.py b/tests/mcp/test_mcp_container.py
index 334108666..f6e3f9a66 100644
--- a/tests/mcp/test_mcp_container.py
+++ b/tests/mcp/test_mcp_container.py
@@ -18,6 +18,7 @@ def test_create_from_config(self, app_config):
         container = McpContainer(config=app_config, mode=RuntimeMode.LOCAL)
         assert container.config == app_config
         assert container.mode == RuntimeMode.LOCAL
+        assert container.read_cache is None
 
     def test_should_watch_files_when_enabled_local_mode(self, app_config):
         """Local watching should be enabled in local mode when config says so."""
diff --git a/tests/mcp/test_server_lifespan_branches.py b/tests/mcp/test_server_lifespan_branches.py
index 26aabc1b1..95dfb2c09 100644
--- a/tests/mcp/test_server_lifespan_branches.py
+++ b/tests/mcp/test_server_lifespan_branches.py
@@ -1,11 +1,25 @@
 import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 
 import pytest
 
 import basic_memory.index.note_content_materialization as note_content_materialization
 import basic_memory.mcp.server as server_module
 from basic_memory import db
+from basic_memory.api.container import resolve_container
+from basic_memory.mcp.container import get_container
 from basic_memory.mcp.server import lifespan, mcp
+from basic_memory.read_cache import ReadCacheInvalidationStatus, ReadCacheUnavailable
+
+
+class UnavailableStartupCache:
+    def __init__(self) -> None:
+        self.project_ids: list[str] = []
+
+    async def invalidate_project(self, project_id: str) -> ReadCacheInvalidationStatus:
+        self.project_ids.append(project_id)
+        raise ReadCacheUnavailable("Redis unavailable during startup")
 
 
 @pytest.mark.asyncio
@@ -35,6 +49,30 @@ async def test_mcp_lifespan_shuts_down_db_when_engine_was_none(config_manager):
         pass
 
 
+@pytest.mark.asyncio
+async def test_mcp_lifespan_disables_cache_when_startup_invalidation_is_unavailable(
+    config_manager,
+    monkeypatch,
+) -> None:
+    cache = UnavailableStartupCache()
+
+    @asynccontextmanager
+    async def open_unavailable_cache(
+        redis_url: str | None,
+        *,
+        max_connections: int,
+    ) -> AsyncIterator[UnavailableStartupCache]:
+        del redis_url, max_connections
+        yield cache
+
+    monkeypatch.setattr(server_module, "open_redis_read_cache", open_unavailable_cache)
+
+    async with lifespan(mcp):
+        assert cache.project_ids
+        assert get_container().read_cache is None
+        assert resolve_container().read_cache is None
+
+
 @pytest.mark.asyncio
 async def test_mcp_lifespan_drains_pending_work_before_db_shutdown(config_manager, monkeypatch):
     """Shutdown must drain queued materializations and background tasks before the
diff --git a/tests/mcp/test_tool_basic_memory_diagnostics.py b/tests/mcp/test_tool_basic_memory_diagnostics.py
index cc390f2cb..31170c989 100644
--- a/tests/mcp/test_tool_basic_memory_diagnostics.py
+++ b/tests/mcp/test_tool_basic_memory_diagnostics.py
@@ -100,6 +100,33 @@ def test_redact_config_scrubs_milvus_uri_credentials():
     assert result["milvus_uri"] == ("https://***@milvus.example.com?token=%2A%2A%2A&timeout=30")
 
 
+def test_redact_config_scrubs_redis_url_credentials():
+    raw = {
+        "redis_url": (
+            "redis://cache-user:cache-password@redis.internal:6379/0"
+            "?password=query-secret&timeout=30"
+        ),
+    }
+
+    result = _redact_config(raw)
+
+    assert result["redis_url"] == (
+        "redis://***@redis.internal:6379/0?password=%2A%2A%2A&timeout=30"
+    )
+
+
+def test_redact_config_scrubs_schemeless_redis_url_credentials():
+    raw = {
+        "redis_url": (
+            "cache-user:cache-password@redis.internal:6379/0?password=query-secret&timeout=30"
+        ),
+    }
+
+    result = _redact_config(raw)
+
+    assert result["redis_url"] == ("***@redis.internal:6379/0?password=%2A%2A%2A&timeout=30")
+
+
 def test_redact_config_passes_through_safe_fields():
     raw = {"default_project": "main", "log_level": "INFO", "env": "dev"}
     result = _redact_config(raw)
@@ -255,6 +282,43 @@ def test_diagnostics_redacts_milvus_credentials(tmp_path):
     )
 
 
+def test_diagnostics_redacts_redis_credentials(tmp_path):
+    config_data = {
+        "redis_url": (
+            "redis://cache-user:cache-password@redis.internal:6379/0"
+            "?password=query-secret&timeout=30"
+        ),
+        "projects": {},
+    }
+    config_file = tmp_path / "config.json"
+    config_file.write_text(json.dumps(config_data))
+
+    result = basic_memory_diagnostics()
+
+    for secret in ("cache-user", "cache-password", "query-secret"):
+        assert secret not in result
+    config_json = result.split("```json\n", 1)[1].split("\n```", 1)[0]
+    assert json.loads(config_json)["redis_url"] == (
+        "redis://***@redis.internal:6379/0?password=%2A%2A%2A&timeout=30"
+    )
+
+
+def test_diagnostics_redacts_schemeless_redis_credentials(tmp_path):
+    config_data = {
+        "redis_url": "cache-user:cache-password@redis.internal:6379/0",
+        "projects": {},
+    }
+    config_file = tmp_path / "config.json"
+    config_file.write_text(json.dumps(config_data))
+
+    result = basic_memory_diagnostics()
+
+    assert "cache-user" not in result
+    assert "cache-password" not in result
+    config_json = result.split("```json\n", 1)[1].split("\n```", 1)[0]
+    assert json.loads(config_json)["redis_url"] == "***@redis.internal:6379/0"
+
+
 def test_diagnostics_config_missing(tmp_path):
     """When config file does not exist, output should say so."""
     config_file = tmp_path / "config.json"
diff --git a/tests/mcp/test_tool_utils.py b/tests/mcp/test_tool_utils.py
index d11d240de..da9dfe8d9 100644
--- a/tests/mcp/test_tool_utils.py
+++ b/tests/mcp/test_tool_utils.py
@@ -12,6 +12,7 @@
     call_get,
     call_patch,
     call_post,
+    call_query,
     call_put,
     get_error_message,
 )
@@ -60,6 +61,11 @@ async def post(self, *args, **kwargs):
         self.calls.append(("post", args, kwargs))
         return self._responses["post"]
 
+    async def request(self, method, *args, **kwargs):
+        normalized_method = method.lower()
+        self.calls.append((normalized_method, args, kwargs))
+        return self._responses[normalized_method]
+
     async def put(self, *args, **kwargs):
         self.calls.append(("put", args, kwargs))
         return self._responses["put"]
@@ -120,6 +126,32 @@ async def test_call_post_error(mock_response):
     assert "Internal server error" in str(exc.value)
 
 
+@pytest.mark.asyncio
+async def test_call_query_sends_json_with_query_method(mock_response):
+    """QUERY carries the validated search document as request content."""
+    client = _Client()
+    client.set_response("query", mock_response())
+
+    query = {"text": "cache semantics"}
+    response = await call_query(_client(client), "http://test.com", json=query)
+
+    assert response.status_code == 200
+    method, _args, kwargs = client.calls[0]
+    assert method == "query"
+    assert kwargs["json"] == query
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status_code", [400, 500])
+async def test_call_query_error(mock_response, status_code):
+    """QUERY uses the same friendly client and server errors as other helpers."""
+    client = _Client()
+    client.set_response("query", mock_response(status_code))
+
+    with pytest.raises(ToolError):
+        await call_query(_client(client), "http://test.com", json={"text": "query"})
+
+
 @pytest.mark.asyncio
 async def test_call_put_success(mock_response):
     """Test successful PUT request."""
@@ -200,6 +232,7 @@ async def test_call_post_adds_workspace_permalink_headers_at_request_time(mock_r
 _ALL_CALL_HELPERS = [
     (call_get, "GET"),
     (call_post, "POST"),
+    (call_query, "QUERY"),
     (call_put, "PUT"),
     (call_patch, "PATCH"),
     (call_delete, "DELETE"),
diff --git a/tests/test_read_cache_lifecycle.py b/tests/test_read_cache_lifecycle.py
new file mode 100644
index 000000000..1c1519b5b
--- /dev/null
+++ b/tests/test_read_cache_lifecycle.py
@@ -0,0 +1,71 @@
+"""Tests for standalone Redis read-cache configuration."""
+
+import pytest
+
+from basic_memory.config import BasicMemoryConfig
+from basic_memory.read_cache.lifecycle import normalize_redis_url, open_redis_read_cache
+
+
+@pytest.mark.parametrize(
+    ("configured", "expected"),
+    [
+        ("redis", "redis://redis"),
+        (" redis://localhost:6379/2 ", "redis://localhost:6379/2"),
+        ("rediss://cache.example.com", "rediss://cache.example.com"),
+    ],
+)
+def test_normalize_redis_url(configured: str, expected: str) -> None:
+    assert normalize_redis_url(configured) == expected
+
+
+def test_redis_url_uses_basic_memory_env_prefix(monkeypatch) -> None:
+    monkeypatch.setenv("BASIC_MEMORY_REDIS_URL", "redis")
+
+    assert BasicMemoryConfig().redis_url == "redis"
+
+
+def test_redis_max_connections_uses_basic_memory_env_prefix(monkeypatch) -> None:
+    monkeypatch.setenv("BASIC_MEMORY_REDIS_MAX_CONNECTIONS", "64")
+
+    assert BasicMemoryConfig().redis_max_connections == 64
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("configured", [None, " "])
+async def test_open_redis_read_cache_is_disabled_without_url(configured: str | None) -> None:
+    async with open_redis_read_cache(configured) as read_cache:
+        assert read_cache is None
+
+
+@pytest.mark.asyncio
+async def test_open_redis_read_cache_configures_pool_size(monkeypatch) -> None:
+    from basic_memory.read_cache import redis as redis_adapter
+
+    configured: dict[str, object] = {}
+
+    class Client:
+        closed = False
+
+        async def aclose(self) -> None:
+            self.closed = True
+
+    client = Client()
+
+    def create_client(url: str, *, max_connections: int):
+        configured.update(url=url, max_connections=max_connections)
+        return client
+
+    monkeypatch.setattr(redis_adapter, "create_redis_read_cache_client", create_client)
+
+    async with open_redis_read_cache("redis", max_connections=64) as read_cache:
+        assert read_cache is not None
+
+    assert configured == {"url": "redis://redis", "max_connections": 64}
+    assert client.closed
+
+
+@pytest.mark.asyncio
+async def test_open_redis_read_cache_rejects_non_positive_pool_size() -> None:
+    with pytest.raises(ValueError, match="max_connections must be positive"):
+        async with open_redis_read_cache(None, max_connections=0):
+            pass
diff --git a/tests/test_read_load_bench.py b/tests/test_read_load_bench.py
new file mode 100644
index 000000000..d5aedd2a5
--- /dev/null
+++ b/tests/test_read_load_bench.py
@@ -0,0 +1,469 @@
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import json
+import sqlite3
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from types import ModuleType
+
+import pytest
+
+
+def load_read_load_bench() -> ModuleType:
+    script_path = Path(__file__).parents[1] / "benchmarks" / "scripts" / "read_load_bench.py"
+    spec = importlib.util.spec_from_file_location("read_load_bench", script_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"could not load benchmark script: {script_path}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+read_load_bench = load_read_load_bench()
+
+
+@dataclass
+class RecordingScalarResult:
+    version: str
+
+    def scalar_one(self) -> str:
+        return self.version
+
+
+@dataclass
+class RecordingConnection:
+    version: str
+    query: str | None = None
+
+    async def __aenter__(self) -> RecordingConnection:
+        return self
+
+    async def __aexit__(self, *args: object) -> None:
+        return None
+
+    async def exec_driver_sql(self, query: str) -> RecordingScalarResult:
+        self.query = query
+        return RecordingScalarResult(self.version)
+
+
+@dataclass
+class RecordingEngine:
+    connection: RecordingConnection
+    disposed: bool = False
+
+    def connect(self) -> RecordingConnection:
+        return self.connection
+
+    async def dispose(self) -> None:
+        self.disposed = True
+
+
+def test_benchmark_manifest_stays_beside_custom_output(tmp_path: Path) -> None:
+    scratch = tmp_path / "scratch"
+    output_path = tmp_path / "published" / "results.jsonl"
+
+    assert (
+        read_load_bench.benchmark_manifest_path(
+            scratch=scratch,
+            output_path=output_path,
+        )
+        == output_path.parent / "manifest.json"
+    )
+    assert (
+        read_load_bench.benchmark_manifest_path(
+            scratch=scratch,
+            output_path=None,
+        )
+        == scratch / "manifest.json"
+    )
+
+
+@pytest.mark.asyncio
+async def test_database_server_version_reports_sqlite_library() -> None:
+    assert (
+        await read_load_bench.database_server_version(backend="sqlite", database_url=None)
+        == sqlite3.sqlite_version
+    )
+
+
+@pytest.mark.asyncio
+async def test_database_server_version_queries_postgres_and_closes_engine(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    connection = RecordingConnection(version="16.9")
+    engine = RecordingEngine(connection=connection)
+    observed_url: str | None = None
+
+    def create_async_engine(database_url: str) -> RecordingEngine:
+        nonlocal observed_url
+        observed_url = database_url
+        return engine
+
+    monkeypatch.setattr(read_load_bench, "create_async_engine", create_async_engine)
+
+    version = await read_load_bench.database_server_version(
+        backend="postgres",
+        database_url="postgresql+asyncpg://benchmark.test/readload",
+    )
+
+    assert version == "16.9"
+    assert observed_url == "postgresql+asyncpg://benchmark.test/readload"
+    assert connection.query == "SHOW server_version"
+    assert engine.disposed
+
+
+@pytest.mark.asyncio
+async def test_database_server_version_requires_postgres_url() -> None:
+    with pytest.raises(ValueError, match="Postgres provenance requires a database URL"):
+        await read_load_bench.database_server_version(backend="postgres", database_url=None)
+
+
+def test_basic_memory_runtime_uses_target_python_environment(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    command_dir = tmp_path / "benchmark-checkout" / ".venv" / "bin"
+    command_dir.mkdir(parents=True)
+    command_path = command_dir / "basic-memory"
+    command_path.touch()
+    python_path = command_dir / "python"
+    python_path.touch()
+
+    module_path = tmp_path / "basic-memory-source" / "src" / "basic_memory" / "__init__.py"
+    module_path.parent.mkdir(parents=True)
+    module_path.touch()
+    observed_commands: list[list[str]] = []
+
+    def checked_output(command: list[str], *, env: dict[str, str] | None = None) -> str:
+        observed_commands.append(command)
+        assert env == {"BENCHMARK_TEST": "target"}
+        return json.dumps(
+            {
+                "module_path": str(module_path),
+                "python_version": "3.12.11",
+            }
+        )
+
+    monkeypatch.setattr(read_load_bench, "checked_output", checked_output)
+
+    runtime = read_load_bench.basic_memory_runtime(
+        command_path,
+        {"BENCHMARK_TEST": "target"},
+    )
+
+    assert runtime.module_path == module_path
+    assert runtime.python_version == "3.12.11"
+    assert observed_commands[0][0] == str(python_path)
+
+
+@pytest.mark.parametrize("runtime_output", ["not json", "{}", "[]"])
+def test_basic_memory_runtime_rejects_invalid_target_metadata(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    runtime_output: str,
+) -> None:
+    command_dir = tmp_path / "benchmark-checkout" / ".venv" / "bin"
+    command_dir.mkdir(parents=True)
+    command_path = command_dir / "basic-memory"
+    command_path.touch()
+    (command_dir / "python").touch()
+    monkeypatch.setattr(read_load_bench, "checked_output", lambda *args, **kwargs: runtime_output)
+
+    with pytest.raises((RuntimeError, TypeError), match="runtime provenance"):
+        read_load_bench.basic_memory_runtime(command_path, {})
+
+
+def test_isolated_env_removes_all_inherited_basic_memory_settings(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    inherited_keys = (
+        "BASIC_MEMORY_PROJECT_ROOT",
+        "BASIC_MEMORY_CLOUD_API_KEY",
+        "BASIC_MEMORY_DATABASE_POOL_SIZE",
+    )
+    for key in inherited_keys:
+        monkeypatch.setenv(key, "must-not-escape")
+    monkeypatch.setenv("BASIC_MEMORY_AUTO_UPDATE", "true")
+    monkeypatch.setenv("BENCHMARK_TEST", "preserved")
+
+    env = read_load_bench.isolated_env(
+        tmp_path / "config",
+        redis_url="redis",
+        redis_max_connections=64,
+    )
+
+    assert env["BENCHMARK_TEST"] == "preserved"
+    assert env["BASIC_MEMORY_AUTO_UPDATE"] == "false"
+    assert env["BASIC_MEMORY_REDIS_MAX_CONNECTIONS"] == "64"
+    assert all(key not in env for key in inherited_keys)
+    assert {key for key in env if key.startswith("BASIC_MEMORY_")} == {
+        "BASIC_MEMORY_AUTO_UPDATE",
+        "BASIC_MEMORY_CONFIG_DIR",
+        "BASIC_MEMORY_LOG_LEVEL",
+        "BASIC_MEMORY_REDIS_MAX_CONNECTIONS",
+        "BASIC_MEMORY_REDIS_URL",
+        "BASIC_MEMORY_SEMANTIC_SEARCH_ENABLED",
+    }
+
+
+def benchmark_args(scratch: Path, **overrides: object) -> argparse.Namespace:
+    values: dict[str, object] = {
+        "backend": "postgres",
+        "bm_command": "basic-memory",
+        "concurrency": "1",
+        "database_url": None,
+        "label": "test",
+        "notes_per_size": 1,
+        "output": None,
+        "quiesce_seconds": 0.0,
+        "reads": 1,
+        "ready_timeout": 1.0,
+        "redis_url": "redis://unreachable.test",
+        "scratch": str(scratch),
+        "seed": 1021,
+        "seed_concurrency": 1,
+        "sizes": "1024",
+        "truncate": False,
+    }
+    values.update(overrides)
+    return argparse.Namespace(**values)
+
+
+@dataclass
+class RecordingPostgresContainer:
+    stopped: bool = False
+
+    def stop(self) -> None:
+        self.stopped = True
+
+    def get_connection_url(
+        self,
+        host: str | None = None,
+        driver: str | None = None,
+    ) -> str:
+        return "postgresql://benchmark:benchmark@localhost:5432/benchmark"
+
+
+@dataclass
+class FailingStartPostgresContainer(RecordingPostgresContainer):
+    def start(self) -> object:
+        raise RuntimeError("Postgres readiness failed")
+
+
+def test_start_postgres_stops_container_when_startup_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from testcontainers import postgres as postgres_module
+
+    container = FailingStartPostgresContainer()
+    monkeypatch.setattr(postgres_module, "PostgresContainer", lambda image: container)
+
+    with pytest.raises(RuntimeError, match="Postgres readiness failed"):
+        read_load_bench.start_postgres()
+
+    assert container.stopped
+
+
+@pytest.mark.asyncio
+async def test_run_stops_postgres_when_setup_fails(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    container = RecordingPostgresContainer()
+    monkeypatch.setattr(read_load_bench, "start_postgres", lambda: container)
+
+    async def fail_redis_provenance(redis_url: str) -> str:
+        raise RuntimeError(f"Redis unavailable: {redis_url}")
+
+    async def database_provenance(*, backend: str, database_url: str | None) -> str:
+        assert backend == "postgres"
+        assert database_url is not None
+        return "16.9"
+
+    monkeypatch.setattr(read_load_bench, "database_server_version", database_provenance)
+    monkeypatch.setattr(read_load_bench, "redis_server_version", fail_redis_provenance)
+
+    with pytest.raises(RuntimeError, match="Redis unavailable"):
+        await read_load_bench.run(benchmark_args(tmp_path / "run"))
+
+    assert container.stopped
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("option", ["notes_per_size", "reads", "seed_concurrency"])
+async def test_run_rejects_non_positive_counts_before_starting_postgres(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    option: str,
+) -> None:
+    postgres_started = False
+
+    def start_postgres() -> RecordingPostgresContainer:
+        nonlocal postgres_started
+        postgres_started = True
+        return RecordingPostgresContainer()
+
+    monkeypatch.setattr(read_load_bench, "start_postgres", start_postgres)
+
+    with pytest.raises(ValueError, match=f"--{option.replace('_', '-')}"):
+        await read_load_bench.run(benchmark_args(tmp_path / option, **{option: 0}))
+
+    assert not postgres_started
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_duplicate_sizes_before_starting_postgres(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    postgres_started = False
+
+    def start_postgres() -> RecordingPostgresContainer:
+        nonlocal postgres_started
+        postgres_started = True
+        return RecordingPostgresContainer()
+
+    monkeypatch.setattr(read_load_bench, "start_postgres", start_postgres)
+
+    with pytest.raises(ValueError, match="--sizes must not contain duplicates"):
+        await read_load_bench.run(benchmark_args(tmp_path / "duplicate-sizes", sizes="1024,1024"))
+
+    assert not postgres_started
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_duplicate_concurrency_before_starting_postgres(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    postgres_started = False
+
+    def start_postgres() -> RecordingPostgresContainer:
+        nonlocal postgres_started
+        postgres_started = True
+        return RecordingPostgresContainer()
+
+    monkeypatch.setattr(read_load_bench, "start_postgres", start_postgres)
+
+    with pytest.raises(ValueError, match="--concurrency must not contain duplicates"):
+        await read_load_bench.run(
+            benchmark_args(tmp_path / "duplicate-concurrency", concurrency="1,8,8")
+        )
+
+    assert not postgres_started
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("runtime_directory", ["config", "project", "main-home"])
+async def test_run_rejects_output_inside_runtime_directories_before_cleanup(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    runtime_directory: str,
+) -> None:
+    postgres_started = False
+    scratch = tmp_path / "runtime-output"
+    output_dir = scratch / runtime_directory
+    output_dir.mkdir(parents=True)
+    sentinel = output_dir / "keep.txt"
+    sentinel.write_text("preserved\n", encoding="utf-8")
+
+    def start_postgres() -> RecordingPostgresContainer:
+        nonlocal postgres_started
+        postgres_started = True
+        return RecordingPostgresContainer()
+
+    monkeypatch.setattr(read_load_bench, "start_postgres", start_postgres)
+
+    with pytest.raises(ValueError, match="outside benchmark runtime directories"):
+        await read_load_bench.run(benchmark_args(scratch, output=str(output_dir / "results.jsonl")))
+
+    assert sentinel.read_text(encoding="utf-8") == "preserved\n"
+    assert not postgres_started
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_manifest_as_output_before_starting_postgres(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    postgres_started = False
+
+    def start_postgres() -> RecordingPostgresContainer:
+        nonlocal postgres_started
+        postgres_started = True
+        return RecordingPostgresContainer()
+
+    monkeypatch.setattr(read_load_bench, "start_postgres", start_postgres)
+
+    with pytest.raises(ValueError, match="must not be manifest.json"):
+        await read_load_bench.run(
+            benchmark_args(
+                tmp_path / "manifest-output",
+                output=str(tmp_path / "published" / "manifest.json"),
+            )
+        )
+
+    assert not postgres_started
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_existing_output_without_truncate_before_starting_postgres(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    postgres_started = False
+    output_path = tmp_path / "published" / "results.jsonl"
+    output_path.parent.mkdir(parents=True)
+    output_path.write_text("existing run\n", encoding="utf-8")
+
+    def start_postgres() -> RecordingPostgresContainer:
+        nonlocal postgres_started
+        postgres_started = True
+        return RecordingPostgresContainer()
+
+    monkeypatch.setattr(read_load_bench, "start_postgres", start_postgres)
+
+    with pytest.raises(ValueError, match="--output already exists"):
+        await read_load_bench.run(
+            benchmark_args(tmp_path / "existing-output", output=str(output_path))
+        )
+
+    assert output_path.read_text(encoding="utf-8") == "existing run\n"
+    assert not postgres_started
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_second_output_in_manifest_directory_before_starting_postgres(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    postgres_started = False
+    output_dir = tmp_path / "published"
+    output_dir.mkdir(parents=True)
+    first_output = output_dir / "first.jsonl"
+    first_output.write_text("first run\n", encoding="utf-8")
+    manifest_path = output_dir / "manifest.json"
+    manifest_path.write_text('{"run_id": "first"}\n', encoding="utf-8")
+    second_output = output_dir / "second.jsonl"
+
+    def start_postgres() -> RecordingPostgresContainer:
+        nonlocal postgres_started
+        postgres_started = True
+        return RecordingPostgresContainer()
+
+    monkeypatch.setattr(read_load_bench, "start_postgres", start_postgres)
+
+    with pytest.raises(ValueError, match="manifest.json for another run"):
+        await read_load_bench.run(
+            benchmark_args(tmp_path / "second-output", output=str(second_output))
+        )
+
+    assert first_output.read_text(encoding="utf-8") == "first run\n"
+    assert json.loads(manifest_path.read_text(encoding="utf-8"))["run_id"] == "first"
+    assert not second_output.exists()
+    assert not postgres_started