From 4959f11d730099e8717bd195267e7a73b947379d Mon Sep 17 00:00:00 2001 From: unicornz-code Date: Sat, 5 Sep 2026 19:46:35 +0100 Subject: [PATCH 1/3] fix: avoid duplicate fresh-index embedding Port MinishLab/semble a79078007f21842eb8325c8217b3acd4d050d578's indexing correction and add benchmark-only cold-path phase instrumentation. --- benchmarks/README.md | 36 ++ benchmarks/profile_cold_index.py | 512 ++++++++++++++++++++ src/semble/index/create.py | 13 +- tests/benchmarks/__init__.py | 0 tests/benchmarks/test_profile_cold_index.py | 79 +++ tests/index/test_create.py | 16 + 6 files changed, 648 insertions(+), 8 deletions(-) create mode 100644 benchmarks/profile_cold_index.py create mode 100644 tests/benchmarks/__init__.py create mode 100644 tests/benchmarks/test_profile_cold_index.py diff --git a/benchmarks/README.md b/benchmarks/README.md index 72d365595..b5ada4b95 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -194,6 +194,42 @@ Writes to `benchmarks/results/speed-.json`. +
+Cold index phase profiler + +Profile fresh `create_index_from_path` calls without enabling production telemetry: + +```bash +uv run python -m benchmarks.profile_cold_index \ + --corpus-path /path/to/source-tree \ + --model-path /path/to/model \ + --repetitions 5 \ + --label experiment-name \ + --revision revision-id \ + --output /path/to/profile.json +``` + +The model is loaded before measurement; every repetition passes `previous=None`. `records` contains raw per-run +nanosecond timings, counts, call-size arrays, index shape/vector bytes, and process peak RSS. `summary` recursively +reports median/min/max for numeric fields. Phase `wall_ns` is inclusive; `exclusive_wall_ns` removes nested measured +boundaries, while `overlap.nested_wall_ns` names each parent/child overlap. Inclusive parent and child times must not +be summed. + +`total.wall_ns` and `total.process_cpu_ns` cover only `create_index_from_path`. For each phase, `calls` counts boundary +entries and `items` counts yielded files, checked files, reads, produced chunks, BM25 documents, or model texts as +appropriate. `call_sizes` and its count/total/median/min/max distribution are emitted for BM25 updates, `embed_chunks`, +`StaticModel.encode`, and Model2Vec tokenizer batches. `source_read_bytes` counts stat-observed bytes for every source +read. `index` reports files, chunks, dimensions, and vector bytes. `memory.peak_rss_bytes` is the process high-water RSS +and therefore includes the preloaded model; the before-index and increase fields provide context. + +`static_model_encode` is the exact `StaticModel.encode` API boundary, not pure native model time. +`model2vec_tokenization` is nested within it; `model2vec_lookup_mean_stack_normalization` is the derived remainder +(`encode - tokenize`) and includes Python overhead. `vector_backend_assembly_residual` is the derived root remainder +after named boundaries, so it also includes unwrapped index orchestration. The `counts` equality fields compare +embedded chunks and encoded texts with produced and unique chunk IDs, making an extra embedding pass visible. + +
+
Ablations diff --git a/benchmarks/profile_cold_index.py b/benchmarks/profile_cold_index.py new file mode 100644 index 000000000..243621059 --- /dev/null +++ b/benchmarks/profile_cold_index.py @@ -0,0 +1,512 @@ +from __future__ import annotations + +import argparse +import contextlib +import gc +import json +import platform +import statistics +import sys +import time +from collections.abc import Callable, Iterator, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime, timezone +from importlib import import_module +from importlib.metadata import version +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import semble.chunking.chunking as chunking_module +import semble.index.create as create_module +import semble.index.files as files_module +from semble.index.dense import load_model + +_PHASE_NAMES = ( + "file_walk_iterator", + "path_stat", + "file_status_checks", + "language_detection", + "source_reads", + "chunk_source", + "tree_sitter_chunking", + "fallback_line_chunking", + "bm25_replace_add_total", + "bm25_tokenization", + "embed_chunks", + "static_model_encode", + "model2vec_tokenization", +) + + +@dataclass +class _PhaseTotals: + """Aggregate measurements for one instrumented boundary.""" + + wall_ns: int = 0 + exclusive_wall_ns: int = 0 + calls: int = 0 + items: int = 0 + call_sizes: list[int] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + """Return a JSON-compatible phase record.""" + result: dict[str, Any] = { + "wall_ns": self.wall_ns, + "exclusive_wall_ns": self.exclusive_wall_ns, + "calls": self.calls, + "items": self.items, + } + if self.call_sizes: + result["call_sizes"] = list(self.call_sizes) + result["call_size_distribution"] = _number_summary(self.call_sizes) + return result + + +@dataclass +class _ActivePhase: + """One live timing frame used to remove nested intervals.""" + + name: str + started_ns: int + child_wall_ns: int = 0 + + +class _PhaseRecorder: + """Collect inclusive boundaries and an exact exclusive-time accounting.""" + + def __init__(self, clock: Callable[[], int] = time.perf_counter_ns) -> None: + """Create a recorder, optionally with a deterministic test clock.""" + self._clock = clock + self._phases = {name: _PhaseTotals() for name in ("total_index", *_PHASE_NAMES)} + self._stack: list[_ActivePhase] = [] + self.counters: dict[str, int] = {} + self.nested_wall_ns: dict[str, int] = {} + + @contextlib.contextmanager + def measure(self, name: str, *, items: int = 0, call_size: int | None = None) -> Iterator[None]: + """Measure one call and subtract nested instrumented boundaries from its exclusive time.""" + phase = self._phases.setdefault(name, _PhaseTotals()) + phase.calls += 1 + phase.items += items + if call_size is not None: + phase.call_sizes.append(call_size) + frame = _ActivePhase(name=name, started_ns=self._clock()) + self._stack.append(frame) + try: + yield + finally: + elapsed_ns = self._clock() - frame.started_ns + if self._stack.pop() is not frame: + raise RuntimeError("Profiler phase stack became unbalanced") + phase.wall_ns += elapsed_ns + phase.exclusive_wall_ns += elapsed_ns - frame.child_wall_ns + if self._stack: + parent = self._stack[-1] + parent.child_wall_ns += elapsed_ns + edge = f"{parent.name}>{name}" + self.nested_wall_ns[edge] = self.nested_wall_ns.get(edge, 0) + elapsed_ns + + def add_counter(self, name: str, value: int = 1) -> None: + """Add to a raw event counter.""" + self.counters[name] = self.counters.get(name, 0) + value + + def add_items(self, phase: str, value: int) -> None: + """Add output items whose count is known only after a boundary returns.""" + self._phases[phase].items += value + + def phase(self, name: str) -> _PhaseTotals: + """Return aggregate totals for a named phase.""" + return self._phases[name] + + def phase_records(self) -> dict[str, dict[str, Any]]: + """Return all non-root phase records in stable order.""" + return {name: self._phases[name].as_dict() for name in _PHASE_NAMES} + + +def _number_summary(values: Sequence[int | float]) -> dict[str, int | float]: + """Return count, total, median, minimum, and maximum for numeric values.""" + return { + "count": len(values), + "total": sum(values), + "median": float(statistics.median(values)), + "min": min(values), + "max": max(values), + } + + +def _summarize_mappings(records: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + """Recursively summarize numeric leaves shared by every mapping.""" + if not records: + return {} + shared_keys = set(records[0]) + for record in records[1:]: + shared_keys.intersection_update(record) + + result: dict[str, Any] = {} + for key in sorted(shared_keys): + values = [record[key] for record in records] + if all(isinstance(value, Mapping) for value in values): + nested = _summarize_mappings([value for value in values if isinstance(value, Mapping)]) + if nested: + result[key] = nested + elif all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in values): + result[key] = { + "median": float(statistics.median(values)), + "min": min(values), + "max": max(values), + } + return result + + +def _summarize_records(records: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + """Build median/min/max summaries without treating repetition numbers as measurements.""" + stripped = [{key: value for key, value in record.items() if key != "repetition"} for record in records] + return _summarize_mappings(stripped) + + +def _sequence_size(value: Sequence[str] | str) -> int: + """Return the number of texts represented by a model API argument.""" + return 1 if isinstance(value, str) else len(value) + + +@contextlib.contextmanager +def _instrument_cold_path(recorder: _PhaseRecorder, model: Any) -> Iterator[None]: + """Install benchmark-local wrappers around the production cold-index boundaries.""" + original_walk_files = create_module.walk_files + original_path_stat = Path.stat + original_get_file_status = create_module.get_file_status + original_detect_language = create_module.detect_language + original_read_file_text = files_module.read_file_text + original_chunk_source = create_module.chunk_source + original_tree_sitter_chunk = chunking_module.chunk + original_fallback_chunk = chunking_module.chunk_lines + original_reindex_file = create_module._reindex_file + original_bm25_tokenize = create_module.tokenize + original_embed_chunks = create_module.embed_chunks + original_model_encode = model.encode + original_model_tokenize = model.tokenize + file_sizes: dict[Path, int] = {} + + def profiled_walk_files(*args: Any, **kwargs: Any) -> Iterator[Path]: + """Measure only time spent advancing the lazy file iterator.""" + recorder.add_counter("file_walk_iterators") + iterator = iter(original_walk_files(*args, **kwargs)) + while True: + try: + with recorder.measure("file_walk_iterator"): + item = next(iterator) + except StopIteration: + return + recorder.add_items("file_walk_iterator", 1) + recorder.add_counter("walked_files") + yield item + + def profiled_path_stat(path: Path, *args: Any, **kwargs: Any) -> Any: + """Measure explicit Path.stat calls and retain their observed byte sizes.""" + with recorder.measure("path_stat", items=1): + result = original_path_stat(path, *args, **kwargs) + file_sizes[path] = int(result.st_size) + return result + + def profiled_get_file_status(*args: Any, **kwargs: Any) -> Any: + """Measure file eligibility checks, including nested stat and small-file reads.""" + with recorder.measure("file_status_checks", items=1): + result = original_get_file_status(*args, **kwargs) + recorder.add_counter(f"file_status_{result.value}") + return result + + def profiled_detect_language(*args: Any, **kwargs: Any) -> Any: + """Measure extension-based language detection.""" + with recorder.measure("language_detection", items=1): + result = original_detect_language(*args, **kwargs) + if result is not None: + recorder.add_counter("languages_detected") + return result + + def profiled_read_file_text(file_path: Path) -> str: + """Measure source reads and count bytes from the preceding production stat.""" + with recorder.measure("source_reads", items=1): + text = original_read_file_text(file_path) + byte_count = file_sizes.get(file_path) + if byte_count is None: + byte_count = len(text.encode("utf-8")) + recorder.add_counter("source_read_bytes", byte_count) + return text + + def profiled_chunk_source(*args: Any, **kwargs: Any) -> Any: + """Measure the full chunk-source boundary and its produced chunks.""" + with recorder.measure("chunk_source"): + result = original_chunk_source(*args, **kwargs) + recorder.add_items("chunk_source", len(result)) + recorder.add_counter("chunk_source_calls") + return result + + def profiled_tree_sitter_chunk(*args: Any, **kwargs: Any) -> Any: + """Measure tree-sitter chunking attempts before any fallback.""" + with recorder.measure("tree_sitter_chunking"): + result = original_tree_sitter_chunk(*args, **kwargs) + if result is not None: + recorder.add_items("tree_sitter_chunking", len(result)) + recorder.add_counter("tree_sitter_successes") + return result + + def profiled_fallback_chunk(*args: Any, **kwargs: Any) -> Any: + """Measure line-based fallback chunking.""" + with recorder.measure("fallback_line_chunking"): + result = original_fallback_chunk(*args, **kwargs) + recorder.add_items("fallback_line_chunking", len(result)) + return result + + def profiled_reindex_file( + bm25_index: Any, + indexed_path: str, + file_chunks: list[Any], + previous_entry: Any, + ) -> None: + """Measure BM25 replacement/addition around its nested tokenization.""" + size = len(file_chunks) + with recorder.measure("bm25_replace_add_total", items=size, call_size=size): + original_reindex_file(bm25_index, indexed_path, file_chunks, previous_entry) + recorder.add_counter("bm25_documents_added", size) + if previous_entry is not None: + recorder.add_counter("bm25_document_slots_removed", previous_entry.count) + + def profiled_bm25_tokenize(text: str) -> list[str]: + """Measure BM25 tokenization nested in replacement/addition.""" + with recorder.measure("bm25_tokenization", items=1): + result = original_bm25_tokenize(text) + recorder.add_counter("bm25_tokens", len(result)) + return result + + def profiled_embed_chunks(profiled_model: Any, chunks: list[Any]) -> Any: + """Measure the exact production embed_chunks boundary.""" + size = len(chunks) + with recorder.measure("embed_chunks", items=size, call_size=size): + return original_embed_chunks(profiled_model, chunks) + + def profiled_model_encode(sentences: Sequence[str] | str, *args: Any, **kwargs: Any) -> Any: + """Measure the exact StaticModel.encode API boundary.""" + size = _sequence_size(sentences) + with recorder.measure("static_model_encode", items=size, call_size=size): + return original_model_encode(sentences, *args, **kwargs) + + def profiled_model_tokenize(sentences: Sequence[str] | str, *args: Any, **kwargs: Any) -> Any: + """Measure Model2Vec tokenization nested inside StaticModel.encode.""" + size = _sequence_size(sentences) + with recorder.measure("model2vec_tokenization", items=size, call_size=size): + return original_model_tokenize(sentences, *args, **kwargs) + + with contextlib.ExitStack() as stack: + stack.enter_context(patch.object(create_module, "walk_files", profiled_walk_files)) + stack.enter_context(patch.object(Path, "stat", profiled_path_stat)) + stack.enter_context(patch.object(create_module, "get_file_status", profiled_get_file_status)) + stack.enter_context(patch.object(create_module, "detect_language", profiled_detect_language)) + stack.enter_context(patch.object(create_module, "read_file_text", profiled_read_file_text)) + stack.enter_context(patch.object(files_module, "read_file_text", profiled_read_file_text)) + stack.enter_context(patch.object(create_module, "chunk_source", profiled_chunk_source)) + stack.enter_context(patch.object(chunking_module, "chunk", profiled_tree_sitter_chunk)) + stack.enter_context(patch.object(chunking_module, "chunk_lines", profiled_fallback_chunk)) + stack.enter_context(patch.object(create_module, "_reindex_file", profiled_reindex_file)) + stack.enter_context(patch.object(create_module, "tokenize", profiled_bm25_tokenize)) + stack.enter_context(patch.object(create_module, "embed_chunks", profiled_embed_chunks)) + stack.enter_context(patch.object(model, "encode", profiled_model_encode)) + stack.enter_context(patch.object(model, "tokenize", profiled_model_tokenize)) + yield + + +def _peak_rss_bytes() -> int | None: + """Return the process high-water RSS in bytes when the platform exposes it.""" + try: + resource_module: Any = import_module("resource") + except ModuleNotFoundError: # pragma: no cover - resource is available on supported Unix platforms + return None + peak = int(resource_module.getrusage(resource_module.RUSAGE_SELF).ru_maxrss) + return peak if sys.platform == "darwin" else peak * 1024 + + +def _derived_timings(recorder: _PhaseRecorder) -> dict[str, dict[str, Any]]: + """Build explicitly labeled residuals from exclusive boundary accounting.""" + return { + "chunk_object_assembly_residual": { + "wall_ns": recorder.phase("chunk_source").exclusive_wall_ns, + "derived_from": "chunk_source inclusive minus nested tree-sitter/fallback boundaries", + }, + "bm25_postings_residual": { + "wall_ns": recorder.phase("bm25_replace_add_total").exclusive_wall_ns, + "derived_from": "BM25 replacement/addition inclusive minus nested BM25 tokenization", + }, + "embed_array_conversion_residual": { + "wall_ns": recorder.phase("embed_chunks").exclusive_wall_ns, + "derived_from": "embed_chunks inclusive minus nested StaticModel.encode", + }, + "model2vec_lookup_mean_stack_normalization": { + "wall_ns": recorder.phase("static_model_encode").exclusive_wall_ns, + "derived_from": "StaticModel.encode inclusive minus nested Model2Vec tokenization", + "native_only": False, + }, + "vector_backend_assembly_residual": { + "wall_ns": recorder.phase("total_index").exclusive_wall_ns, + "derived_from": "total index wall minus all directly nested instrumented boundaries", + "includes": "vector stacking, BM25 order/backend construction, and unwrapped index orchestration", + }, + } + + +def _embedding_counters( + produced_chunks: int, + unique_chunk_ids: int, + embedded_chunks: int, + encoded_texts: int, +) -> dict[str, int | bool]: + """Return raw embedding totals and explicit fresh-index equality evidence.""" + return { + "chunks": produced_chunks, + "unique_chunk_ids": unique_chunk_ids, + "embedded_chunks": embedded_chunks, + "encoded_texts": encoded_texts, + "embedded_chunks_equal_produced_chunks": embedded_chunks == produced_chunks, + "encoded_texts_equal_produced_chunks": encoded_texts == produced_chunks, + "encoded_texts_equal_unique_chunks": encoded_texts == unique_chunk_ids, + } + + +def _profile_once(corpus_path: Path, model: Any, repetition: int) -> dict[str, Any]: + """Profile one fresh create_index_from_path call with no previous index.""" + recorder = _PhaseRecorder() + peak_rss_before = _peak_rss_bytes() + with _instrument_cold_path(recorder, model): + with recorder.measure("total_index"): + cpu_started_ns = time.process_time_ns() + bm25_index, semantic_index, chunks, manifest = create_module.create_index_from_path( + corpus_path, + model, + display_root=corpus_path, + previous=None, + ) + process_cpu_ns = time.process_time_ns() - cpu_started_ns + peak_rss_bytes = _peak_rss_bytes() + + produced_chunks = len(chunks) + unique_chunk_ids = len(set(bm25_index.doc_order)) + embedded_chunks = recorder.phase("embed_chunks").items + encoded_texts = recorder.phase("static_model_encode").items + counters: dict[str, int | bool] = dict(sorted(recorder.counters.items())) + counters["files"] = len(manifest) + counters.update(_embedding_counters(produced_chunks, unique_chunk_ids, embedded_chunks, encoded_texts)) + + phases = recorder.phase_records() + exclusive_phase_wall_ns = {name: phase["exclusive_wall_ns"] for name, phase in phases.items()} + residual_ns = recorder.phase("total_index").exclusive_wall_ns + accounted_wall_ns = residual_ns + sum(exclusive_phase_wall_ns.values()) + total_wall_ns = recorder.phase("total_index").wall_ns + + memory: dict[str, int | None] = { + "peak_rss_before_index_bytes": peak_rss_before, + "peak_rss_bytes": peak_rss_bytes, + "peak_rss_increase_bytes": ( + None + if peak_rss_before is None or peak_rss_bytes is None + else max(0, peak_rss_bytes - peak_rss_before) + ), + } + vectors = semantic_index.vectors + return { + "repetition": repetition, + "total": { + "wall_ns": total_wall_ns, + "process_cpu_ns": process_cpu_ns, + }, + "phases": phases, + "derived": _derived_timings(recorder), + "counts": counters, + "index": { + "files": len(manifest), + "chunks": produced_chunks, + "dimensions": int(vectors.shape[1]), + "vector_bytes": int(vectors.nbytes), + }, + "memory": memory, + "overlap": { + "nested_wall_ns": dict(sorted(recorder.nested_wall_ns.items())), + "exclusive_phase_wall_ns": exclusive_phase_wall_ns, + "residual_wall_ns": residual_ns, + "accounted_wall_ns": accounted_wall_ns, + "accounting_difference_ns": total_wall_ns - accounted_wall_ns, + }, + } + + +def _positive_int(value: str) -> int: + """Parse a strictly positive command-line integer.""" + parsed = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be at least 1") + return parsed + + +def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + """Parse cold-index profiler arguments.""" + parser = argparse.ArgumentParser(description="Profile Semble create_index_from_path phase boundaries.") + parser.add_argument("--corpus-path", type=Path, required=True, help="Source tree to index.") + parser.add_argument("--model-path", required=True, help="Local model path or Model2Vec model identifier.") + parser.add_argument("--repetitions", type=_positive_int, default=5, help="Number of fresh index builds.") + parser.add_argument("--label", required=True, help="Free-form run label stored as metadata.") + parser.add_argument("--revision", required=True, help="Corpus or experiment revision stored as metadata.") + parser.add_argument("--output", type=Path, required=True, help="Destination JSON path.") + args = parser.parse_args(argv) + args.corpus_path = args.corpus_path.expanduser().resolve() + args.output = args.output.expanduser().resolve() + if not args.corpus_path.is_dir(): + parser.error(f"corpus path is not a directory: {args.corpus_path}") + return args + + +def main(argv: Sequence[str] | None = None) -> None: + """Run repeated fresh-index profiles and write raw records plus summaries.""" + args = _parse_args(argv) + model, resolved_model_path = load_model(args.model_path) + records: list[dict[str, Any]] = [] + for repetition in range(1, args.repetitions + 1): + records.append(_profile_once(args.corpus_path, model, repetition)) + gc.collect() + + payload = { + "schema_version": 1, + "benchmark": "semble-create-index-cold-profile", + "generated_at": datetime.now(timezone.utc).isoformat(), + "metadata": { + "label": args.label, + "revision": args.revision, + "corpus_path": str(args.corpus_path), + "model_path_requested": args.model_path, + "model_path_resolved": resolved_model_path, + "repetitions": args.repetitions, + "python_version": platform.python_version(), + "platform": platform.platform(), + "model2vec_version": version("model2vec"), + }, + "timing_semantics": { + "wall_unit": "nanoseconds", + "inclusive": "wall_ns includes nested instrumented boundaries and must not be summed with them", + "exclusive": "exclusive_wall_ns removes directly nested instrumented intervals and is non-overlapping", + "static_model_encode": "exact StaticModel.encode API wall boundary, not pure native model time", + "model2vec_residual": ( + "derived encode remainder after tokenization; includes lookup/mean/stack/" + "normalization and Python overhead" + ), + "vector_backend_residual": ( + "derived root remainder; includes assembly/backend work and unwrapped orchestration" + ), + "cold_index": "every repetition passes previous=None; model loading is outside the measured index boundary", + }, + "records": records, + "summary": _summarize_records(records), + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print(args.output) + + +if __name__ == "__main__": + main() diff --git a/src/semble/index/create.py b/src/semble/index/create.py index c2c8a80d5..54e390285 100644 --- a/src/semble/index/create.py +++ b/src/semble/index/create.py @@ -135,15 +135,12 @@ def create_index_from_path( if not chunks: raise ValueError(f"No supported files found under {path}.") - if previous is None: - embeddings = embed_chunks(model, chunks) + if previous is not None and _has_same_vector_layout(manifest, previous_manifest): + embeddings = previous.vectors + for vector_part, start, count in embedding_parts: + embeddings[start : start + count] = vector_parts[vector_part] else: - if _has_same_vector_layout(manifest, previous_manifest): - embeddings = previous.vectors - for vector_part, start, count in embedding_parts: - embeddings[start : start + count] = vector_parts[vector_part] - else: - embeddings = np.vstack(vector_parts) + embeddings = np.vstack(vector_parts) bm25_index.set_doc_order(chunk_ids) semantic_index = SelectableBasicBackend(embeddings, BasicArgs()) diff --git a/tests/benchmarks/__init__.py b/tests/benchmarks/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/benchmarks/test_profile_cold_index.py b/tests/benchmarks/test_profile_cold_index.py new file mode 100644 index 000000000..b80c63ede --- /dev/null +++ b/tests/benchmarks/test_profile_cold_index.py @@ -0,0 +1,79 @@ +from collections.abc import Iterator + +from benchmarks.profile_cold_index import _PhaseRecorder, _embedding_counters, _summarize_records + + +def _clock(values: list[int]) -> Iterator[int]: + yield from values + + +def test_phase_recorder_tracks_exclusive_time_counters_and_call_sizes() -> None: + """Nested time is removed once, while counters and call sizes remain raw.""" + ticks = _clock([0, 10, 40, 100]) + recorder = _PhaseRecorder(clock=lambda: next(ticks)) + + with recorder.measure("total_index"): + with recorder.measure("static_model_encode", items=3, call_size=3): + recorder.add_counter("encoded_texts", 3) + + total = recorder.phase("total_index") + encode = recorder.phase("static_model_encode") + assert total.wall_ns == 100 + assert total.exclusive_wall_ns == 70 + assert encode.wall_ns == encode.exclusive_wall_ns == 30 + assert encode.calls == 1 + assert encode.items == recorder.counters["encoded_texts"] == 3 + assert total.exclusive_wall_ns + encode.exclusive_wall_ns == total.wall_ns + assert encode.as_dict()["call_size_distribution"] == { + "count": 1, + "total": 3, + "median": 3.0, + "min": 3, + "max": 3, + } + assert recorder.nested_wall_ns == {"total_index>static_model_encode": 30} + + +def test_summary_reports_median_min_max_for_numeric_leaves_only() -> None: + """Per-repetition metadata, booleans, and raw arrays do not pollute arithmetic summaries.""" + records = [ + { + "repetition": 1, + "total": {"wall_ns": 30, "process_cpu_ns": 9}, + "counts": {"chunks": 3, "proof": True}, + "sizes": [1, 2], + }, + { + "repetition": 2, + "total": {"wall_ns": 10, "process_cpu_ns": 7}, + "counts": {"chunks": 1, "proof": True}, + "sizes": [2, 3], + }, + { + "repetition": 3, + "total": {"wall_ns": 20, "process_cpu_ns": 8}, + "counts": {"chunks": 2, "proof": True}, + "sizes": [3, 4], + }, + ] + + summary = _summarize_records(records) + + assert "repetition" not in summary + assert "sizes" not in summary + assert "proof" not in summary["counts"] + assert summary["total"]["wall_ns"] == {"median": 20.0, "min": 10, "max": 30} + assert summary["total"]["process_cpu_ns"] == {"median": 8.0, "min": 7, "max": 9} + assert summary["counts"]["chunks"] == {"median": 2.0, "min": 1, "max": 3} + + +def test_embedding_counters_make_duplicate_pass_visible() -> None: + """Equality evidence distinguishes one text per unique chunk from an extra pass.""" + exact = _embedding_counters(4, 4, 4, 4) + duplicate = _embedding_counters(4, 4, 8, 8) + + assert exact["embedded_chunks_equal_produced_chunks"] is True + assert exact["encoded_texts_equal_unique_chunks"] is True + assert duplicate["embedded_chunks_equal_produced_chunks"] is False + assert duplicate["encoded_texts_equal_produced_chunks"] is False + assert duplicate["encoded_texts_equal_unique_chunks"] is False diff --git a/tests/index/test_create.py b/tests/index/test_create.py index 59e22f575..d1b5806ff 100644 --- a/tests/index/test_create.py +++ b/tests/index/test_create.py @@ -22,6 +22,17 @@ def _write_files(root: Path, files: dict[str, str]) -> None: def test_incremental_reindex_reuses_updates_and_prunes(mock_model: Any, tmp_path: Path) -> None: """One incremental pass reuses unchanged vectors, re-embeds changes, and keeps BM25 slots current.""" + encoded_texts: list[str] = [] + encoded_vector_parts: list[np.ndarray] = [] + original_encode = mock_model.encode.side_effect + + def tracking_encode(texts: list[str], **kwargs: Any) -> np.ndarray: + vectors = original_encode(texts, **kwargs) + encoded_texts.extend(texts) + encoded_vector_parts.append(vectors.copy()) + return vectors + + mock_model.encode.side_effect = tracking_encode _write_files( tmp_path, { @@ -34,6 +45,11 @@ def test_incremental_reindex_reuses_updates_and_prunes(mock_model: Any, tmp_path bm25_before, semantic_before, chunks_before, manifest_before = create_index_from_path( tmp_path, mock_model, display_root=tmp_path ) + fresh_calls = list(mock_model.encode.call_args_list) + assert mock_model.encode.call_count == 4 # once per file, no second full pass + assert sum(len(call.args[0]) for call in fresh_calls) == len(chunks_before) + assert encoded_texts == [chunk.content for chunk in chunks_before] + np.testing.assert_array_equal(semantic_before.vectors, np.vstack(encoded_vector_parts)) a_entry = manifest_before["a.py"] b_entry = manifest_before["b.py"] a_vectors_before = semantic_before.vectors[a_entry.start : a_entry.end].copy() From 9ea79ae6ecea82d11336b28aa2690dd26accb035 Mon Sep 17 00:00:00 2001 From: unicornz-code Date: Sat, 5 Sep 2026 20:11:41 +0100 Subject: [PATCH 2/3] fix: correct cold profiler measurement boundaries Keep source/status timing scoped to production call sites, label instrumented totals and unattributed overhead honestly, aggregate embedding invariants, and cover zero-chunk fresh indexes. --- benchmarks/README.md | 38 +-- benchmarks/profile_cold_index.py | 284 +++++++++++--------- tests/benchmarks/test_profile_cold_index.py | 36 ++- tests/index/test_create.py | 25 ++ 4 files changed, 229 insertions(+), 154 deletions(-) diff --git a/benchmarks/README.md b/benchmarks/README.md index b5ada4b95..a21af5eda 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -209,24 +209,32 @@ uv run python -m benchmarks.profile_cold_index \ --output /path/to/profile.json ``` -The model is loaded before measurement; every repetition passes `previous=None`. `records` contains raw per-run -nanosecond timings, counts, call-size arrays, index shape/vector bytes, and process peak RSS. `summary` recursively -reports median/min/max for numeric fields. Phase `wall_ns` is inclusive; `exclusive_wall_ns` removes nested measured -boundaries, while `overlap.nested_wall_ns` names each parent/child overlap. Inclusive parent and child times must not -be summed. - -`total.wall_ns` and `total.process_cpu_ns` cover only `create_index_from_path`. For each phase, `calls` counts boundary -entries and `items` counts yielded files, checked files, reads, produced chunks, BM25 documents, or model texts as -appropriate. `call_sizes` and its count/total/median/min/max distribution are emitted for BM25 updates, `embed_chunks`, -`StaticModel.encode`, and Model2Vec tokenizer batches. `source_read_bytes` counts stat-observed bytes for every source -read. `index` reports files, chunks, dimensions, and vector bytes. `memory.peak_rss_bytes` is the process high-water RSS -and therefore includes the preloaded model; the before-index and increase fields provide context. +The model is loaded before measurement; every repetition passes `previous=None`. Cold means fresh index state, not a +fresh process or cold parser/filesystem caches. `records` contains raw per-run nanosecond timings, counts, call-size +arrays, index shape/vector bytes, and process peak RSS. `summary` recursively reports median/min/max for numeric fields +and `embedding_invariants` states whether each duplicate-pass check held across every repetition. + +`total.instrumented_wall_ns` and `total.instrumented_process_cpu_ns` surround `create_index_from_path`, including all +wrapper and recorder overhead. They must not be compared with an uninstrumented benchmark; measure that total +separately. Phase `wall_ns` is inclusive, `exclusive_wall_ns` removes nested measured boundaries, and +`overlap.nested_wall_ns` names each parent/child overlap. Inclusive parent and child times must not be summed. + +For each phase, `calls` counts boundary entries and `items` counts outputs appropriate to that boundary. +`file_walk_iterator` includes traversal work and its internal filesystem checks. `file_status_checks` includes file +eligibility stats and sub-128-byte emptiness probes; its `items` count is checked files, not individual stat calls. +`source_reads` covers only the subsequent production source read; +`counts.source_read_bytes` is the UTF-8 size of its returned text and is counted after the timer. Direct manifest mtime +stats are unattributed. `call_sizes` plus count/total/median/min/max are emitted for BM25 updates, `embed_chunks`, +`StaticModel.encode`, and Model2Vec tokenizer batches. `index` reports files, chunks, dimensions, and vector bytes. `static_model_encode` is the exact `StaticModel.encode` API boundary, not pure native model time. `model2vec_tokenization` is nested within it; `model2vec_lookup_mean_stack_normalization` is the derived remainder -(`encode - tokenize`) and includes Python overhead. `vector_backend_assembly_residual` is the derived root remainder -after named boundaries, so it also includes unwrapped index orchestration. The `counts` equality fields compare -embedded chunks and encoded texts with produced and unique chunk IDs, making an extra embedding pass visible. +(`encode - tokenize`) and includes Python overhead. `unattributed_including_profiler_overhead` is the root exclusive +remainder: unwrapped index orchestration, vector/BM25 finalization, manifest mtime stats, and profiler bookkeeping. + +`memory.process_peak_rss_bytes` is a monotonic process-lifetime high-water mark. It includes the preloaded model and +all earlier repetitions, so it is not a per-repetition allocation delta. The recorder is single-thread-only; a future +parallel indexer requires thread-local or synchronized instrumentation.
diff --git a/benchmarks/profile_cold_index.py b/benchmarks/profile_cold_index.py index 243621059..1db22d499 100644 --- a/benchmarks/profile_cold_index.py +++ b/benchmarks/profile_cold_index.py @@ -19,12 +19,10 @@ import semble.chunking.chunking as chunking_module import semble.index.create as create_module -import semble.index.files as files_module from semble.index.dense import load_model _PHASE_NAMES = ( "file_walk_iterator", - "path_stat", "file_status_checks", "language_detection", "source_reads", @@ -73,12 +71,12 @@ class _ActivePhase: class _PhaseRecorder: - """Collect inclusive boundaries and an exact exclusive-time accounting.""" + """Collect single-threaded inclusive boundaries and exclusive-time accounting.""" def __init__(self, clock: Callable[[], int] = time.perf_counter_ns) -> None: """Create a recorder, optionally with a deterministic test clock.""" self._clock = clock - self._phases = {name: _PhaseTotals() for name in ("total_index", *_PHASE_NAMES)} + self._phases = {name: _PhaseTotals() for name in ("instrumented_index", *_PHASE_NAMES)} self._stack: list[_ActivePhase] = [] self.counters: dict[str, int] = {} self.nested_wall_ns: dict[str, int] = {} @@ -160,9 +158,38 @@ def _summarize_mappings(records: Sequence[Mapping[str, Any]]) -> dict[str, Any]: def _summarize_records(records: Sequence[Mapping[str, Any]]) -> dict[str, Any]: - """Build median/min/max summaries without treating repetition numbers as measurements.""" + """Build numeric summaries plus duplicate-pass verdicts across all repetitions.""" stripped = [{key: value for key, value in record.items() if key != "repetition"} for record in records] - return _summarize_mappings(stripped) + summary = _summarize_mappings(stripped) + invariants = _summarize_embedding_invariants(records) + if invariants: + summary["embedding_invariants"] = invariants + return summary + + +def _summarize_embedding_invariants(records: Sequence[Mapping[str, Any]]) -> dict[str, dict[str, int | bool]]: + """Report whether each duplicate-pass invariant held in every repetition.""" + names = ( + "embedded_chunks_equal_produced_chunks", + "encoded_texts_equal_produced_chunks", + "encoded_texts_equal_unique_chunks", + ) + if not records or any( + not isinstance(record.get("counts"), Mapping) + or any(name not in record["counts"] for name in names) + for record in records + ): + return {} + + result: dict[str, dict[str, int | bool]] = {} + for name in names: + matches = sum(record["counts"][name] is True for record in records) + result[name] = { + "all_repetitions": matches == len(records), + "matching_repetitions": matches, + "repetitions": len(records), + } + return result def _sequence_size(value: Sequence[str] | str) -> int: @@ -170,95 +197,87 @@ def _sequence_size(value: Sequence[str] | str) -> int: return 1 if isinstance(value, str) else len(value) -@contextlib.contextmanager -def _instrument_cold_path(recorder: _PhaseRecorder, model: Any) -> Iterator[None]: - """Install benchmark-local wrappers around the production cold-index boundaries.""" - original_walk_files = create_module.walk_files - original_path_stat = Path.stat - original_get_file_status = create_module.get_file_status - original_detect_language = create_module.detect_language - original_read_file_text = files_module.read_file_text - original_chunk_source = create_module.chunk_source - original_tree_sitter_chunk = chunking_module.chunk - original_fallback_chunk = chunking_module.chunk_lines - original_reindex_file = create_module._reindex_file - original_bm25_tokenize = create_module.tokenize - original_embed_chunks = create_module.embed_chunks - original_model_encode = model.encode - original_model_tokenize = model.tokenize - file_sizes: dict[Path, int] = {} - - def profiled_walk_files(*args: Any, **kwargs: Any) -> Iterator[Path]: +class _ColdPathWrappers: + """Single-threaded benchmark wrappers for cold-index boundaries.""" + + def __init__(self, recorder: _PhaseRecorder, model: Any) -> None: + """Capture original callables before ExitStack applies replacements.""" + self._recorder = recorder + self._walk_files = create_module.walk_files + self._get_file_status = create_module.get_file_status + self._detect_language = create_module.detect_language + self._read_file_text = create_module.read_file_text + self._chunk_source = create_module.chunk_source + self._tree_sitter_chunk = chunking_module.chunk + self._fallback_chunk = chunking_module.chunk_lines + self._reindex_file = create_module._reindex_file + self._bm25_tokenize = create_module.tokenize + self._embed_chunks = create_module.embed_chunks + self._model_encode = model.encode + self._model_tokenize = model.tokenize + + def walk_files(self, *args: Any, **kwargs: Any) -> Iterator[Path]: """Measure only time spent advancing the lazy file iterator.""" - recorder.add_counter("file_walk_iterators") - iterator = iter(original_walk_files(*args, **kwargs)) + self._recorder.add_counter("file_walk_iterators") + iterator = iter(self._walk_files(*args, **kwargs)) while True: try: - with recorder.measure("file_walk_iterator"): + with self._recorder.measure("file_walk_iterator"): item = next(iterator) except StopIteration: return - recorder.add_items("file_walk_iterator", 1) - recorder.add_counter("walked_files") + self._recorder.add_items("file_walk_iterator", 1) + self._recorder.add_counter("walked_files") yield item - def profiled_path_stat(path: Path, *args: Any, **kwargs: Any) -> Any: - """Measure explicit Path.stat calls and retain their observed byte sizes.""" - with recorder.measure("path_stat", items=1): - result = original_path_stat(path, *args, **kwargs) - file_sizes[path] = int(result.st_size) - return result - - def profiled_get_file_status(*args: Any, **kwargs: Any) -> Any: - """Measure file eligibility checks, including nested stat and small-file reads.""" - with recorder.measure("file_status_checks", items=1): - result = original_get_file_status(*args, **kwargs) - recorder.add_counter(f"file_status_{result.value}") + def get_file_status(self, *args: Any, **kwargs: Any) -> Any: + """Measure eligibility checks, including their stats and small-file emptiness probes.""" + with self._recorder.measure("file_status_checks", items=1): + result = self._get_file_status(*args, **kwargs) + self._recorder.add_counter(f"file_status_{result.value}") return result - def profiled_detect_language(*args: Any, **kwargs: Any) -> Any: + def detect_language(self, *args: Any, **kwargs: Any) -> Any: """Measure extension-based language detection.""" - with recorder.measure("language_detection", items=1): - result = original_detect_language(*args, **kwargs) + with self._recorder.measure("language_detection", items=1): + result = self._detect_language(*args, **kwargs) if result is not None: - recorder.add_counter("languages_detected") + self._recorder.add_counter("languages_detected") return result - def profiled_read_file_text(file_path: Path) -> str: - """Measure source reads and count bytes from the preceding production stat.""" - with recorder.measure("source_reads", items=1): - text = original_read_file_text(file_path) - byte_count = file_sizes.get(file_path) - if byte_count is None: - byte_count = len(text.encode("utf-8")) - recorder.add_counter("source_read_bytes", byte_count) + def read_file_text(self, file_path: Path) -> str: + """Measure the production source read and count returned UTF-8 bytes afterward.""" + with self._recorder.measure("source_reads", items=1): + text = self._read_file_text(file_path) + self._recorder.add_counter("source_read_bytes", len(text.encode("utf-8"))) return text - def profiled_chunk_source(*args: Any, **kwargs: Any) -> Any: + def chunk_source(self, *args: Any, **kwargs: Any) -> Any: """Measure the full chunk-source boundary and its produced chunks.""" - with recorder.measure("chunk_source"): - result = original_chunk_source(*args, **kwargs) - recorder.add_items("chunk_source", len(result)) - recorder.add_counter("chunk_source_calls") + with self._recorder.measure("chunk_source"): + result = self._chunk_source(*args, **kwargs) + self._recorder.add_items("chunk_source", len(result)) + self._recorder.add_counter("chunk_source_calls") return result - def profiled_tree_sitter_chunk(*args: Any, **kwargs: Any) -> Any: + def tree_sitter_chunk(self, *args: Any, **kwargs: Any) -> Any: """Measure tree-sitter chunking attempts before any fallback.""" - with recorder.measure("tree_sitter_chunking"): - result = original_tree_sitter_chunk(*args, **kwargs) + with self._recorder.measure("tree_sitter_chunking"): + result = self._tree_sitter_chunk(*args, **kwargs) if result is not None: - recorder.add_items("tree_sitter_chunking", len(result)) - recorder.add_counter("tree_sitter_successes") + self._recorder.add_items("tree_sitter_chunking", len(result)) + self._recorder.add_counter("tree_sitter_successes") return result - def profiled_fallback_chunk(*args: Any, **kwargs: Any) -> Any: + def fallback_chunk(self, *args: Any, **kwargs: Any) -> Any: """Measure line-based fallback chunking.""" - with recorder.measure("fallback_line_chunking"): - result = original_fallback_chunk(*args, **kwargs) - recorder.add_items("fallback_line_chunking", len(result)) + with self._recorder.measure("fallback_line_chunking"): + result = self._fallback_chunk(*args, **kwargs) + self._recorder.add_items("fallback_line_chunking", len(result)) return result - def profiled_reindex_file( + def reindex_file( + self, bm25_index: Any, indexed_path: str, file_chunks: list[Any], @@ -266,52 +285,55 @@ def profiled_reindex_file( ) -> None: """Measure BM25 replacement/addition around its nested tokenization.""" size = len(file_chunks) - with recorder.measure("bm25_replace_add_total", items=size, call_size=size): - original_reindex_file(bm25_index, indexed_path, file_chunks, previous_entry) - recorder.add_counter("bm25_documents_added", size) + with self._recorder.measure("bm25_replace_add_total", items=size, call_size=size): + self._reindex_file(bm25_index, indexed_path, file_chunks, previous_entry) + self._recorder.add_counter("bm25_documents_added", size) if previous_entry is not None: - recorder.add_counter("bm25_document_slots_removed", previous_entry.count) + self._recorder.add_counter("bm25_document_slots_removed", previous_entry.count) - def profiled_bm25_tokenize(text: str) -> list[str]: + def bm25_tokenize(self, text: str) -> list[str]: """Measure BM25 tokenization nested in replacement/addition.""" - with recorder.measure("bm25_tokenization", items=1): - result = original_bm25_tokenize(text) - recorder.add_counter("bm25_tokens", len(result)) + with self._recorder.measure("bm25_tokenization", items=1): + result = self._bm25_tokenize(text) + self._recorder.add_counter("bm25_tokens", len(result)) return result - def profiled_embed_chunks(profiled_model: Any, chunks: list[Any]) -> Any: + def embed_chunks(self, profiled_model: Any, chunks: list[Any]) -> Any: """Measure the exact production embed_chunks boundary.""" size = len(chunks) - with recorder.measure("embed_chunks", items=size, call_size=size): - return original_embed_chunks(profiled_model, chunks) + with self._recorder.measure("embed_chunks", items=size, call_size=size): + return self._embed_chunks(profiled_model, chunks) - def profiled_model_encode(sentences: Sequence[str] | str, *args: Any, **kwargs: Any) -> Any: + def model_encode(self, sentences: Sequence[str] | str, *args: Any, **kwargs: Any) -> Any: """Measure the exact StaticModel.encode API boundary.""" size = _sequence_size(sentences) - with recorder.measure("static_model_encode", items=size, call_size=size): - return original_model_encode(sentences, *args, **kwargs) + with self._recorder.measure("static_model_encode", items=size, call_size=size): + return self._model_encode(sentences, *args, **kwargs) - def profiled_model_tokenize(sentences: Sequence[str] | str, *args: Any, **kwargs: Any) -> Any: + def model_tokenize(self, sentences: Sequence[str] | str, *args: Any, **kwargs: Any) -> Any: """Measure Model2Vec tokenization nested inside StaticModel.encode.""" size = _sequence_size(sentences) - with recorder.measure("model2vec_tokenization", items=size, call_size=size): - return original_model_tokenize(sentences, *args, **kwargs) + with self._recorder.measure("model2vec_tokenization", items=size, call_size=size): + return self._model_tokenize(sentences, *args, **kwargs) + +@contextlib.contextmanager +def _instrument_cold_path(recorder: _PhaseRecorder, model: Any) -> Iterator[None]: + """Install and exception-safely restore benchmark-local cold-path wrappers.""" + wrappers = _ColdPathWrappers(recorder, model) with contextlib.ExitStack() as stack: - stack.enter_context(patch.object(create_module, "walk_files", profiled_walk_files)) - stack.enter_context(patch.object(Path, "stat", profiled_path_stat)) - stack.enter_context(patch.object(create_module, "get_file_status", profiled_get_file_status)) - stack.enter_context(patch.object(create_module, "detect_language", profiled_detect_language)) - stack.enter_context(patch.object(create_module, "read_file_text", profiled_read_file_text)) - stack.enter_context(patch.object(files_module, "read_file_text", profiled_read_file_text)) - stack.enter_context(patch.object(create_module, "chunk_source", profiled_chunk_source)) - stack.enter_context(patch.object(chunking_module, "chunk", profiled_tree_sitter_chunk)) - stack.enter_context(patch.object(chunking_module, "chunk_lines", profiled_fallback_chunk)) - stack.enter_context(patch.object(create_module, "_reindex_file", profiled_reindex_file)) - stack.enter_context(patch.object(create_module, "tokenize", profiled_bm25_tokenize)) - stack.enter_context(patch.object(create_module, "embed_chunks", profiled_embed_chunks)) - stack.enter_context(patch.object(model, "encode", profiled_model_encode)) - stack.enter_context(patch.object(model, "tokenize", profiled_model_tokenize)) + stack.enter_context(patch.object(create_module, "walk_files", wrappers.walk_files)) + stack.enter_context(patch.object(create_module, "get_file_status", wrappers.get_file_status)) + stack.enter_context(patch.object(create_module, "detect_language", wrappers.detect_language)) + stack.enter_context(patch.object(create_module, "read_file_text", wrappers.read_file_text)) + stack.enter_context(patch.object(create_module, "chunk_source", wrappers.chunk_source)) + stack.enter_context(patch.object(chunking_module, "chunk", wrappers.tree_sitter_chunk)) + stack.enter_context(patch.object(chunking_module, "chunk_lines", wrappers.fallback_chunk)) + stack.enter_context(patch.object(create_module, "_reindex_file", wrappers.reindex_file)) + stack.enter_context(patch.object(create_module, "tokenize", wrappers.bm25_tokenize)) + stack.enter_context(patch.object(create_module, "embed_chunks", wrappers.embed_chunks)) + stack.enter_context(patch.object(model, "encode", wrappers.model_encode)) + stack.enter_context(patch.object(model, "tokenize", wrappers.model_tokenize)) yield @@ -345,10 +367,13 @@ def _derived_timings(recorder: _PhaseRecorder) -> dict[str, dict[str, Any]]: "derived_from": "StaticModel.encode inclusive minus nested Model2Vec tokenization", "native_only": False, }, - "vector_backend_assembly_residual": { - "wall_ns": recorder.phase("total_index").exclusive_wall_ns, - "derived_from": "total index wall minus all directly nested instrumented boundaries", - "includes": "vector stacking, BM25 order/backend construction, and unwrapped index orchestration", + "unattributed_including_profiler_overhead": { + "wall_ns": recorder.phase("instrumented_index").exclusive_wall_ns, + "derived_from": "instrumented index wall minus directly nested measured boundaries", + "includes": ( + "unwrapped index orchestration, vector/BM25 finalization, manifest mtime stats, " + "and profiler bookkeeping" + ), }, } @@ -374,9 +399,8 @@ def _embedding_counters( def _profile_once(corpus_path: Path, model: Any, repetition: int) -> dict[str, Any]: """Profile one fresh create_index_from_path call with no previous index.""" recorder = _PhaseRecorder() - peak_rss_before = _peak_rss_bytes() with _instrument_cold_path(recorder, model): - with recorder.measure("total_index"): + with recorder.measure("instrumented_index"): cpu_started_ns = time.process_time_ns() bm25_index, semantic_index, chunks, manifest = create_module.create_index_from_path( corpus_path, @@ -384,7 +408,7 @@ def _profile_once(corpus_path: Path, model: Any, repetition: int) -> dict[str, A display_root=corpus_path, previous=None, ) - process_cpu_ns = time.process_time_ns() - cpu_started_ns + instrumented_process_cpu_ns = time.process_time_ns() - cpu_started_ns peak_rss_bytes = _peak_rss_bytes() produced_chunks = len(chunks) @@ -397,25 +421,17 @@ def _profile_once(corpus_path: Path, model: Any, repetition: int) -> dict[str, A phases = recorder.phase_records() exclusive_phase_wall_ns = {name: phase["exclusive_wall_ns"] for name, phase in phases.items()} - residual_ns = recorder.phase("total_index").exclusive_wall_ns - accounted_wall_ns = residual_ns + sum(exclusive_phase_wall_ns.values()) - total_wall_ns = recorder.phase("total_index").wall_ns + instrumented_wall_ns = recorder.phase("instrumented_index").wall_ns memory: dict[str, int | None] = { - "peak_rss_before_index_bytes": peak_rss_before, - "peak_rss_bytes": peak_rss_bytes, - "peak_rss_increase_bytes": ( - None - if peak_rss_before is None or peak_rss_bytes is None - else max(0, peak_rss_bytes - peak_rss_before) - ), + "process_peak_rss_bytes": peak_rss_bytes, } vectors = semantic_index.vectors return { "repetition": repetition, "total": { - "wall_ns": total_wall_ns, - "process_cpu_ns": process_cpu_ns, + "instrumented_wall_ns": instrumented_wall_ns, + "instrumented_process_cpu_ns": instrumented_process_cpu_ns, }, "phases": phases, "derived": _derived_timings(recorder), @@ -430,9 +446,6 @@ def _profile_once(corpus_path: Path, model: Any, repetition: int) -> dict[str, A "overlap": { "nested_wall_ns": dict(sorted(recorder.nested_wall_ns.items())), "exclusive_phase_wall_ns": exclusive_phase_wall_ns, - "residual_wall_ns": residual_ns, - "accounted_wall_ns": accounted_wall_ns, - "accounting_difference_ns": total_wall_ns - accounted_wall_ns, }, } @@ -465,22 +478,21 @@ def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: def main(argv: Sequence[str] | None = None) -> None: """Run repeated fresh-index profiles and write raw records plus summaries.""" args = _parse_args(argv) - model, resolved_model_path = load_model(args.model_path) + model = load_model(args.model_path)[0] records: list[dict[str, Any]] = [] for repetition in range(1, args.repetitions + 1): records.append(_profile_once(args.corpus_path, model, repetition)) gc.collect() payload = { - "schema_version": 1, + "schema_version": 2, "benchmark": "semble-create-index-cold-profile", "generated_at": datetime.now(timezone.utc).isoformat(), "metadata": { "label": args.label, "revision": args.revision, "corpus_path": str(args.corpus_path), - "model_path_requested": args.model_path, - "model_path_resolved": resolved_model_path, + "model_path": args.model_path, "repetitions": args.repetitions, "python_version": platform.python_version(), "platform": platform.platform(), @@ -490,15 +502,29 @@ def main(argv: Sequence[str] | None = None) -> None: "wall_unit": "nanoseconds", "inclusive": "wall_ns includes nested instrumented boundaries and must not be summed with them", "exclusive": "exclusive_wall_ns removes directly nested instrumented intervals and is non-overlapping", + "instrumented_total": ( + "total instrumented fields include wrapper/recorder overhead and are not an uninstrumented benchmark" + ), "static_model_encode": "exact StaticModel.encode API wall boundary, not pure native model time", "model2vec_residual": ( "derived encode remainder after tokenization; includes lookup/mean/stack/" "normalization and Python overhead" ), - "vector_backend_residual": ( - "derived root remainder; includes assembly/backend work and unwrapped orchestration" + "unattributed": ( + "derived root remainder includes unwrapped index orchestration, vector/BM25 finalization, " + "manifest mtime stats, and profiler bookkeeping" + ), + "cold_index": ( + "every repetition passes previous=None; model loading is outside the boundary, while process, parser, " + "and filesystem caches may be warm" + ), + "process_peak_rss": ( + "monotonic process-lifetime high-water RSS; reflects earlier peaks including model load and repetitions" + ), + "single_thread_only": ( + "recorder state is unsynchronized; a parallel indexer requires thread-local or " + "synchronized instrumentation" ), - "cold_index": "every repetition passes previous=None; model loading is outside the measured index boundary", }, "records": records, "summary": _summarize_records(records), diff --git a/tests/benchmarks/test_profile_cold_index.py b/tests/benchmarks/test_profile_cold_index.py index b80c63ede..ef8c4b842 100644 --- a/tests/benchmarks/test_profile_cold_index.py +++ b/tests/benchmarks/test_profile_cold_index.py @@ -1,6 +1,10 @@ from collections.abc import Iterator -from benchmarks.profile_cold_index import _PhaseRecorder, _embedding_counters, _summarize_records +from benchmarks.profile_cold_index import ( + _PhaseRecorder, + _embedding_counters, + _summarize_records, +) def _clock(values: list[int]) -> Iterator[int]: @@ -12,18 +16,17 @@ def test_phase_recorder_tracks_exclusive_time_counters_and_call_sizes() -> None: ticks = _clock([0, 10, 40, 100]) recorder = _PhaseRecorder(clock=lambda: next(ticks)) - with recorder.measure("total_index"): + with recorder.measure("instrumented_index"): with recorder.measure("static_model_encode", items=3, call_size=3): recorder.add_counter("encoded_texts", 3) - total = recorder.phase("total_index") + total = recorder.phase("instrumented_index") encode = recorder.phase("static_model_encode") assert total.wall_ns == 100 assert total.exclusive_wall_ns == 70 assert encode.wall_ns == encode.exclusive_wall_ns == 30 assert encode.calls == 1 assert encode.items == recorder.counters["encoded_texts"] == 3 - assert total.exclusive_wall_ns + encode.exclusive_wall_ns == total.wall_ns assert encode.as_dict()["call_size_distribution"] == { "count": 1, "total": 3, @@ -31,7 +34,7 @@ def test_phase_recorder_tracks_exclusive_time_counters_and_call_sizes() -> None: "min": 3, "max": 3, } - assert recorder.nested_wall_ns == {"total_index>static_model_encode": 30} + assert recorder.nested_wall_ns == {"instrumented_index>static_model_encode": 30} def test_summary_reports_median_min_max_for_numeric_leaves_only() -> None: @@ -39,19 +42,19 @@ def test_summary_reports_median_min_max_for_numeric_leaves_only() -> None: records = [ { "repetition": 1, - "total": {"wall_ns": 30, "process_cpu_ns": 9}, + "total": {"instrumented_wall_ns": 30, "instrumented_process_cpu_ns": 9}, "counts": {"chunks": 3, "proof": True}, "sizes": [1, 2], }, { "repetition": 2, - "total": {"wall_ns": 10, "process_cpu_ns": 7}, + "total": {"instrumented_wall_ns": 10, "instrumented_process_cpu_ns": 7}, "counts": {"chunks": 1, "proof": True}, "sizes": [2, 3], }, { "repetition": 3, - "total": {"wall_ns": 20, "process_cpu_ns": 8}, + "total": {"instrumented_wall_ns": 20, "instrumented_process_cpu_ns": 8}, "counts": {"chunks": 2, "proof": True}, "sizes": [3, 4], }, @@ -62,8 +65,8 @@ def test_summary_reports_median_min_max_for_numeric_leaves_only() -> None: assert "repetition" not in summary assert "sizes" not in summary assert "proof" not in summary["counts"] - assert summary["total"]["wall_ns"] == {"median": 20.0, "min": 10, "max": 30} - assert summary["total"]["process_cpu_ns"] == {"median": 8.0, "min": 7, "max": 9} + assert summary["total"]["instrumented_wall_ns"] == {"median": 20.0, "min": 10, "max": 30} + assert summary["total"]["instrumented_process_cpu_ns"] == {"median": 8.0, "min": 7, "max": 9} assert summary["counts"]["chunks"] == {"median": 2.0, "min": 1, "max": 3} @@ -71,9 +74,22 @@ def test_embedding_counters_make_duplicate_pass_visible() -> None: """Equality evidence distinguishes one text per unique chunk from an extra pass.""" exact = _embedding_counters(4, 4, 4, 4) duplicate = _embedding_counters(4, 4, 8, 8) + summary = _summarize_records( + [ + {"repetition": 1, "counts": exact}, + {"repetition": 2, "counts": duplicate}, + ] + ) assert exact["embedded_chunks_equal_produced_chunks"] is True assert exact["encoded_texts_equal_unique_chunks"] is True assert duplicate["embedded_chunks_equal_produced_chunks"] is False assert duplicate["encoded_texts_equal_produced_chunks"] is False assert duplicate["encoded_texts_equal_unique_chunks"] is False + assert summary["embedding_invariants"]["embedded_chunks_equal_produced_chunks"] == { + "all_repetitions": False, + "matching_repetitions": 1, + "repetitions": 2, + } + assert summary["embedding_invariants"]["encoded_texts_equal_produced_chunks"]["all_repetitions"] is False + assert summary["embedding_invariants"]["encoded_texts_equal_unique_chunks"]["all_repetitions"] is False diff --git a/tests/index/test_create.py b/tests/index/test_create.py index d1b5806ff..bc04043e6 100644 --- a/tests/index/test_create.py +++ b/tests/index/test_create.py @@ -103,6 +103,31 @@ def tracking_encode(texts: list[str], **kwargs: Any) -> np.ndarray: assert set(bm25_after.doc_order) == expected_ids +def test_fresh_index_stacks_zero_chunk_vector_part(mock_model: Any, tmp_path: Path) -> None: + """A valid zero-chunk file keeps its empty vector part beside a nonempty file.""" + _write_files( + tmp_path, + { + "empty.py": " " * 128, + "live.py": "def live_value():\n return 1\n", + }, + ) + + bm25_index, semantic_index, chunks, manifest = create_index_from_path( + tmp_path, + mock_model, + display_root=tmp_path, + ) + + assert manifest["empty.py"].count == 0 + assert chunks + assert all(chunk.file_path == "live.py" for chunk in chunks) + assert semantic_index.vectors.shape == (len(chunks), mock_model.dim) + assert len(bm25_index.doc_order) == len(chunks) + assert mock_model.encode.call_count == 1 + assert mock_model.encode.call_args.args[0] == [chunk.content for chunk in chunks] + + def _build_valid_cache(index_path: Path, mock_model: Any) -> dict: """Build a real, well-formed on-disk index and return its metadata dict for mutation.""" src = index_path.parent / "src" From 4317a5bc579c307b14137c83d1ea6745fe47ba27 Mon Sep 17 00:00:00 2001 From: unicornz-code Date: Sat, 5 Sep 2026 20:18:46 +0100 Subject: [PATCH 3/3] Finalize cold-index profiler validation --- benchmarks/profile_cold_index.py | 3 +-- tests/benchmarks/test_profile_cold_index.py | 2 +- tests/test_git_workspace.py | 10 ++++++++-- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/benchmarks/profile_cold_index.py b/benchmarks/profile_cold_index.py index 1db22d499..6dc1f0cf7 100644 --- a/benchmarks/profile_cold_index.py +++ b/benchmarks/profile_cold_index.py @@ -175,8 +175,7 @@ def _summarize_embedding_invariants(records: Sequence[Mapping[str, Any]]) -> dic "encoded_texts_equal_unique_chunks", ) if not records or any( - not isinstance(record.get("counts"), Mapping) - or any(name not in record["counts"] for name in names) + not isinstance(record.get("counts"), Mapping) or any(name not in record["counts"] for name in names) for record in records ): return {} diff --git a/tests/benchmarks/test_profile_cold_index.py b/tests/benchmarks/test_profile_cold_index.py index ef8c4b842..c2698b88a 100644 --- a/tests/benchmarks/test_profile_cold_index.py +++ b/tests/benchmarks/test_profile_cold_index.py @@ -1,8 +1,8 @@ from collections.abc import Iterator from benchmarks.profile_cold_index import ( - _PhaseRecorder, _embedding_counters, + _PhaseRecorder, _summarize_records, ) diff --git a/tests/test_git_workspace.py b/tests/test_git_workspace.py index 26deeb462..274e8b417 100644 --- a/tests/test_git_workspace.py +++ b/tests/test_git_workspace.py @@ -163,11 +163,17 @@ async def test_open_git_workspace_requires_clean_base_and_indexes_existing_delta "new.py": ChangeKind.ADDED, } changed = session.index.search("workspace violet marker", scope=SearchScope.CHANGED) - assert changed.delta_results[0].result.chunk.file_path == "auth.py" + assert any( + result.result.chunk.file_path == "auth.py" and "workspace violet marker" in result.result.chunk.content + for result in changed.delta_results + ) (worktree / "auth.py").write_text("def authenticate():\n return 'immediate synchronized marker'\n") await session.synchronize() synchronized = session.index.search("immediate synchronized marker", scope=SearchScope.CHANGED) - assert synchronized.delta_results[0].result.chunk.file_path == "auth.py" + assert any( + result.result.chunk.file_path == "auth.py" and "immediate synchronized marker" in result.result.chunk.content + for result in synchronized.delta_results + ) assert registry.references(identity) == 1 await session.close() assert registry.references(identity) == 0