diff --git a/docs/plans/2026-07-10-writerd-phase1-store-split.md b/docs/plans/2026-07-10-writerd-phase1-store-split.md new file mode 100644 index 00000000..87fd7078 --- /dev/null +++ b/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. diff --git a/scripts/benchmark_runtime_store_open.py b/scripts/benchmark_runtime_store_open.py new file mode 100644 index 00000000..fb3b61e7 --- /dev/null +++ b/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()) diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index 9ebee0fd..a171e155 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -2379,11 +2379,29 @@ def serve() -> None: @app.command("migrate") -def migrate() -> None: - """Migrate from ChromaDB to sqlite-vec (one-time).""" +def migrate( + db_path: Path = typer.Argument(..., help="Explicit offline database copy to receive ChromaDB data."), +) -> None: + """Migrate ChromaDB into an explicit offline sqlite-vec copy.""" from ..cli_new import migrate_command - migrate_command() + migrate_command(db_path) + + +@app.command("migrate-store") +def migrate_store( + db_path: Path = typer.Argument(..., help="Explicit offline database copy to initialize or migrate."), +) -> None: + """Apply schema migrations to an offline copy, never canonical production.""" + from ..runtime_store import OfflineMigrator + + try: + store = OfflineMigrator(db_path) + store.close() + except PermissionError as exc: + rprint(f"[bold red]Migration refused:[/] {exc}") + raise typer.Exit(1) from exc + rprint(f"[green]Migrated offline copy:[/] {db_path}") @app.command("consolidate") @@ -2669,16 +2687,21 @@ def wal_checkpoint( @app.command("repair-fts") -def repair_fts() -> None: - """Explicitly rebuild FTS maintenance tables outside normal startup.""" - from ..paths import get_db_path - from ..vector_store import VectorStore +def repair_fts( + db_path: Path = typer.Argument(..., help="Explicit offline database copy to repair."), +) -> None: + """Rebuild FTS maintenance tables on an offline copy only.""" + from ..runtime_store import OfflineMigrator - store = VectorStore(get_db_path()) try: - result = store.repair_fts(rebuild_trigram=True) - finally: - store.close() + store = OfflineMigrator(db_path) + try: + result = store.repair_fts(rebuild_trigram=True) + finally: + store.close() + except PermissionError as exc: + rprint(f"[bold red]Repair refused:[/] {exc}") + raise typer.Exit(1) from exc console.print_json(data=result) @@ -3530,9 +3553,11 @@ def index_fast( ) from ..index_new import index_chunks_to_sqlite + from ..paths import get_db_path from ..pipeline.chunk import chunk_content from ..pipeline.classify import classify_content from ..pipeline.extract import parse_jsonl + from ..runtime_store import ReadonlyStore, open_writer_store from ..vector_store import IndexDeadlineExceeded if not source.exists(): @@ -3565,16 +3590,19 @@ def index_fast( max_runtime_s = _index_max_runtime_s() deadline_monotonic = start_monotonic + max_runtime_s - with Progress( - SpinnerColumn(), - TextColumn("[progress.description]{task.description}"), - BarColumn(), - MofNCompleteColumn(), - TaskProgressColumn(), - TimeElapsedColumn(), - TimeRemainingColumn(), - console=console, - ) as progress: + with ( + open_writer_store(get_db_path()) as runtime_store, + Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + MofNCompleteColumn(), + TaskProgressColumn(), + TimeElapsedColumn(), + TimeRemainingColumn(), + console=console, + ) as progress, + ): task = progress.add_task("Processing files...", total=len(jsonl_files)) for i, jsonl_file in enumerate(jsonl_files): @@ -3605,6 +3633,7 @@ def progress_callback(embedded_count, total_embed): project=proj_name, on_progress=progress_callback, deadline_monotonic=deadline_monotonic, + store=runtime_store, ) total_chunks += indexed files_completed = i + 1 @@ -3638,9 +3667,7 @@ def progress_callback(embedded_count, total_embed): ) if SUPABASE_URL and SUPABASE_SERVICE_KEY: - from ..vector_store import VectorStore - - store = VectorStore(DEFAULT_DB_PATH) + store = ReadonlyStore(DEFAULT_DB_PATH) _sync_stats_to_supabase(store) store.close() rprint("[dim]Synced enrichment stats to Supabase[/]") diff --git a/src/brainlayer/cli_new.py b/src/brainlayer/cli_new.py index e44abe01..95bf4e00 100644 --- a/src/brainlayer/cli_new.py +++ b/src/brainlayer/cli_new.py @@ -12,16 +12,16 @@ from rich.table import Table from .paths import get_db_path -from .vector_store import VectorStore +from .runtime_store import ReadonlyStore console = Console() @contextmanager -def _readonly_store() -> Iterator[VectorStore]: +def _readonly_store() -> Iterator[ReadonlyStore]: db_path = get_db_path() _ensure_readonly_db_ready(db_path) - store = VectorStore(db_path, readonly=True) + store = ReadonlyStore(db_path) try: yield store finally: @@ -29,14 +29,11 @@ def _readonly_store() -> Iterator[VectorStore]: def _ensure_readonly_db_ready(db_path: Path) -> None: - if _has_readonly_schema(db_path): - return - - bootstrap_store = VectorStore(db_path) - try: - pass - finally: - bootstrap_store.close() + if not _has_readonly_schema(db_path): + raise RuntimeError( + "BrainLayer schema is missing or stale; migrate an offline copy with " + "'brainlayer migrate-store ' and use the gated atomic-swap flow" + ) def _has_readonly_schema(db_path: Path) -> bool: @@ -186,14 +183,15 @@ def stats_command() -> None: raise typer.Exit(1) -def migrate_command() -> None: - """Migrate from ChromaDB to sqlite-vec.""" +def migrate_command(sqlite_path: Path) -> None: + """Migrate from ChromaDB to an explicit offline sqlite-vec copy.""" try: from .migrate import migrate_from_chromadb + from .runtime_store import resolve_offline_database_path rprint("[bold blue]זיכרון[/] - Migration Tool\n") - sqlite_path = get_db_path() + sqlite_path = resolve_offline_database_path(sqlite_path) if sqlite_path.exists(): response = typer.confirm("sqlite-vec database already exists. Overwrite?") @@ -201,9 +199,11 @@ def migrate_command() -> None: rprint("Migration cancelled") return sqlite_path.unlink() + for suffix in ("-shm", "-wal"): + sqlite_path.with_name(sqlite_path.name + suffix).unlink(missing_ok=True) with console.status("[bold green]Migrating data..."): - success = migrate_from_chromadb() + success = migrate_from_chromadb(sqlite_path) if success: rprint("[bold green]✓[/] Migration completed successfully!") diff --git a/src/brainlayer/index_new.py b/src/brainlayer/index_new.py index 09a967e4..93243838 100644 --- a/src/brainlayer/index_new.py +++ b/src/brainlayer/index_new.py @@ -8,6 +8,7 @@ from .claude_paths import extract_claude_conversation_id as _extract_claude_conversation_id from .embeddings import embed_chunks from .pipeline.chunk import Chunk +from .runtime_store import ReadonlyStore, open_writer_store from .system_prompt_guard import looks_like_system_prompt from .vector_store import IndexDeadlineExceeded, VectorStore @@ -23,6 +24,7 @@ def index_chunks_to_sqlite( db_path: Path = DEFAULT_DB_PATH, on_progress: Optional[Callable[[int, int], None]] = None, deadline_monotonic: float | None = None, + store: VectorStore | None = None, ) -> int: """Index chunks to sqlite-vec database.""" if not chunks: @@ -115,12 +117,15 @@ def embedding_progress(completed: int, total: int) -> None: embeddings.append(ec.embedding) - # Store in database - with VectorStore(db_path) as store: + # A complete CLI index run injects one shared runtime store. Standalone + # callers still get one bounded runtime open for this adapter invocation. + if store is not None: return store.upsert_chunks(chunk_data, embeddings, deadline_monotonic=deadline_monotonic) + with open_writer_store(db_path) as opened_store: + return opened_store.upsert_chunks(chunk_data, embeddings, deadline_monotonic=deadline_monotonic) def get_stats(db_path: Path = DEFAULT_DB_PATH) -> dict: """Get database statistics.""" - with VectorStore(db_path) as store: + with ReadonlyStore(db_path) as store: return store.get_stats() diff --git a/src/brainlayer/mcp/_shared.py b/src/brainlayer/mcp/_shared.py index e24f1591..84a1a441 100644 --- a/src/brainlayer/mcp/_shared.py +++ b/src/brainlayer/mcp/_shared.py @@ -72,16 +72,6 @@ def _search_store_needs_bootstrap(db_path) -> bool: conn.close() -def _bootstrap_search_store(db_path) -> None: - if not _search_store_needs_bootstrap(db_path): - return - - from ..vector_store import VectorStore - - bootstrap_store = VectorStore(db_path) - bootstrap_store.close() - - def _detected_default_read_pool_size() -> int: """Return the platform default for the readonly WAL pool.""" if platform.system() == "Darwin": @@ -129,17 +119,15 @@ def _initialize_search_vector_store_pool() -> None: global _search_vector_store, _search_vector_store_pool, _search_vector_store_pool_handles from ..paths import get_db_path - from ..vector_store import VectorStore + from ..runtime_store import ReadonlyStore db_path = get_db_path() pool_size = _read_pool_size() _assert_read_pool_ram_clamp(pool_size) - _bootstrap_search_store(db_path) - handles = [] try: for _ in range(pool_size): - handles.append(VectorStore(db_path, readonly=True)) + handles.append(ReadonlyStore(db_path)) except Exception: for store in handles: store.close() @@ -164,9 +152,9 @@ def _get_vector_store(timeout: float | None = None): try: if _vector_store is None: from ..paths import get_db_path - from ..vector_store import VectorStore + from ..runtime_store import open_writer_store - _vector_store = VectorStore(get_db_path()) + _vector_store = open_writer_store(get_db_path()) finally: _store_lock.release() return _vector_store diff --git a/src/brainlayer/mcp/store_handler.py b/src/brainlayer/mcp/store_handler.py index dfa56fe0..6b065a16 100644 --- a/src/brainlayer/mcp/store_handler.py +++ b/src/brainlayer/mcp/store_handler.py @@ -54,12 +54,14 @@ def _store_busy_deadline() -> float: def _is_lock_error(exc: BaseException) -> bool: + from ..runtime_store import SchemaFingerprintMismatch from ..vector_store import WriterInUseError text = str(exc).lower() return ( isinstance(exc, apsw.BusyError) or isinstance(exc, WriterInUseError) + or isinstance(exc, SchemaFingerprintMismatch) or "locked" in text or "busy" in text or "sqlite prepare failed" in text @@ -901,11 +903,11 @@ async def _store( db_path = store.db_path def _background_embed_and_flush(): - from ..vector_store import VectorStore as _VS + from ..runtime_store import open_writer_store bg_store = None try: - bg_store = _VS(db_path) + bg_store = open_writer_store(db_path) model = _get_embedding_model() embed_fn = model.embed_query if embed_hot_chunk(store=bg_store, embed_fn=embed_fn, chunk_id=chunk_id): diff --git a/src/brainlayer/migrate.py b/src/brainlayer/migrate.py index 2e4e935b..944f950a 100644 --- a/src/brainlayer/migrate.py +++ b/src/brainlayer/migrate.py @@ -1,5 +1,6 @@ """Migration script to convert ChromaDB data to sqlite-vec.""" +import argparse import logging import os import sys @@ -17,18 +18,17 @@ CHROMADB_AVAILABLE = False from .embeddings import get_embedding_model -from .paths import get_db_path -from .vector_store import VectorStore +from .runtime_store import OfflineMigrator, ReadonlyStore, resolve_offline_database_path logger = logging.getLogger(__name__) # Paths CHROMADB_PATH = Path.home() / ".local" / "share" / "brainlayer" / "chromadb.backup" -SQLITE_PATH = get_db_path() -def migrate_from_chromadb() -> bool: +def migrate_from_chromadb(sqlite_path: Path) -> bool: """Migrate data from ChromaDB to sqlite-vec.""" + sqlite_path = resolve_offline_database_path(sqlite_path) if not CHROMADB_AVAILABLE: print("ChromaDB not available, skipping migration") return False @@ -54,8 +54,8 @@ def migrate_from_chromadb() -> bool: print(f"Found {len(collections)} collections with {total_all} total chunks: {collection_names}") # Create sqlite-vec store - print(f"Creating sqlite-vec database at {SQLITE_PATH}") - vector_store = VectorStore(SQLITE_PATH) + print(f"Creating sqlite-vec database at {sqlite_path}") + vector_store = OfflineMigrator(sqlite_path) # Lazy-load embedding model only if needed embedding_model = None @@ -157,7 +157,7 @@ def migrate_from_chromadb() -> bool: vector_store.close() # Verify migration - vector_store = VectorStore(SQLITE_PATH) + vector_store = ReadonlyStore(sqlite_path) final_count = vector_store.count() vector_store.close() @@ -174,24 +174,28 @@ def migrate_from_chromadb() -> bool: def main(): """Main migration entry point.""" logging.basicConfig(level=logging.INFO) + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("sqlite_path", type=Path, help="Explicit offline sqlite-vec copy target") + args = parser.parse_args() + sqlite_path = resolve_offline_database_path(args.sqlite_path) print("זיכרון - Migration Tool") print("=" * 50) - if SQLITE_PATH.exists(): - response = input(f"sqlite-vec database already exists at {SQLITE_PATH}. Overwrite? (y/N): ") + if sqlite_path.exists(): + response = input(f"sqlite-vec database already exists at {sqlite_path}. Overwrite? (y/N): ") if response.lower() != "y": print("Migration cancelled") return # Remove existing database and WAL/SHM files - SQLITE_PATH.unlink() + sqlite_path.unlink() for suffix in ["-shm", "-wal"]: - p = SQLITE_PATH.parent / (SQLITE_PATH.name + suffix) + p = sqlite_path.parent / (sqlite_path.name + suffix) if p.exists(): p.unlink() - success = migrate_from_chromadb() + success = migrate_from_chromadb(sqlite_path) if success: print("\nMigration completed successfully!") diff --git a/src/brainlayer/runtime_store.py b/src/brainlayer/runtime_store.py new file mode 100644 index 00000000..7eb13f33 --- /dev/null +++ b/src/brainlayer/runtime_store.py @@ -0,0 +1,494 @@ +"""Explicit read, runtime-write, and offline-migration store entrypoints.""" + +from __future__ import annotations + +import hashlib +import json +import os +import threading +from contextlib import contextmanager +from pathlib import Path +from typing import Any + +import apsw +import sqlite_vec + +from .vector_store import ( + _CONNECTION_HOOK_STATE, + VectorStore, + _write_busy_timeout_ms, +) +from .writer_telemetry import start_writer_span + +RUNTIME_SCHEMA_CONTRACT_VERSION = 1 + +_REQUIRED_SCHEMA_OBJECTS = { + "agent_profiles": "table", + "chunk_fts_rowids": "table", + "chunk_id_alias": "table", + "chunk_tags": "table", + "chunk_vectors": "table", + "chunks": "table", + "chunks_fts": "table", + "chunks_fts_operational": "table", + "chunks_fts_trigram": "table", + "dedupe_audit": "table", + "entity_facts": "table", + "health_events": "table", + "kg_entities": "table", + "kg_entity_chunks": "table", + "kg_relations": "table", + "operations": "table", + "schema_migrations": "table", + "session_context": "table", + "session_enrichments": "table", + "tag_tombstones": "table", + "chunk_tags_delete": "trigger", + "chunk_tags_insert": "trigger", + "chunk_tags_update": "trigger", + "chunk_tags_update_clear": "trigger", + "chunks_fts_delete": "trigger", + "chunks_fts_insert": "trigger", + "chunks_fts_operational_insert": "trigger", + "chunks_fts_trigram_delete": "trigger", + "chunks_fts_trigram_insert": "trigger", + "chunks_fts_update": "trigger", + "idx_chunk_id_alias_canonical": "index", + "idx_chunks_content_hash": "index", + "idx_chunks_dedupe_hash": "index", + "idx_chunks_simhash_band_0": "index", + "idx_chunks_simhash_band_1": "index", + "idx_chunks_simhash_band_2": "index", + "idx_chunks_simhash_band_3": "index", +} + +_OPTIONAL_SCHEMA_OBJECTS = { + "chunk_vectors_binary", +} + +_REQUIRED_CHUNK_COLUMNS = { + "aggregated_into", + "archived", + "archived_at", + "brick_id", + "char_count", + "chunk_origin", + "content", + "content_class", + "content_hash", + "content_type", + "conversation_id", + "created_at", + "decay_score", + "dedupe_hash", + "enrich_status", + "enriched_at", + "enrichment_backend", + "enrichment_model", + "enrichment_version", + "epistemic_level", + "external_deps", + "half_life_days", + "id", + "importance", + "ingested_at", + "intent", + "key_facts", + "language", + "last_retrieved", + "last_seen_at", + "metadata", + "position", + "primary_symbols", + "project", + "provenance_class", + "raw_entities_json", + "resolved_queries", + "resolved_query", + "retrieval_count", + "seen_count", + "sender", + "sentiment_label", + "sentiment_score", + "sentiment_signals", + "simhash", + "simhash_band_0", + "simhash_band_1", + "simhash_band_2", + "simhash_band_3", + "source", + "source_file", + "source_project_id", + "source_uri", + "status", + "summary", + "summary_v2", + "superseded_by", + "tag_confidence", + "tags", + "topic_cluster", + "value_type", + "version_scope", +} + +_INTEGER_CHUNK_COLUMNS = { + "archived", + "char_count", + "ingested_at", + "position", + "retrieval_count", + "seen_count", +} +_REAL_CHUNK_COLUMNS = { + "decay_score", + "half_life_days", + "importance", + "last_retrieved", + "sentiment_score", + "tag_confidence", +} +_NOT_NULL_CHUNK_COLUMNS = {"content", "metadata", "source_file"} +_PRIMARY_KEY_CHUNK_COLUMNS = {"id"} +_CHUNK_COLUMN_DEFAULTS: dict[str, frozenset[str | None]] = { + "archived": frozenset({"0"}), + "chunk_origin": frozenset({"'unknown'"}), + "content_class": frozenset({"'knowledge'"}), + "created_at": frozenset({None, "strftime('%Y-%m-%dT%H:%M:%fZ','now')"}), + "decay_score": frozenset({"1.0"}), + "enrichment_version": frozenset({"'1.0'"}), + "half_life_days": frozenset({"30.0"}), + "last_retrieved": frozenset({"NULL"}), + "retrieval_count": frozenset({"0"}), + "seen_count": frozenset({"1"}), + "status": frozenset({"'active'"}), +} + +# Exact normalized definitions for runtime-critical triggers, indexes, and +# virtual tables. chunks_fts deliberately accepts both schemas present in the +# current canonical DB (prefix index) and a freshly migrated DB (no prefix). +_REQUIRED_SQL_HASHES: dict[str, frozenset[str]] = { + "chunk_tags_delete": frozenset({"afe039dd59665608da1d2dfca5693524739713b32ec94d820d9be3e91d91f34d"}), + "chunk_tags_insert": frozenset({"b0da90e12220eec074a72e34efc961f350365965b0566d92d493ade15da2aae6"}), + "chunk_tags_update": frozenset({"0aa51ed4303432c7a1dad25a99a48dd39975f75ac844727ad9bf5df2ff801150"}), + "chunk_tags_update_clear": frozenset({"334ac88d3dec680d9078e9e735ecd67ef96ec97d0b24c292d4ae5c2256b83d1c"}), + "chunk_vectors": frozenset({"75f11e949563f6f93baeaea6c8de8a68d4931b48581808ef06b43099a2eb4c90"}), + "chunks_fts": frozenset( + { + "0b5b4a3dafe4921c2272b1c7536254cd8ef1e69388ac928266b48083f5702034", + "9192e74ca4e3bd65bcaf3f85c596693d61460c549ec87df44aba644b588612dd", + } + ), + "chunks_fts_delete": frozenset({"78d8809fd5864f9c8a8f66bd9547b19872476b838c50dc74bf5dd8a36d1f79b1"}), + "chunks_fts_insert": frozenset({"8d6d5ab6d16a03a8b56dc4ba89918c0d07da5c69639f7fcbd33e29e7ea3a29c0"}), + "chunks_fts_operational": frozenset({"d9541e259b0bf038d1ac5e155951f5637ac9c67f3a5d674919438b644183c85b"}), + "chunks_fts_operational_insert": frozenset({"ba5b681dcc5c4734e39ba0bd5693562653c134b9a75e7887d961ec85428c8e81"}), + "chunks_fts_trigram": frozenset({"86456656a1ec54a6623009d19c7e3da670af08a1474aef81f99e4fe42e46dbab"}), + "chunks_fts_trigram_delete": frozenset({"95b91ed7a93f772fef45020a035f287c9e9c05f23e9464ce824e16517784c127"}), + "chunks_fts_trigram_insert": frozenset({"7bc292a92499f33e38e381c2ba9dc315f57cd20232cb284d7429623cdccf6db8"}), + "chunks_fts_update": frozenset({"04a43eb7d91f2463fdfd8c0c98fd2ef2a0f64e2fabc2468a9ab7bef687365986"}), + "idx_chunk_id_alias_canonical": frozenset({"f14ffbac95a091c3329f3fb8d98bb98b6759d56c41a9c1a5725944ebcc9b45e5"}), + "idx_chunks_content_hash": frozenset({"972b62d3b45116689ab7f38a361de9a98ea109ce58ba724367480e621df9895b"}), + "idx_chunks_dedupe_hash": frozenset({"c66667f3761edc89a9e968e7d184fc8b2eacf05b74012a560aad4b21881940c3"}), + "idx_chunks_simhash_band_0": frozenset({"07c03830b2443d490c6e9997599ab7849c49743729bfa2a1ddc0ae642fbcafd3"}), + "idx_chunks_simhash_band_1": frozenset({"3f8901a6bf1dc70f01f8f8b17b2fc68215f45c062a633a48e6832326340899f3"}), + "idx_chunks_simhash_band_2": frozenset({"1fcdaedbce66e519aa5e25ff55fde96dcc48db507011dd6ec92eb85b030378c2"}), + "idx_chunks_simhash_band_3": frozenset({"ef9832d08371d0f7afd5e72f868b716e1cb84afd849664d5ff9e0d8d74741e57"}), +} + + +def _chunk_column_contract(name: str) -> tuple[str, bool, tuple[str | None, ...], bool]: + if name in _INTEGER_CHUNK_COLUMNS: + column_type = "INTEGER" + elif name in _REAL_CHUNK_COLUMNS: + column_type = "REAL" + else: + column_type = "TEXT" + allowed_defaults = _CHUNK_COLUMN_DEFAULTS.get(name, frozenset({None})) + return ( + column_type, + name in _NOT_NULL_CHUNK_COLUMNS, + tuple(sorted(allowed_defaults, key=lambda value: "" if value is None else value)), + name in _PRIMARY_KEY_CHUNK_COLUMNS, + ) + + +def _normalize_schema_sql(sql: str | None) -> str: + return " ".join(str(sql or "").split()).lower() + + +def _schema_sql_hash(sql: str | None) -> str: + return hashlib.sha256(_normalize_schema_sql(sql).encode("utf-8")).hexdigest() + + +def _contract_payload() -> dict[str, Any]: + return { + "version": RUNTIME_SCHEMA_CONTRACT_VERSION, + "objects": sorted(_REQUIRED_SCHEMA_OBJECTS.items()), + "chunks_columns": [(name, *_chunk_column_contract(name)) for name in sorted(_REQUIRED_CHUNK_COLUMNS)], + "sql_hashes": [(name, sorted(hashes)) for name, hashes in sorted(_REQUIRED_SQL_HASHES.items())], + } + + +def _fingerprint(payload: dict[str, Any]) -> str: + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +EXPECTED_RUNTIME_SCHEMA_FINGERPRINT = _fingerprint(_contract_payload()) + + +class SchemaFingerprintMismatch(RuntimeError): + """Raised when a runtime store cannot prove the expected schema contract.""" + + def __init__( + self, + message: str, + *, + expected_fingerprint: str = EXPECTED_RUNTIME_SCHEMA_FINGERPRINT, + actual_fingerprint: str, + ) -> None: + super().__init__(message) + self.expected_fingerprint = expected_fingerprint + self.actual_fingerprint = actual_fingerprint + + +class RuntimeStoreModeError(ValueError): + """Raised for an invalid runtime-store rollback selector.""" + + +def _canonical_db_path() -> Path: + from .paths import get_db_path + + return get_db_path() + + +def resolve_offline_database_path(db_path: Path) -> Path: + """Resolve an offline target once and reject the configured live database.""" + path = Path(db_path).expanduser().resolve() + canonical = _canonical_db_path().expanduser().resolve() + if path == canonical and os.environ.get("BRAINLAYER_OFFLINE_MIGRATOR_GATED_SWAP") != "1": + raise PermissionError( + "offline migration refuses the configured canonical BrainLayer database; migrate and repair an " + "explicit copy, then use the gated atomic-swap phase" + ) + return path + + +def _mismatch_fingerprint( + *, + objects: dict[str, str], + chunk_columns: dict[str, tuple[str, bool, str | None, bool]], + sql_hashes: dict[str, str], +) -> str: + return _fingerprint( + { + "version": RUNTIME_SCHEMA_CONTRACT_VERSION, + "objects": sorted(objects.items()), + "chunks_columns": sorted((name, *metadata) for name, metadata in chunk_columns.items()), + "sql_hashes": sorted(sql_hashes.items()), + } + ) + + +def _validate_runtime_schema(cursor: apsw.Cursor) -> tuple[str, dict[str, str], set[str]]: + object_names = sorted(_REQUIRED_SCHEMA_OBJECTS.keys() | _OPTIONAL_SCHEMA_OBJECTS) + placeholders = ",".join("?" for _ in object_names) + schema_rows = { + row[0]: (row[1], row[2]) + for row in cursor.execute( + f"SELECT name, type, sql FROM sqlite_schema WHERE name IN ({placeholders})", # noqa: S608 + object_names, + ) + } + actual_objects = {name: object_type for name, (object_type, _sql) in schema_rows.items()} + chunk_column_rows = { + row[1]: (str(row[2]).upper(), bool(row[3]), row[4], bool(row[5])) + for row in cursor.execute("PRAGMA table_info(chunks)") + } + chunk_columns = set(chunk_column_rows) + actual_sql_hashes = { + name: _schema_sql_hash(schema_rows[name][1]) for name in _REQUIRED_SQL_HASHES if name in schema_rows + } + required_actual_objects = { + name: object_type for name, object_type in actual_objects.items() if name in _REQUIRED_SCHEMA_OBJECTS + } + required_actual_columns = { + name: metadata for name, metadata in chunk_column_rows.items() if name in _REQUIRED_CHUNK_COLUMNS + } + actual_fingerprint = _mismatch_fingerprint( + objects=required_actual_objects, + chunk_columns=required_actual_columns, + sql_hashes=actual_sql_hashes, + ) + + missing_objects = sorted( + name for name, expected_type in _REQUIRED_SCHEMA_OBJECTS.items() if actual_objects.get(name) != expected_type + ) + missing_columns = sorted(_REQUIRED_CHUNK_COLUMNS - chunk_columns) + column_definition_mismatches = sorted( + name + for name in _REQUIRED_CHUNK_COLUMNS & chunk_columns + if ( + chunk_column_rows[name][0] != _chunk_column_contract(name)[0] + or chunk_column_rows[name][1] != _chunk_column_contract(name)[1] + or chunk_column_rows[name][2] not in _chunk_column_contract(name)[2] + or chunk_column_rows[name][3] != _chunk_column_contract(name)[3] + ) + ) + sql_definition_mismatches = sorted( + name + for name, allowed_hashes in _REQUIRED_SQL_HASHES.items() + if name in actual_sql_hashes and actual_sql_hashes[name] not in allowed_hashes + ) + if missing_objects or missing_columns or column_definition_mismatches or sql_definition_mismatches: + details: list[str] = [] + if missing_objects: + details.append(f"missing or wrong-type objects: {', '.join(missing_objects)}") + if missing_columns: + details.append(f"missing chunks columns: {', '.join(missing_columns)}") + if column_definition_mismatches: + details.append(f"chunks column definition mismatches: {', '.join(column_definition_mismatches)}") + if sql_definition_mismatches: + details.append(f"definition mismatches: {', '.join(sql_definition_mismatches)}") + raise SchemaFingerprintMismatch( + "runtime schema fingerprint mismatch; " + "; ".join(details), + actual_fingerprint=actual_fingerprint, + ) + return EXPECTED_RUNTIME_SCHEMA_FINGERPRINT, actual_objects, chunk_columns + + +@contextmanager +def _without_connection_maintenance_hooks(): + previous = getattr(_CONNECTION_HOOK_STATE, "skip_maintenance", None) + _CONNECTION_HOOK_STATE.skip_maintenance = True + try: + yield + finally: + if previous is None: + delattr(_CONNECTION_HOOK_STATE, "skip_maintenance") + else: + _CONNECTION_HOOK_STATE.skip_maintenance = previous + + +def _load_vector_extension(conn: apsw.Connection) -> None: + conn.enableloadextension(True) + try: + conn.loadextension(sqlite_vec.loadable_path()) + finally: + conn.enableloadextension(False) + + +class ReadonlyStore(VectorStore): + """Existing-database reader that never initializes or migrates schema.""" + + def __init__(self, db_path: Path): + path = Path(db_path) + if not path.exists(): + actual = _fingerprint({"version": RUNTIME_SCHEMA_CONTRACT_VERSION, "missing_database": True}) + raise SchemaFingerprintMismatch( + f"readonly database does not exist: {path}", + actual_fingerprint=actual, + ) + super().__init__(path, readonly=True) + try: + fingerprint, _objects, _columns = _validate_runtime_schema(self.conn.cursor()) + self.schema_fingerprint = fingerprint + except BaseException: + self.close() + raise + + +class WriterRuntimeStore(VectorStore): + """Existing-database writer with a bounded, schema-only open path.""" + + def __init__(self, db_path: Path): + db_path = Path(db_path) + if not db_path.exists(): + actual = _fingerprint({"version": RUNTIME_SCHEMA_CONTRACT_VERSION, "missing_database": True}) + raise SchemaFingerprintMismatch( + f"runtime database does not exist: {db_path}", + actual_fingerprint=actual, + ) + + self._initialize_instance_state(db_path, readonly=False, create_parent=False) + if self._readonly: + actual = _fingerprint({"version": RUNTIME_SCHEMA_CONTRACT_VERSION, "readonly_database": True}) + raise SchemaFingerprintMismatch( + f"runtime database is not writable: {db_path}", + actual_fingerprint=actual, + ) + + self._acquire_writer_pidfile() + try: + self._init_runtime_db() + except Exception: + self._release_writer_pidfile() + raise + + def _init_runtime_db(self) -> None: + with _without_connection_maintenance_hooks(): + self.conn = apsw.Connection(str(self.db_path), flags=apsw.SQLITE_OPEN_READWRITE) + try: + self.conn.setbusytimeout(_write_busy_timeout_ms()) + _load_vector_extension(self.conn) + except BaseException: + self.conn.close() + raise + + span = start_writer_span( + self.conn, + db_path=self.db_path, + producer="vector_store", + lane="runtime", + operation="runtime_open", + span_kind="writer_operation", + transaction_mode="schema_probe_only", + sample_fts=False, + ) + try: + cursor = self.conn.cursor() + fingerprint, objects, chunk_columns = _validate_runtime_schema(cursor) + self.schema_fingerprint = fingerprint + span.add_metadata( + schema_fingerprint=fingerprint, + schema_contract_version=RUNTIME_SCHEMA_CONTRACT_VERSION, + ) + self._schema_user_version = cursor.execute("PRAGMA user_version").fetchone()[0] + self._has_chunk_origin = "chunk_origin" in chunk_columns + self._has_content_class = "content_class" in chunk_columns + self._has_provenance_class = "provenance_class" in chunk_columns + self._has_superseded_by = "superseded_by" in chunk_columns + self._has_invalid_at = "invalid_at" in chunk_columns + self._binary_index_available = "chunk_vectors_binary" in objects + self._trigram_fts_available = "chunks_fts_trigram" in objects + self._chunk_tags_available = "chunk_tags" in objects + self._local = threading.local() + span.finish("completed") + except BaseException as exc: + actual = getattr(exc, "actual_fingerprint", None) + if actual: + span.add_metadata( + schema_fingerprint=actual, + schema_contract_version=RUNTIME_SCHEMA_CONTRACT_VERSION, + ) + span.finish("error", error=f"{type(exc).__name__}: {exc}") + self.conn.close() + raise + + +class OfflineMigrator(VectorStore): + """Legacy schema/repair store restricted to an explicit offline copy.""" + + def __init__(self, db_path: Path): + path = resolve_offline_database_path(Path(db_path)) + super().__init__(path) + + +def open_writer_store(db_path: Path) -> WriterRuntimeStore | VectorStore: + """Open the default runtime writer, or the guarded legacy rollback path.""" + mode = os.environ.get("BRAINLAYER_RUNTIME_STORE", "runtime").strip().lower() + if mode == "runtime": + return WriterRuntimeStore(Path(db_path)) + if mode == "legacy": + return VectorStore(Path(db_path)) + raise RuntimeStoreModeError("BRAINLAYER_RUNTIME_STORE must be 'runtime' (default) or 'legacy' for rollback") diff --git a/src/brainlayer/vector_store.py b/src/brainlayer/vector_store.py index 76eeb511..b06b5b9e 100644 --- a/src/brainlayer/vector_store.py +++ b/src/brainlayer/vector_store.py @@ -204,11 +204,32 @@ def _set_busy_timeout_hook(conn: apsw.Connection) -> None: conn.setbusytimeout(timeout_ms) -# Register busy_timeout hook BEFORE bestpractice hooks so it fires first. -# bestpractice.apply() adds hooks that run PRAGMA optimize inside Connection(), -# which needs busy_timeout active or it crashes under contention. +# Register busy_timeout hook BEFORE best-practice hooks so it fires first. +# RuntimeStore temporarily suppresses connection_wal/connection_optimize while +# its APSW connection is constructed. Those hooks mutate the database before +# a caller can install telemetry, which violates the runtime-open contract. apsw.connection_hooks.insert(0, _set_busy_timeout_hook) -apsw.bestpractice.apply(apsw.bestpractice.recommended) +_CONNECTION_HOOK_STATE = threading.local() +_CONNECTION_MAINTENANCE_HOOKS = { + apsw.bestpractice.connection_optimize, + apsw.bestpractice.connection_wal, +} + + +def _apply_brainlayer_best_practices(connection: apsw.Connection) -> None: + skip_maintenance = bool(getattr(_CONNECTION_HOOK_STATE, "skip_maintenance", False)) + for hook in apsw.bestpractice.recommended: + if not hook.__name__.startswith("connection_"): + continue + if skip_maintenance and hook in _CONNECTION_MAINTENANCE_HOOKS: + continue + hook(connection) + + +for _best_practice in apsw.bestpractice.recommended: + if not _best_practice.__name__.startswith("connection_"): + _best_practice() +apsw.connection_hooks.append(_apply_brainlayer_best_practices) def _int_env(name: str, default: int) -> int: @@ -283,9 +304,22 @@ class VectorStore(SearchMixin, KGMixin, SessionMixin): _INIT_DB_LOCKS_LOCK = threading.Lock() def __init__(self, db_path: Path, readonly: bool = False): + self._initialize_instance_state(db_path, readonly=readonly, create_parent=not readonly) + if self._readonly: + self._init_readonly_db() + else: + self._acquire_writer_pidfile() + try: + with self._init_db_thread_lock(): + self._init_db_with_retry() + except Exception: + self._release_writer_pidfile() + raise + + def _initialize_instance_state(self, db_path: Path, *, readonly: bool, create_parent: bool) -> None: self.db_path = db_path self._writer_pidfile_acquired = False - if not readonly: + if create_parent: self.db_path.parent.mkdir(parents=True, exist_ok=True) self._fts5_health_cache: dict[str, Any] = {} self._retrieval_strengthening_pending: dict[str, dict[str, float]] = {} @@ -297,16 +331,6 @@ def __init__(self, db_path: Path, readonly: bool = False): self._audit_recursion_count_cache: int | None = None self._audit_recursion_count_cache_data_version: int | None = None self._readonly = readonly or (self.db_path.exists() and not os.access(self.db_path, os.W_OK)) - if self._readonly: - self._init_readonly_db() - else: - self._acquire_writer_pidfile() - try: - with self._init_db_thread_lock(): - self._init_db_with_retry() - except Exception: - self._release_writer_pidfile() - raise def _init_db_thread_lock(self) -> threading.Lock: """Serialize same-process schema init for a DB path.""" diff --git a/src/brainlayer/writer_telemetry.py b/src/brainlayer/writer_telemetry.py index c155f743..d9d6764e 100644 --- a/src/brainlayer/writer_telemetry.py +++ b/src/brainlayer/writer_telemetry.py @@ -338,6 +338,9 @@ def rollback(self, *, error: str | None = None) -> None: def complete(self, *, rows_touched: int | None = None) -> None: return + def add_metadata(self, **_metadata: Any) -> None: + return + def finish( self, outcome: str, @@ -363,6 +366,7 @@ def __init__( span_kind: str, transaction_mode: str, metadata: dict[str, Any] | None, + sample_fts: bool, ) -> None: self.conn = conn self.db_path = Path(db_path) @@ -375,6 +379,7 @@ def __init__( self.span_kind = span_kind self.transaction_mode = transaction_mode self.metadata = dict(metadata or {}) + self.sample_fts = sample_fts self.txn_id = uuid.uuid4().hex self.started_at = _utc_now() self.started_monotonic = _MONOTONIC() @@ -401,10 +406,11 @@ def _total_changes(self) -> int: return 0 def __enter__(self): - try: - self.fts_segments_before = _fts_segment_counts(self.conn, self.db_path) - except Exception: - self.fts_segments_before = {} + if self.sample_fts: + try: + self.fts_segments_before = _fts_segment_counts(self.conn, self.db_path) + except Exception: + self.fts_segments_before = {} try: self.conn.trace_v2(_TRACE_MASK, self._trace, id=self._trace_id) self._trace_installed = True @@ -530,6 +536,9 @@ def complete(self, *, rows_touched: int | None = None) -> None: self._outcome = "completed" self._rows_touched = rows_touched + def add_metadata(self, **metadata: Any) -> None: + self.metadata.update(metadata) + def finish( self, outcome: str, @@ -551,10 +560,13 @@ def _finish(self) -> None: self.conn.trace_v2(_TRACE_MASK, None, id=self._trace_id) except Exception: pass - try: - fts_segments = _fts_segment_counts(self.conn, self.db_path) - except Exception: - fts_segments = dict(self.fts_segments_before) + if self.sample_fts: + try: + fts_segments = _fts_segment_counts(self.conn, self.db_path) + except Exception: + fts_segments = dict(self.fts_segments_before) + else: + fts_segments = {} wal_bytes_after, wal_frames_after = _wal_metrics(self.db_path) changed_rows = max(0, self._total_changes() - self._total_changes_before) with self._lock: @@ -600,6 +612,7 @@ def writer_span( span_kind: str = "transaction", transaction_mode: str = "explicit", metadata: dict[str, Any] | None = None, + sample_fts: bool = True, ): """Create a fail-open observer; the caller retains transaction ownership.""" if not telemetry_enabled(): @@ -617,6 +630,7 @@ def writer_span( span_kind=span_kind, transaction_mode=transaction_mode, metadata=metadata, + sample_fts=sample_fts, ) except Exception: return _NoopWriterSpan() diff --git a/tests/test_cli_direct_sqlite.py b/tests/test_cli_direct_sqlite.py index eff3d949..7c026c29 100644 --- a/tests/test_cli_direct_sqlite.py +++ b/tests/test_cli_direct_sqlite.py @@ -76,7 +76,7 @@ def test_cli_search_does_not_spawn_or_contact_daemon_process(tmp_path: Path) -> assert _daemon_pids() == [] -def test_cli_stats_bootstraps_missing_db_before_readonly(tmp_path: Path) -> None: +def test_cli_stats_missing_db_fails_closed_without_bootstrap(tmp_path: Path) -> None: db_path = tmp_path / "brainlayer.db" env = os.environ.copy() @@ -94,13 +94,13 @@ def test_cli_stats_bootstraps_missing_db_before_readonly(tmp_path: Path) -> None check=False, ) - assert result.returncode == 0, result.stderr + result.stdout - assert db_path.exists() - assert "Total Chunks" in result.stdout + assert result.returncode == 1 + assert "migrate-store " in result.stdout + assert not db_path.exists() assert _daemon_pids() == [] -def test_cli_stats_bootstraps_empty_existing_db_before_readonly(tmp_path: Path) -> None: +def test_cli_stats_empty_existing_db_fails_closed_without_bootstrap(tmp_path: Path) -> None: db_path = tmp_path / "brainlayer.db" db_path.touch() @@ -119,8 +119,9 @@ def test_cli_stats_bootstraps_empty_existing_db_before_readonly(tmp_path: Path) check=False, ) - assert result.returncode == 0, result.stderr + result.stdout - assert "Total Chunks" in result.stdout + assert result.returncode == 1 + assert "migrate-store " in result.stdout + assert db_path.stat().st_size == 0 assert _daemon_pids() == [] @@ -130,11 +131,11 @@ def test_cli_search_opens_vectorstore_readonly_directly(monkeypatch: pytest.Monk db_path = tmp_path / "brainlayer.db" _seed_empty_db(db_path) - calls: list[tuple[Path, bool]] = [] + calls: list[Path] = [] class SpyStore: - def __init__(self, path: Path, readonly: bool = False): - calls.append((Path(path), readonly)) + def __init__(self, path: Path): + calls.append(Path(path)) def hybrid_search(self, **kwargs): return {"ids": [["chunk-1"]], "documents": [["foo result"]], "metadatas": [[{}]], "distances": [[0.1]]} @@ -147,13 +148,13 @@ def embed_query(self, query: str) -> list[float]: return [0.0] * 8 monkeypatch.setenv("BRAINLAYER_DB", str(db_path)) - monkeypatch.setattr(cli_new, "VectorStore", SpyStore) + monkeypatch.setattr(cli_new, "ReadonlyStore", SpyStore) monkeypatch.setattr(cli_new, "get_embedding_model", lambda: SpyModel()) result = CliRunner().invoke(app, ["search", "foo", "--num", "1"]) assert result.exit_code == 0, result.output - assert calls == [(db_path, True)] + assert calls == [db_path] def test_cli_search_agent_flag_threads_to_hybrid_search(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: @@ -165,8 +166,8 @@ def test_cli_search_agent_flag_threads_to_hybrid_search(monkeypatch: pytest.Monk hybrid_calls: list[dict] = [] class SpyStore: - def __init__(self, _path: Path, readonly: bool = False): - self.readonly = readonly + def __init__(self, _path: Path): + pass def hybrid_search(self, **kwargs): hybrid_calls.append(kwargs) @@ -180,7 +181,7 @@ def embed_query(self, query: str) -> list[float]: return [0.0] * 8 monkeypatch.setenv("BRAINLAYER_DB", str(db_path)) - monkeypatch.setattr(cli_new, "VectorStore", SpyStore) + monkeypatch.setattr(cli_new, "ReadonlyStore", SpyStore) monkeypatch.setattr(cli_new, "get_embedding_model", lambda: SpyModel()) result = CliRunner().invoke(app, ["search", "--agent", "codex-test-agent", "foo", "--num", "1"]) @@ -195,11 +196,11 @@ def test_cli_stats_p95_under_200ms_with_direct_readonly_store(monkeypatch: pytes db_path = tmp_path / "brainlayer.db" _seed_empty_db(db_path) - calls: list[tuple[Path, bool]] = [] + calls: list[Path] = [] class FastStatsStore: - def __init__(self, path: Path, readonly: bool = False): - calls.append((Path(path), readonly)) + def __init__(self, path: Path): + calls.append(Path(path)) def get_stats(self): return {"total_chunks": 0, "projects": [], "content_types": []} @@ -208,7 +209,7 @@ def close(self) -> None: pass monkeypatch.setenv("BRAINLAYER_DB", str(db_path)) - monkeypatch.setattr(cli_new, "VectorStore", FastStatsStore) + monkeypatch.setattr(cli_new, "ReadonlyStore", FastStatsStore) runner = CliRunner() durations: list[float] = [] @@ -220,7 +221,7 @@ def close(self) -> None: p95 = sorted(durations)[math.ceil(len(durations) * 0.95) - 1] assert p95 < _cli_stats_p95_budget_seconds() - assert calls == [(db_path, True)] * 10 + assert calls == [db_path] * 10 def test_daemon_and_client_modules_are_gone() -> None: diff --git a/tests/test_cli_index_watchdog.py b/tests/test_cli_index_watchdog.py index b14f3429..75f41b64 100644 --- a/tests/test_cli_index_watchdog.py +++ b/tests/test_cli_index_watchdog.py @@ -23,11 +23,23 @@ def _prepare_index_source(tmp_path, monkeypatch): return source +class _FakeRuntimeStore: + def __init__(self): + self.closed = False + + def __enter__(self): + return self + + def __exit__(self, *_args): + self.closed = True + + def test_index_deadline_exits_nonzero_with_loud_alarm(tmp_path, monkeypatch): source = _prepare_index_source(tmp_path, monkeypatch) monkeypatch.setenv("BRAINLAYER_INDEX_MAX_RUNTIME_S", "7") captured: dict[str, object] = {} alarms: list[BrainLayerAlarm] = [] + runtime_store = _FakeRuntimeStore() def fake_index( _chunks, @@ -36,8 +48,10 @@ def fake_index( project, on_progress, deadline_monotonic=None, + store=None, ): captured["deadline"] = deadline_monotonic + captured["store"] = store raise IndexDeadlineExceeded(processed_count=2) def fake_emit_alarm(alarm): @@ -47,11 +61,14 @@ def fake_emit_alarm(alarm): monkeypatch.setattr("brainlayer.index_new.index_chunks_to_sqlite", fake_index) monkeypatch.setattr("brainlayer.alarm.emit_alarm", fake_emit_alarm) + monkeypatch.setattr("brainlayer.runtime_store.open_writer_store", lambda _path: runtime_store) result = CliRunner().invoke(app, ["index", str(source)]) assert result.exit_code == 1 assert captured["deadline"] == 107.0 + assert captured["store"] is runtime_store + assert runtime_store.closed is True assert "BRAINLAYER_ALARM INDEX_RUNTIME_EXCEEDED" in result.stderr assert len(alarms) == 1 assert alarms[0].context["max_runtime_s"] == 7.0 @@ -63,6 +80,7 @@ def test_fast_index_under_deadline_completes_without_alarm(tmp_path, monkeypatch monkeypatch.setenv("BRAINLAYER_INDEX_MAX_RUNTIME_S", "7") captured: dict[str, object] = {} alarms: list[BrainLayerAlarm] = [] + runtime_store = _FakeRuntimeStore() def fake_index( _chunks, @@ -71,17 +89,21 @@ def fake_index( project, on_progress, deadline_monotonic=None, + store=None, ): captured["deadline"] = deadline_monotonic + captured["store"] = store return 3 monkeypatch.setattr("brainlayer.index_new.index_chunks_to_sqlite", fake_index) monkeypatch.setattr("brainlayer.alarm.emit_alarm", lambda alarm: alarms.append(alarm) or True) + monkeypatch.setattr("brainlayer.runtime_store.open_writer_store", lambda _path: runtime_store) result = CliRunner().invoke(app, ["index", str(source)]) assert result.exit_code == 0 assert captured["deadline"] == 107.0 + assert captured["store"] is runtime_store assert alarms == [] assert "Indexed 3 chunks" in result.stdout @@ -92,9 +114,11 @@ def test_index_deadline_trips_after_file_that_produces_no_writes(tmp_path, monke monotonic_values = iter([100.0, 100.0, 108.0]) monkeypatch.setattr(cli.time, "monotonic", lambda: next(monotonic_values, 108.0)) alarms: list[BrainLayerAlarm] = [] + runtime_store = _FakeRuntimeStore() monkeypatch.setattr("brainlayer.index_new.index_chunks_to_sqlite", lambda *_args, **_kwargs: 0) monkeypatch.setattr("brainlayer.alarm.emit_alarm", lambda alarm: alarms.append(alarm) or True) + monkeypatch.setattr("brainlayer.runtime_store.open_writer_store", lambda _path: runtime_store) result = CliRunner().invoke(app, ["index", str(source)]) @@ -110,6 +134,7 @@ def test_index_deadline_stops_before_parsing_next_entry(tmp_path, monkeypatch): monotonic_values = iter([100.0, 100.0, 108.0]) monkeypatch.setattr(cli.time, "monotonic", lambda: next(monotonic_values, 108.0)) index_called = False + runtime_store = _FakeRuntimeStore() def fake_index(*_args, **_kwargs): nonlocal index_called @@ -118,6 +143,7 @@ def fake_index(*_args, **_kwargs): monkeypatch.setattr("brainlayer.index_new.index_chunks_to_sqlite", fake_index) monkeypatch.setattr("brainlayer.alarm.emit_alarm", lambda _alarm: True) + monkeypatch.setattr("brainlayer.runtime_store.open_writer_store", lambda _path: runtime_store) result = CliRunner().invoke(app, ["index", str(source)]) @@ -155,7 +181,7 @@ def fake_embed(chunks, on_progress=None): ) -def test_index_adapter_forwards_deadline_to_vector_store(tmp_path, monkeypatch): +def test_index_adapter_forwards_deadline_to_runtime_store(tmp_path, monkeypatch): import brainlayer.index_new as index_new source_file = tmp_path / "session.jsonl" @@ -169,7 +195,7 @@ def test_index_adapter_forwards_deadline_to_vector_store(tmp_path, monkeypatch): ) captured: dict[str, object] = {} - class FakeVectorStore: + class FakeRuntimeStore: def __init__(self, db_path): captured["db_path"] = db_path @@ -186,7 +212,7 @@ def upsert_chunks(self, chunks, embeddings, *, deadline_monotonic=None): monkeypatch.setattr( index_new, "embed_chunks", lambda chunks, on_progress=None: [SimpleNamespace(chunk=chunk, embedding=[0.1])] ) - monkeypatch.setattr(index_new, "VectorStore", FakeVectorStore) + monkeypatch.setattr(index_new, "open_writer_store", FakeRuntimeStore) result = index_new.index_chunks_to_sqlite( [chunk], @@ -197,3 +223,31 @@ def upsert_chunks(self, chunks, embeddings, *, deadline_monotonic=None): assert result == 1 assert captured["deadline"] == 42.0 + + +def test_index_reuses_one_runtime_store_across_source_files(tmp_path, monkeypatch): + source = _prepare_index_source(tmp_path, monkeypatch) + project = source / "watchdog-test" + (project / "session-two.jsonl").write_text("{}\n") + opened: list[_FakeRuntimeStore] = [] + seen_stores: list[object] = [] + + def fake_open(_path): + store = _FakeRuntimeStore() + opened.append(store) + return store + + def fake_index(_chunks, **kwargs): + seen_stores.append(kwargs["store"]) + return 1 + + monkeypatch.setattr("brainlayer.runtime_store.open_writer_store", fake_open) + monkeypatch.setattr("brainlayer.index_new.index_chunks_to_sqlite", fake_index) + monkeypatch.setattr("brainlayer.alarm.emit_alarm", lambda _alarm: True) + + result = CliRunner().invoke(app, ["index", str(source)]) + + assert result.exit_code == 0, result.output + assert len(opened) == 1 + assert seen_stores == [opened[0], opened[0]] + assert opened[0].closed is True diff --git a/tests/test_context_pipeline.py b/tests/test_context_pipeline.py index edba80ca..52f32caa 100644 --- a/tests/test_context_pipeline.py +++ b/tests/test_context_pipeline.py @@ -12,6 +12,13 @@ from brainlayer.pipeline.classify import ClassifiedContent, ContentType, classify_content from brainlayer.vector_store import VectorStore + +def _migrate_runtime_copy(db_path): + from brainlayer.runtime_store import OfflineMigrator + + OfflineMigrator(db_path).close() + + # ── classify_content should extract session_id ── @@ -159,6 +166,7 @@ def test_conversation_id_stored(self, tmp_path): from unittest.mock import patch db_path = tmp_path / "test.db" + _migrate_runtime_copy(db_path) # Create chunks with session_id in metadata chunks = [ @@ -225,6 +233,7 @@ def test_position_is_sequential(self, tmp_path): from unittest.mock import patch db_path = tmp_path / "test.db" + _migrate_runtime_copy(db_path) chunks = [] for i in range(5): @@ -264,6 +273,7 @@ def test_missing_session_id_uses_file_stem(self, tmp_path): from unittest.mock import patch db_path = tmp_path / "test.db" + _migrate_runtime_copy(db_path) chunk = Chunk( content="A message without session ID metadata for some reason or another here", @@ -303,6 +313,7 @@ def test_skips_system_prompt_chunks_marked_in_metadata(self, tmp_path): from unittest.mock import patch db_path = tmp_path / "test.db" + _migrate_runtime_copy(db_path) from brainlayer.pipeline.classify import ContentValue chunks = [ @@ -347,6 +358,7 @@ def test_skips_system_prompt_chunks_by_content_pattern(self, tmp_path): from unittest.mock import patch db_path = tmp_path / "test.db" + _migrate_runtime_copy(db_path) from brainlayer.pipeline.classify import ContentValue chunks = [ @@ -396,6 +408,7 @@ class TestGetContextWorks: def _make_store_with_conversation(self, tmp_path): """Create a store with a 5-chunk conversation.""" db_path = tmp_path / "test.db" + _migrate_runtime_copy(db_path) store = VectorStore(db_path) cursor = store.conn.cursor() @@ -502,6 +515,7 @@ def test_index_fast_populates_context(self, tmp_path): self._write_jsonl(jsonl_path, entries) db_path = tmp_path / "test.db" + _migrate_runtime_copy(db_path) # Mock embeddings to avoid loading model with patch("brainlayer.index_new.embed_chunks") as mock_embed: diff --git a/tests/test_runtime_store.py b/tests/test_runtime_store.py new file mode 100644 index 00000000..0bf7108d --- /dev/null +++ b/tests/test_runtime_store.py @@ -0,0 +1,295 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import apsw +import pytest +from typer.testing import CliRunner + +import brainlayer.runtime_store as runtime_store_module +from brainlayer.runtime_store import ( + OfflineMigrator, + ReadonlyStore, + RuntimeStoreModeError, + SchemaFingerprintMismatch, + WriterRuntimeStore, + open_writer_store, +) +from brainlayer.vector_store import VectorStore + + +def test_migrate_store_command_requires_explicit_copy_path(tmp_path, monkeypatch): + from brainlayer.cli import app + + canonical = tmp_path / "canonical" / "brainlayer.db" + copy_path = tmp_path / "copy" / "brainlayer.db" + monkeypatch.setattr("brainlayer.runtime_store._canonical_db_path", lambda: canonical) + + missing = CliRunner().invoke(app, ["migrate-store"]) + assert missing.exit_code != 0 + + migrated = CliRunner().invoke(app, ["migrate-store", str(copy_path)]) + assert migrated.exit_code == 0, migrated.output + assert copy_path.exists() + + refused = CliRunner().invoke(app, ["migrate-store", str(canonical)]) + assert refused.exit_code == 1 + assert "canonical" in refused.output.lower() + + +def test_chromadb_migrate_command_requires_explicit_offline_path(tmp_path, monkeypatch): + from brainlayer.cli import app + + canonical = tmp_path / "canonical" / "brainlayer.db" + monkeypatch.setattr("brainlayer.runtime_store._canonical_db_path", lambda: canonical) + + missing = CliRunner().invoke(app, ["migrate"]) + assert missing.exit_code != 0 + + refused = CliRunner().invoke(app, ["migrate", str(canonical)]) + assert refused.exit_code == 1 + assert "canonical" in refused.output.lower() + assert not canonical.exists() + + +def test_repair_fts_command_requires_explicit_offline_copy(tmp_path, monkeypatch): + from brainlayer.cli import app + + canonical = tmp_path / "canonical" / "brainlayer.db" + copy_path = tmp_path / "copy" / "brainlayer.db" + monkeypatch.setattr("brainlayer.runtime_store._canonical_db_path", lambda: canonical) + OfflineMigrator(copy_path).close() + + missing = CliRunner().invoke(app, ["repair-fts"]) + assert missing.exit_code != 0 + + repaired = CliRunner().invoke(app, ["repair-fts", str(copy_path)]) + assert repaired.exit_code == 0, repaired.output + + refused = CliRunner().invoke(app, ["repair-fts", str(canonical)]) + assert refused.exit_code == 1 + assert "canonical" in refused.output.lower() + + +def _bootstrap(db_path: Path) -> None: + store = VectorStore(db_path) + store.close() + + +def _telemetry_events(log_path: Path, operation: str) -> list[dict]: + if not log_path.exists(): + return [] + return [ + event + for line in log_path.read_text(encoding="utf-8").splitlines() + if (event := json.loads(line)).get("operation") == operation + ] + + +def test_readonly_store_uses_readonly_connection_without_legacy_init(tmp_path, monkeypatch): + db_path = tmp_path / "readonly.db" + _bootstrap(db_path) + + def fail_legacy_init(self): + raise AssertionError("readonly store must not run legacy initialization") + + monkeypatch.setattr(VectorStore, "_init_db_with_retry", fail_legacy_init) + + with ReadonlyStore(db_path) as store: + assert store._readonly is True + assert store.conn.readonly("main") is True + with pytest.raises(apsw.ReadOnlyError): + store.conn.cursor().execute( + "INSERT INTO chunks (id, content, metadata, source_file) VALUES ('x','x','{}','x')" + ) + + +def test_runtime_store_opens_existing_schema_without_legacy_init(tmp_path, monkeypatch): + db_path = tmp_path / "runtime.db" + _bootstrap(db_path) + + def fail_legacy_init(self): + raise AssertionError("runtime store must not run legacy initialization") + + monkeypatch.setattr(VectorStore, "_init_db_with_retry", fail_legacy_init) + + with WriterRuntimeStore(db_path) as store: + assert store._readonly is False + assert store.conn.readonly("main") is False + assert len(store.schema_fingerprint) == 64 + + +def test_runtime_store_closes_connection_when_extension_setup_fails(tmp_path, monkeypatch): + db_path = tmp_path / "extension-failure.db" + _bootstrap(db_path) + real_connection = runtime_store_module.apsw.Connection + closed: list[bool] = [] + + class TrackedConnection: + def __init__(self, *args, **kwargs): + self._connection = real_connection(*args, **kwargs) + + def __getattr__(self, name): + return getattr(self._connection, name) + + def close(self): + closed.append(True) + return self._connection.close() + + monkeypatch.setattr(runtime_store_module.apsw, "Connection", TrackedConnection) + monkeypatch.setattr( + runtime_store_module, + "_load_vector_extension", + lambda _conn: (_ for _ in ()).throw(RuntimeError("extension load failed")), + ) + + with pytest.raises(RuntimeError, match="extension load failed"): + WriterRuntimeStore(db_path) + + assert closed == [True] + + +def test_runtime_store_missing_database_fails_closed_without_creating_it(tmp_path): + db_path = tmp_path / "missing" / "runtime.db" + + with pytest.raises(SchemaFingerprintMismatch, match="does not exist"): + WriterRuntimeStore(db_path) + + assert not db_path.exists() + assert not db_path.parent.exists() + + +def test_runtime_store_schema_mismatch_fails_closed_and_releases_pidfile(tmp_path, monkeypatch): + db_path = tmp_path / "stale.db" + pidfile_dir = tmp_path / "pidfiles" + monkeypatch.setenv("BRAINLAYER_WRITER_PIDFILE_DIR", str(pidfile_dir)) + _bootstrap(db_path) + + conn = apsw.Connection(str(db_path)) + conn.cursor().execute("DROP TRIGGER chunks_fts_insert") + conn.close() + + with pytest.raises(SchemaFingerprintMismatch, match="chunks_fts_insert") as exc_info: + WriterRuntimeStore(db_path) + + assert exc_info.value.expected_fingerprint + assert exc_info.value.actual_fingerprint + assert list(pidfile_dir.glob("*.pid")) == [] + + +def test_runtime_store_rejects_same_name_trigger_with_wrong_definition(tmp_path): + db_path = tmp_path / "wrong-trigger.db" + _bootstrap(db_path) + conn = apsw.Connection(str(db_path)) + cursor = conn.cursor() + cursor.execute("DROP TRIGGER chunks_fts_insert") + cursor.execute("CREATE TRIGGER chunks_fts_insert AFTER INSERT ON chunks BEGIN SELECT 1; END") + conn.close() + + with pytest.raises(SchemaFingerprintMismatch, match="definition mismatches: chunks_fts_insert"): + WriterRuntimeStore(db_path) + + +def test_runtime_open_telemetry_has_fingerprint_and_no_scan_or_mutation_statements(tmp_path, monkeypatch): + db_path = tmp_path / "telemetry.db" + log_path = tmp_path / "runtime-open.jsonl" + monkeypatch.setenv("BRAINLAYER_WRITER_TELEMETRY", "1") + monkeypatch.setenv("BRAINLAYER_WRITER_TELEMETRY_PATH", str(log_path)) + monkeypatch.setenv("BRAINLAYER_WRITER_HEARTBEAT_DIR", str(tmp_path / "heartbeats")) + monkeypatch.setenv("BRAINLAYER_WRITER_TELEMETRY_FTS_SAMPLE_TTL_SECONDS", "0") + _bootstrap(db_path) + log_path.unlink(missing_ok=True) + + with WriterRuntimeStore(db_path) as store: + fingerprint = store.schema_fingerprint + + finished = [event for event in _telemetry_events(log_path, "runtime_open") if event["event"] == "txn_finished"] + assert len(finished) == 1 + assert finished[0]["outcome"] == "completed" + assert finished[0]["schema_fingerprint"] == fingerprint + assert finished[0]["duration_ms"] < 100 + assert finished[0]["fts_segments_before"] == {} + assert finished[0]["fts_segments"] == {} + statements = finished[0]["statements"] + assert statements + normalized = [statement["normalized_sql"].upper() for statement in statements] + forbidden = ("CREATE ", "DROP ", "ALTER ", "INSERT ", "UPDATE ", "DELETE ", "COUNT(", "OPTIMIZE") + assert not any(token in statement for statement in normalized for token in forbidden) + corpus_tables = ("FROM CHUNKS ", "FROM CHUNKS_FTS", "FROM CHUNK_FTS_ROWIDS") + assert not any(table in statement for statement in normalized for table in corpus_tables) + assert all( + statement["fullscan_steps"] == 0 + or "SQLITE_SCHEMA" in statement["normalized_sql"].upper() + or statement["normalized_sql"].upper().startswith("PRAGMA TABLE_INFO") + for statement in statements + ) + + +def test_open_writer_store_defaults_new_and_legacy_flag_rolls_back(tmp_path, monkeypatch): + db_path = tmp_path / "factory.db" + _bootstrap(db_path) + + monkeypatch.delenv("BRAINLAYER_RUNTIME_STORE", raising=False) + with open_writer_store(db_path) as store: + assert type(store) is WriterRuntimeStore + + monkeypatch.setenv("BRAINLAYER_RUNTIME_STORE", "legacy") + with open_writer_store(db_path) as store: + assert type(store) is VectorStore + + +def test_open_writer_store_rejects_unknown_mode(tmp_path, monkeypatch): + db_path = tmp_path / "factory.db" + _bootstrap(db_path) + monkeypatch.setenv("BRAINLAYER_RUNTIME_STORE", "surprise") + + with pytest.raises(RuntimeStoreModeError, match="BRAINLAYER_RUNTIME_STORE"): + open_writer_store(db_path) + + +def test_offline_migrator_allows_copy_and_refuses_canonical(tmp_path, monkeypatch): + canonical = tmp_path / "canonical" / "brainlayer.db" + copy_path = tmp_path / "copy" / "brainlayer.db" + monkeypatch.setattr("brainlayer.runtime_store._canonical_db_path", lambda: canonical) + + with OfflineMigrator(copy_path): + pass + assert copy_path.exists() + + with pytest.raises(PermissionError, match="canonical"): + OfflineMigrator(canonical) + assert not canonical.exists() + + +def test_offline_migrator_refuses_configured_database_path(tmp_path, monkeypatch): + configured = tmp_path / "configured-live.db" + monkeypatch.setenv("BRAINLAYER_DB", str(configured)) + + with pytest.raises(PermissionError, match="canonical"): + OfflineMigrator(configured) + + assert not configured.exists() + + +def test_offline_migrator_canonical_requires_gated_swap_override(tmp_path, monkeypatch): + canonical = tmp_path / "canonical" / "brainlayer.db" + monkeypatch.setattr("brainlayer.runtime_store._canonical_db_path", lambda: canonical) + monkeypatch.setenv("BRAINLAYER_OFFLINE_MIGRATOR_GATED_SWAP", "1") + + with OfflineMigrator(canonical): + pass + + assert canonical.exists() + + +def test_offline_migrator_opens_the_once_resolved_copy_path(tmp_path, monkeypatch): + canonical = tmp_path / "canonical" / "brainlayer.db" + copy_path = tmp_path / "copy" / "brainlayer.db" + link_path = tmp_path / "copy-link.db" + monkeypatch.setattr("brainlayer.runtime_store._canonical_db_path", lambda: canonical) + OfflineMigrator(copy_path).close() + link_path.symlink_to(copy_path) + + with OfflineMigrator(link_path) as store: + assert store.db_path == copy_path.resolve() diff --git a/tests/test_runtime_store_benchmark.py b/tests/test_runtime_store_benchmark.py new file mode 100644 index 00000000..acf128a3 --- /dev/null +++ b/tests/test_runtime_store_benchmark.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import pytest + +import scripts.benchmark_runtime_store_open as benchmark_module +from brainlayer.runtime_store import OfflineMigrator +from scripts.benchmark_runtime_store_open import benchmark_runtime_open + + +def test_runtime_open_benchmark_refuses_canonical_path(tmp_path, monkeypatch): + canonical = tmp_path / "canonical" / "brainlayer.db" + monkeypatch.setattr("scripts.benchmark_runtime_store_open._canonical_db_path", lambda: canonical) + + with pytest.raises(PermissionError, match="canonical"): + benchmark_runtime_open(canonical, iterations=1, telemetry_path=tmp_path / "telemetry.jsonl") + + +def test_runtime_open_benchmark_reports_percentiles_and_zero_corpus_scans(tmp_path): + db_path = tmp_path / "copy" / "brainlayer.db" + telemetry_path = tmp_path / "runtime-open.jsonl" + OfflineMigrator(db_path).close() + + result = benchmark_runtime_open(db_path, iterations=5, telemetry_path=telemetry_path) + + assert result["mode"] == "runtime" + assert result["iterations"] == 5 + assert result["runtime_open_events"] == 5 + assert result["p50_ms"] <= result["p99_ms"] <= result["max_ms"] + assert result["p99_ms"] < 100 + assert result["schema_fingerprints"] + assert result["statement_violations"] == [] + assert result["corpus_scan_statements"] == [] + + +def test_runtime_open_benchmark_requires_positive_iterations(tmp_path): + with pytest.raises(ValueError, match="iterations"): + benchmark_runtime_open( + tmp_path / "copy.db", + iterations=0, + telemetry_path=tmp_path / "telemetry.jsonl", + ) + + +@pytest.mark.parametrize("telemetry_target", ["copy", "canonical"]) +def test_runtime_open_benchmark_refuses_database_as_telemetry_path(tmp_path, monkeypatch, telemetry_target): + canonical = tmp_path / "canonical" / "brainlayer.db" + copy_path = tmp_path / "copy" / "brainlayer.db" + monkeypatch.setattr("scripts.benchmark_runtime_store_open._canonical_db_path", lambda: canonical) + OfflineMigrator(copy_path).close() + target = copy_path if telemetry_target == "copy" else canonical + + with pytest.raises(PermissionError, match="telemetry path"): + benchmark_runtime_open(copy_path, iterations=1, telemetry_path=target) + + +def test_runtime_open_benchmark_rejects_incomplete_telemetry(tmp_path, monkeypatch): + db_path = tmp_path / "copy" / "brainlayer.db" + OfflineMigrator(db_path).close() + monkeypatch.setattr(benchmark_module, "_runtime_finished_events", lambda _path: []) + + with pytest.raises(RuntimeError, match="expected telemetry for 2 runtime opens, got 0"): + benchmark_runtime_open( + db_path, + iterations=2, + telemetry_path=tmp_path / "runtime-open.jsonl", + ) diff --git a/tests/test_store_handler.py b/tests/test_store_handler.py index 47e42de3..03d6e339 100644 --- a/tests/test_store_handler.py +++ b/tests/test_store_handler.py @@ -100,6 +100,34 @@ async def test_busy_queue_fallback_returns_queued_chunk_id(tmp_path): assert any(expected_chunk_id in item.text for item in texts) +@pytest.mark.asyncio +async def test_schema_fingerprint_mismatch_fails_closed_to_durable_queue(tmp_path): + """A stale runtime schema must preserve the producer write as a queued intent.""" + from brainlayer.mcp.store_handler import _store + from brainlayer.runtime_store import SchemaFingerprintMismatch + + queue_dir = tmp_path / "queue" + mismatch = SchemaFingerprintMismatch( + "runtime schema fingerprint mismatch", + actual_fingerprint="0" * 64, + ) + + with ( + patch("brainlayer.mcp.store_handler._get_vector_store", side_effect=mismatch), + patch("brainlayer.queue_io.get_queue_dir", return_value=queue_dir), + ): + texts, structured = await _store( + content="schema mismatch must preserve this write", + memory_type="note", + project="brainlayer", + ) + + assert structured["queued"] is True + assert structured["status"] == "DEFERRED" + assert len(list(queue_dir.glob("mcp-*.jsonl"))) == 1 + assert any("DEFERRED" in item.text for item in texts) + + @pytest.mark.asyncio async def test_busy_queue_fallback_returns_loud_deferred_receipt(tmp_path): """DB-busy fallback is a structured DEFERRED receipt, not quiet success.""" @@ -253,6 +281,9 @@ async def test_store_saves_normally_when_uncontended(tmp_path, monkeypatch): db_path = tmp_path / "brainlayer.db" queue_dir = tmp_path / "queue" started_threads = [] + from brainlayer.runtime_store import OfflineMigrator + + OfflineMigrator(db_path).close() class NoopThread: def __init__(self, *, target, daemon): @@ -339,6 +370,7 @@ def held_write_lock_store_memory(**kwargs): raise apsw.BusyError("database is locked") monkeypatch.setenv("BRAINLAYER_STORE_BUSY_BUDGET_MS", "100") + monkeypatch.setenv("BRAINLAYER_RUNTIME_STORE", "legacy") monkeypatch.setattr("brainlayer.mcp.store_handler._retry_delay", 0.001) with ( @@ -636,6 +668,7 @@ def fake_init_db(self): self.conn.setbusytimeout(vector_store_module._write_busy_timeout_ms()) monkeypatch.setenv("BRAINLAYER_STORE_BUSY_BUDGET_MS", "100") + monkeypatch.setenv("BRAINLAYER_RUNTIME_STORE", "legacy") monkeypatch.setattr(_shared, "_vector_store", None) monkeypatch.setattr("brainlayer.paths.get_db_path", lambda: db_path) monkeypatch.setattr(VectorStore, "_INIT_MAX_RETRIES", 1) diff --git a/tests/test_vector_store_readonly.py b/tests/test_vector_store_readonly.py index e5894cd7..d3c18692 100644 --- a/tests/test_vector_store_readonly.py +++ b/tests/test_vector_store_readonly.py @@ -7,6 +7,7 @@ from brainlayer._helpers import serialize_f32 from brainlayer.mcp import _shared, search_handler +from brainlayer.runtime_store import ReadonlyStore, SchemaFingerprintMismatch from brainlayer.search_repo import _hybrid_cache from brainlayer.vector_store import VectorStore @@ -180,25 +181,19 @@ def fake_init_readonly(self): assert not db_path.parent.exists() -def test_search_vector_store_bootstraps_missing_db_then_reopens_readonly(tmp_path, monkeypatch): +def test_search_vector_store_missing_db_fails_closed_without_bootstrap(tmp_path, monkeypatch): db_path = tmp_path / "fresh" / "brainlayer.db" monkeypatch.setenv("BRAINLAYER_DB", str(db_path)) monkeypatch.setenv("BRAINLAYER_READ_POOL_SIZE", "2") - store = _shared._get_search_vector_store() - try: - assert db_path.exists() - assert store._readonly is True - assert store.count() == 0 - with pytest.raises(apsw.ReadOnlyError): - store.conn.cursor().execute( - "INSERT INTO chunks (id, content, metadata, source_file) VALUES ('x', 'x', '{}', 'x')" - ) - finally: - _shared._close_search_vector_store() + with pytest.raises(SchemaFingerprintMismatch, match="does not exist"): + _shared._get_search_vector_store() + + assert not db_path.exists() + assert not db_path.parent.exists() -def test_search_vector_store_bootstraps_stale_schema_then_reopens_readonly(tmp_path, monkeypatch): +def test_search_vector_store_stale_schema_fails_closed_without_migration(tmp_path, monkeypatch): db_path = tmp_path / "stale.db" conn = apsw.Connection(str(db_path)) conn.cursor().execute( @@ -226,17 +221,13 @@ def test_search_vector_store_bootstraps_stale_schema_then_reopens_readonly(tmp_p monkeypatch.setenv("BRAINLAYER_DB", str(db_path)) monkeypatch.setenv("BRAINLAYER_READ_POOL_SIZE", "2") - store = _shared._get_search_vector_store() - try: - columns = {row[1] for row in store.conn.cursor().execute("PRAGMA table_info(chunks)")} - assert {"status", "archived", "summary", "resolved_queries", "chunk_origin"}.issubset(columns) - assert store._readonly is True - with pytest.raises(apsw.ReadOnlyError): - store.conn.cursor().execute( - "INSERT INTO chunks (id, content, metadata, source_file) VALUES ('x', 'x', '{}', 'x')" - ) - finally: - _shared._close_search_vector_store() + with pytest.raises(SchemaFingerprintMismatch, match="missing or wrong-type objects"): + _shared._get_search_vector_store() + + inspector = apsw.Connection(str(db_path)) + columns = {row[1] for row in inspector.cursor().execute("PRAGMA table_info(chunks)")} + inspector.close() + assert not {"status", "archived", "summary", "resolved_queries"} & columns def test_search_store_bootstrap_required_for_partial_kg_schema(tmp_path): @@ -269,6 +260,7 @@ def test_search_store_pool_preopens_fixed_readonly_handles(tmp_path, monkeypatch id(store) for store in _shared._search_vector_store_pool.queue } assert all(store._readonly for store in _shared._search_vector_store_pool_handles) + assert all(type(store) is ReadonlyStore for store in _shared._search_vector_store_pool_handles) def test_search_store_checkout_deserializes_slow_reads(tmp_path, monkeypatch):