"""Measure the SQLite search index on a deterministic synthetic corpus. This is an adapter benchmark, not the PRD's end-to-end or semantic-quality gate. Run with ``./scripts/python-tool python scripts/benchmark_search.py``. """ from __future__ import annotations import argparse import hashlib import json import math import os import platform import sqlite3 import statistics import tempfile import time from collections.abc import Sequence from datetime import UTC, datetime from pathlib import Path from gewu.adapters.sqlite_search import SQLiteSearchIndex from gewu.domain.contracts import ( Claim, Confidence, EvidenceReference, KnowledgeNode, Lifecycle, Locator, NodeType, ReviewState, SupportAssessment, SupportStatus, ) from gewu.domain.validation import assertion_digest, evidence_digest def base_node() -> KnowledgeNode: text = "Interactive experiences welcomed visitors during the synthetic benchmark." assertion = assertion_digest(text) evidence = EvidenceReference( source_revision_id="benchmark:source:r1", extraction_id="a" * 64, assertion_path="claims[0].text", assertion_digest=assertion, quote=text, locator=Locator(kind="text_span", block_id="benchmark-block", start=0, end=len(text)), normalization_profile="unicode-nfc-lf-v1", ) now = datetime.now(UTC) claim = Claim( claim_id="benchmark:claim", text=text, evidence_type="reported", evidence=(evidence,), assessment=SupportAssessment( assessment_id="benchmark:assessment", assertion_digest=assertion, evidence_digests=(evidence_digest(evidence),), assessor_id="benchmark:assessor", run_id="benchmark:run", method="synthetic benchmark fixture", status=SupportStatus.SUPPORTED, rationale="Fixture only; not a semantic assessment.", ), ) return KnowledgeNode( node_id="benchmark:template", project_id="bench-project-000", node_type=NodeType.FINDING, title="Interactive visitor evidence template", summary=text, created=now, updated=now, review_state=ReviewState.REVIEWED, lifecycle=Lifecycle.ACTIVE, confidence=Confidence.MEDIUM, claim_support="supported", run_id="benchmark:run", agent_id="benchmark:agent", model_id="synthetic-fixture", claims=(claim,), ) class SyntheticRepository: """Small in-memory canonical port that feeds validated production records.""" def __init__(self, nodes_per_project: int, projects: int) -> None: revision = hashlib.sha256(f"benchmark:{nodes_per_project}:{projects}".encode()).hexdigest() self._revision = revision self._records: dict[str, bytes] = {} self.project_ids = tuple(f"bench-project-{index:03d}" for index in range(projects)) template = base_node() for project_index, project_id in enumerate(self.project_ids): project_key = hashlib.sha256(project_id.encode()).hexdigest() for node_index in range(nodes_per_project): node_id = f"benchmark:{project_index:03d}:node-{node_index:06d}" node_key = hashlib.sha256(node_id.encode()).hexdigest() path = f"projects/{project_key}/nodes/{node_key}.yaml" node = template.model_copy( update={ "node_id": node_id, "project_id": project_id, "title": f"Interactive visitor evidence {project_index} {node_index}", "summary": ( "Interactive experiences welcomed visitors; " f"synthetic record {node_index}." ), } ) self._records[path] = node.model_dump_json().encode("utf-8") def head(self) -> str: return self._revision def list_files(self, revision: str, prefix: str) -> tuple[str, ...]: if revision != self._revision: raise ValueError("Unknown synthetic revision") return tuple(path for path in self._records if path.startswith(prefix)) def read_file(self, revision: str, path: str) -> bytes | None: if revision != self._revision: raise ValueError("Unknown synthetic revision") return self._records.get(path) def percentile(samples: Sequence[float], value: float) -> float: ordered = sorted(samples) position = max(0, math.ceil(len(ordered) * value) - 1) return ordered[position] def processor_name() -> str: reported = platform.processor() if reported: return reported if platform.system() == "Linux": try: for line in Path("/proc/cpuinfo").read_text(encoding="utf-8").splitlines(): if line.startswith("model name"): return line.partition(":")[2].strip() except OSError: pass return "not reported" def run(nodes_per_project: int, projects: int, samples: int, root: Path) -> dict[str, object]: build_started = time.perf_counter() repository = SyntheticRepository(nodes_per_project, projects) seed_seconds = time.perf_counter() - build_started index = SQLiteSearchIndex(root / "search.sqlite3", repository) # type: ignore[arg-type] cold_started = time.perf_counter() cold_result = index.search( query="interactive visitors", project_ids=(repository.project_ids[0],), include_shared=False, limit=20, ) cold_seconds = time.perf_counter() - cold_started timings: list[float] = [] for _ in range(samples): started = time.perf_counter() result = index.search( query="interactive visitors", project_ids=(repository.project_ids[0],), include_shared=False, limit=20, ) timings.append((time.perf_counter() - started) * 1000) if len(result.nodes) != 20: raise RuntimeError(f"Expected 20 results, received {len(result.nodes)}") total_nodes = nodes_per_project * projects return { "benchmark": "sqlite-search-adapter-synthetic-v1", "evidence_scope": ( "synthetic adapter benchmark; excludes HTTP/auth, Git object reads, " "and semantic relevance" ), "environment": { "platform": platform.platform(), "processor": processor_name(), "logical_cpus": os.cpu_count(), "python": platform.python_version(), "sqlite": sqlite3.sqlite_version, }, "workload": { "projects": projects, "nodes_per_project": nodes_per_project, "total_nodes": total_nodes, "query_project": repository.project_ids[0], "query": "interactive visitors", "page_size": 20, "warm_samples": samples, }, "measurements": { "synthetic_seed_seconds": round(seed_seconds, 3), "cold_index_rebuild_and_first_search_seconds": round(cold_seconds, 3), "warm_search_ms": { "p50": round(statistics.median(timings), 3), "p95": round(percentile(timings, 0.95), 3), "max": round(max(timings), 3), }, "first_result_count": len(cold_result.nodes), }, "limitations": [ "synthetic records reuse one valid claim/citation shape and are not a relevance corpus", "in-memory canonical reads omit Git filesystem and HTTP/API overhead", "the PRD hardware, cache state, concurrency, and percentile decisions are not fixed", ], } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--nodes-per-project", type=int, default=10_000) parser.add_argument("--projects", type=int, default=1) parser.add_argument("--samples", type=int, default=50) parser.add_argument("--output", type=Path) args = parser.parse_args() if min(args.nodes_per_project, args.projects, args.samples) < 1: parser.error("node, project, and sample counts must be positive") with tempfile.TemporaryDirectory(prefix="gewu-search-benchmark-") as temporary: report = run( args.nodes_per_project, args.projects, args.samples, Path(temporary), ) rendered = json.dumps(report, indent=2, sort_keys=True) + "\n" if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(rendered, encoding="utf-8") print(rendered, end="") if __name__ == "__main__": main()