feat: split runtime and migration stores

docs/plans/2026-07-10-writerd-phase1-store-split.md

@@ -0,0 +1,94 @@
+# Writerd Phase 1 Runtime/Migration Store Split Implementation Plan
+
+> **For Codex:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** Make every indexed file use one cheap, schema-validated runtime writer connection while keeping all schema, FTS reconciliation, repair, and migration work in an explicit copy-only path.
+
+**Architecture:** Keep `VectorStore` as the legacy migration-capable implementation for compatibility and rollback. Add three explicit entrypoints: `ReadonlyStore` for immutable readers, `WriterRuntimeStore` for an existing schema-compatible database with no open-time DDL or corpus scans, and `OfflineMigrator` for explicit schema/repair work on non-canonical copies. A runtime factory defaults to `WriterRuntimeStore` and honors `BRAINLAYER_RUNTIME_STORE=legacy`. Index opens the factory once around its file loop and passes the connection through the adapter. A deterministic schema contract fingerprint is computed from cheap `sqlite_schema` and `PRAGMA table_info` reads and is attached to Phase 0 `runtime_open` telemetry.
+
+**Tech Stack:** Python 3.11+, APSW/sqlite-vec, Typer, pytest, Ruff, Phase 0 writer telemetry.
+
+---
+
+### Task 1: Prove the three-store contract and fail-closed fingerprint
+
+**Files:**
+- Create: `src/brainlayer/runtime_store.py`
+- Create: `tests/test_runtime_store.py`
+
+1. Write failing tests that bootstrap only `tmp_path` databases with legacy `VectorStore`, then require `ReadonlyStore`, `WriterRuntimeStore`, `OfflineMigrator`, and the schema fingerprint API.
+2. Assert runtime open executes only an allowlist of connection/schema-probe statements: no CREATE/DROP/ALTER, INSERT/UPDATE/DELETE, FTS row traversal, COUNT, repair, reconciliation, or optimize.
+3. Assert runtime open fails closed for a missing database and for a removed required trigger/column, releases the writer pidfile, and reports a stable expected/actual fingerprint diagnostic.
+4. Assert `ReadonlyStore` opens `SQLITE_OPEN_READONLY` and never calls legacy initialization.
+5. Implement the minimal shared instance-state setup, vec extension loading, capability flags, deterministic schema contract, and typed `SchemaFingerprintMismatch` error.
+6. Run `pytest -q tests/test_runtime_store.py` until green.
+
+### Task 2: Instrument runtime open and preserve rollback semantics
+
+**Files:**
+- Modify: `src/brainlayer/vector_store.py`
+- Modify: `src/brainlayer/writer_telemetry.py` only if the existing metadata API is insufficient
+- Modify: `tests/test_writer_telemetry_store_paths.py`
+- Modify: `tests/test_runtime_store.py`
+
+1. Add failing tests requiring one `runtime_open` writer-operation span with the schema fingerprint, completed outcome, duration, and zero corpus-scan/fullscan statements.
+2. Ensure APSW's automatic best-practice connection hook cannot run `PRAGMA optimize` or persistent WAL mutation before runtime telemetry begins; retain explicit legacy initialization behavior.
+3. Implement `open_writer_store()`: default new runtime path; exact rollback value `BRAINLAYER_RUNTIME_STORE=legacy`; reject unknown flag values fail closed.
+4. Assert the legacy flag returns the old constructor and the default path never calls `_init_db_with_retry`.
+5. Run focused runtime and telemetry tests until green.
+
+### Task 3: Make migration and repair explicit and copy-only
+
+**Files:**
+- Modify: `src/brainlayer/cli/__init__.py`
+- Modify: `src/brainlayer/cli_new.py`
+- Modify: `src/brainlayer/mcp/_shared.py`
+- Modify: `tests/test_cli_direct_sqlite.py`
+- Modify: `tests/test_vector_store_readonly.py`
+- Modify: `tests/test_runtime_store.py`
+
+1. Write failing tests proving `OfflineMigrator` refuses the resolved canonical path unless the gated atomic-swap override is explicitly set.
+2. Add an explicit copy-path schema migration command and route `repair-fts` through the same offline-only guard; never silently default either operation to canonical.
+3. Replace readonly search pool/CLI constructors with `ReadonlyStore` and remove opportunistic search-time schema bootstrap. Missing/stale schemas must fail closed with an instruction to run the explicit copy migration/swap flow.
+4. Route the cached MCP direct writer through `open_writer_store`; treat a schema fingerprint mismatch as a durable-queue reason so producers can still enqueue.
+5. Preserve `VectorStore` only for legacy rollback and test/offline compatibility.
+6. Run the focused CLI, readonly, MCP-store, and runtime tests until green.
+
+### Task 4: Reuse one runtime writer connection for the complete index run
+
+**Files:**
+- Modify: `src/brainlayer/index_new.py`
+- Modify: `src/brainlayer/cli/__init__.py`
+- Modify: `tests/test_cli_index_watchdog.py`
+- Modify: `tests/test_context_pipeline.py` only where adapter compatibility requires it
+
+1. Write failing tests with two JSONL files that count runtime-store construction and assert exactly one open/close for the whole run.
+2. Let `index_chunks_to_sqlite` accept an injected writer store while preserving standalone factory-open behavior.
+3. Open one `open_writer_store()` around the CLI file loop and pass it to each adapter call.
+4. Keep the watchdog's single run deadline, keyed APSW progress handler, rollback telemetry, committed-chunk accounting, and nonzero alarm exit unchanged under both new and legacy modes.
+5. Run `pytest -q tests/test_cli_index_watchdog.py tests/test_vector_store_upsert_transactions.py tests/test_writer_telemetry_store_paths.py` until green.
+
+### Task 5: Prove the production-size gate on a copy
+
+**Files:**
+- Create: `scripts/benchmark_runtime_store_open.py`
+- Create: `tests/test_runtime_store_benchmark.py`
+
+1. Add a safe benchmark CLI that refuses the canonical path, opens the supplied copy repeatedly, captures per-open duration and runtime-open telemetry, and reports p50/p95/p99/max plus scan/DDL/DML statement violations as JSON.
+2. Unit-test canonical refusal and percentile/statement classification using `tmp_path` only.
+3. Reuse an existing 17 GB copy if present; otherwise create a fresh `VACUUM INTO` snapshot from the canonical database using a read-only source connection and a distinct destination.
+4. Run the benchmark only on that copy. Gate on p99 <100 ms and zero corpus-scan/DDL/DML statements. Measure the legacy constructor separately as the before receipt, with a bounded watchdog so a recurrence cannot wedge the worker.
+5. Preserve the copy for reproducibility; never run the benchmark constructor against the live canonical database.
+
+### Task 6: Verify, publish one PR, and hand off
+
+**Files:**
+- Create: `docs.local/handoffs/W3-PHASE1-REPORT.md`
+
+1. Run focused tests, the handoff-required telemetry/watchdog/transaction suites, `ruff check src/ tests/`, `ruff format --check src/ tests/`, and `git diff --check`.
+2. Review the complete diff and run bounded local CodeRabbit. Address only evidence-backed findings.
+3. Commit intentional files, push with `BRAINLAYER_PREPUSH_SCOPE=changed-only`, and open one ready PR whose description includes incident `brainbar-c07ade22-ecc`, rollback instructions, copy-only safety, and gate measurements.
+4. Post exactly one PR-open line to `~/Gits/orchestrator/docs.local/collab/driver-buddy-2026-07-07.md` under `## brainlayerLead-w3`.
+5. Request Codex, Cursor, Bugbot, and CodeRabbit reviews, inspect CI/review feedback, and leave merging to the lead.
+6. Write the report with branch, PR URL, changed files, commands/results, open p99 before/after, zero-scan telemetry proof, index single-open proof, rollback receipt, and exact final line `DONE_W3_PHASE1_STORE_SPLIT`.
+7. Store the Phase 1 WHAT + WHY milestone in BrainLayer and verify it can be recalled.

scripts/benchmark_runtime_store_open.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+"""Measure legacy and Phase 1 writer-open latency on an explicit DB copy."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import os
+import subprocess
+import sys
+import time
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any, Iterator
+
+from brainlayer.runtime_store import WriterRuntimeStore
+
+_TELEMETRY_ENV = {
+    "BRAINLAYER_WRITER_TELEMETRY": "1",
+    "BRAINLAYER_WRITER_TELEMETRY_FTS_SAMPLE_TTL_SECONDS": "0",
+}
+_MUTATION_TOKENS = ("CREATE ", "DROP ", "ALTER ", "INSERT ", "UPDATE ", "DELETE ", "OPTIMIZE")
+_CORPUS_TABLE_TOKENS = ("FROM CHUNKS ", "FROM CHUNKS_FTS", "FROM CHUNK_FTS_ROWIDS")
+
+
+def _canonical_db_path() -> Path:
+    from brainlayer.paths import get_db_path
+
+    return get_db_path()
+
+
+def _assert_copy_path(db_path: Path) -> Path:
+    resolved = db_path.expanduser().resolve()
+    if resolved == _canonical_db_path().expanduser().resolve():
+        raise PermissionError("runtime-open benchmark refuses the canonical BrainLayer database")
+    return resolved
+
+
+@contextmanager
+def _temporary_environment(updates: dict[str, str]) -> Iterator[None]:
+    previous = {name: os.environ.get(name) for name in updates}
+    os.environ.update(updates)
+    try:
+        yield
+    finally:
+        for name, value in previous.items():
+            if value is None:
+                os.environ.pop(name, None)
+            else:
+                os.environ[name] = value
+
+
+def _percentile(values: list[float], percentile: float) -> float:
+    ordered = sorted(values)
+    index = max(0, math.ceil(len(ordered) * percentile) - 1)
+    return ordered[index]
+
+
+def _timing_summary(durations_ms: list[float]) -> dict[str, float]:
+    return {
+        "p50_ms": round(_percentile(durations_ms, 0.50), 3),
+        "p95_ms": round(_percentile(durations_ms, 0.95), 3),
+        "p99_ms": round(_percentile(durations_ms, 0.99), 3),
+        "max_ms": round(max(durations_ms), 3),
+    }
+
+
+def _runtime_finished_events(telemetry_path: Path) -> list[dict[str, Any]]:
+    if not telemetry_path.exists():
+        return []
+    events: list[dict[str, Any]] = []
+    for line in telemetry_path.read_text(encoding="utf-8").splitlines():
+        try:
+            event = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if event.get("event") == "txn_finished" and event.get("operation") == "runtime_open":
+            events.append(event)
+    return events
+
+
+def benchmark_runtime_open(
+    db_path: Path,
+    *,
+    iterations: int,
+    telemetry_path: Path,
+) -> dict[str, Any]:
+    """Measure runtime opens and prove telemetry contains no corpus work."""
+    if iterations <= 0:
+        raise ValueError("iterations must be positive")
+    path = _assert_copy_path(Path(db_path))
+    telemetry_path = Path(telemetry_path).expanduser().resolve()
+    if telemetry_path in {path, _canonical_db_path().expanduser().resolve()}:
+        raise PermissionError("telemetry path must differ from both the benchmark copy and canonical database")
+    telemetry_path.parent.mkdir(parents=True, exist_ok=True)
+    telemetry_path.unlink(missing_ok=True)
+
+    durations_ms: list[float] = []
+    environment = {
+        **_TELEMETRY_ENV,
+        "BRAINLAYER_WRITER_TELEMETRY_PATH": str(telemetry_path),
+        "BRAINLAYER_WRITER_HEARTBEAT_DIR": str(telemetry_path.parent / "runtime-open-heartbeats"),
+        "BRAINLAYER_WRITER_PIDFILE_DIR": str(telemetry_path.parent / "runtime-open-pidfiles"),
+        "BRAINLAYER_RUNTIME_STORE": "runtime",
+    }
+    with _temporary_environment(environment):
+        for _ in range(iterations):
+            started = time.perf_counter()
+            store = WriterRuntimeStore(path)
+            store.close()
+            durations_ms.append((time.perf_counter() - started) * 1000.0)
+
+    events = _runtime_finished_events(telemetry_path)
+    if len(events) != iterations:
+        raise RuntimeError(f"expected telemetry for {iterations} runtime opens, got {len(events)}")
+    statements = [statement for event in events for statement in event.get("statements", [])]
+    normalized = [str(statement.get("normalized_sql") or "").upper() for statement in statements]
+    statement_violations = sorted(
+        {statement for statement in normalized if any(token in statement for token in _MUTATION_TOKENS)}
+    )
+    corpus_scan_statements = sorted(
+        {statement for statement in normalized if any(token in statement for token in _CORPUS_TABLE_TOKENS)}
+    )
+    fingerprints = sorted(
+        {str(event["schema_fingerprint"]) for event in events if event.get("schema_fingerprint")}
+    )
+    return {
+        "mode": "runtime",
+        "db_path": str(path),
+        "db_bytes": path.stat().st_size,
+        "iterations": iterations,
+        **_timing_summary(durations_ms),
+        "runtime_open_events": len(events),
+        "schema_fingerprints": fingerprints,
+        "statement_violations": statement_violations,
+        "corpus_scan_statements": corpus_scan_statements,
+    }
+
+
+def benchmark_legacy_open(
+    db_path: Path,
+    *,
+    iterations: int,
+    timeout_seconds: float,
+) -> dict[str, Any]:
+    """Measure the rollback constructor in killable subprocesses on a copy."""
+    if iterations <= 0:
+        raise ValueError("iterations must be positive")
+    if timeout_seconds <= 0:
+        raise ValueError("timeout_seconds must be positive")
+    path = _assert_copy_path(Path(db_path))
+    durations_ms: list[float] = []
+    timeouts = 0
+    code = "from pathlib import Path; from brainlayer.vector_store import VectorStore; VectorStore(Path(__import__('sys').argv[1])).close()"
+    environment = {**os.environ, "BRAINLAYER_WRITER_TELEMETRY": "0"}
+    for _ in range(iterations):
+        started = time.perf_counter()
+        try:
+            subprocess.run(
+                [sys.executable, "-c", code, str(path)],
+                check=True,
+                capture_output=True,
+                text=True,
+                timeout=timeout_seconds,
+                env=environment,
+            )
+            durations_ms.append((time.perf_counter() - started) * 1000.0)
+        except subprocess.TimeoutExpired:
+            durations_ms.append(timeout_seconds * 1000.0)
+            timeouts += 1
+            break
+    return {
+        "mode": "legacy",
+        "db_path": str(path),
+        "db_bytes": path.stat().st_size,
+        "iterations_requested": iterations,
+        "iterations_completed": len(durations_ms) - timeouts,
+        "timeouts": timeouts,
+        "timeout_seconds": timeout_seconds,
+        **_timing_summary(durations_ms),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("db_path", type=Path, help="Explicit non-canonical database copy")
+    parser.add_argument("--mode", choices=("runtime", "legacy"), default="runtime")
+    parser.add_argument("--iterations", type=int, default=100)
+    parser.add_argument("--telemetry-path", type=Path, default=Path("runtime-open-benchmark.jsonl"))
+    parser.add_argument("--legacy-timeout-seconds", type=float, default=30.0)
+    args = parser.parse_args()
+
+    if args.mode == "runtime":
+        result = benchmark_runtime_open(
+            args.db_path,
+            iterations=args.iterations,
+            telemetry_path=args.telemetry_path,
+        )
+    else:
+        result = benchmark_legacy_open(
+            args.db_path,
+            iterations=args.iterations,
+            timeout_seconds=args.legacy_timeout_seconds,
+        )
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())