diff --git a/CHANGELOG.md b/CHANGELOG.md
index f1ba84f..289e3e5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,58 @@ All notable changes to HotMem will be documented in this file.
 
 Format follows [Keep a Changelog](https://keepachangelog.com/).
 
+## [Unreleased]
+
+### Changed — JSONL inspection validation policy (#89)
+- Inspection is **advisory** and now declares its assurance level:
+  `FileInspection.metadata["validation"]` is `sampled` (default — only the
+  declared sample window is parsed) or `full`. A malformed line beyond the
+  sampled window is reported only under `validation="full"`; `row_count`
+  semantics are unchanged. Full validation remains available via
+  `inspect_file(..., validation="full")` and `hotmem inspect
+  --full-validation`. Measured: 34 ms vs 115 ms per 11.6 MB file
+  (~5.5x at 100 MB per the spike baseline). Authoritative verification of
+  canonical content is unaffected — provenance checksums, not inspection.
+
+### Changed — streaming verification for large ranges (#88)
+- `provenance.verify_range` hashes ranges above `STREAM_VERIFY_THRESHOLD`
+  (8 MiB) while streaming through the new optional adapter capability
+  `LocalFilesystemAdapter.read_range_chunked` — O(chunk) memory instead of
+  O(range), identical digest and `ProvenanceError` semantics. Adapters
+  without the capability and ranges at/below the threshold keep the simple
+  single-read path.
+
+### Changed — single-read verified hydration (#87)
+- Verified hydration now hashes the bytes it already read instead of
+  re-reading the range through `provenance.verify_range` — one read per
+  range (spike B1: ~+16% at 100 MB). Digest and `ProvenanceError`
+  semantics are unchanged; new internal `provenance.verify_bytes` helper
+  carries the identical checksum contract.
+
+### Changed — embed_text trigram hashing (#90)
+- Trigram hashing now runs through a bounded per-gram cache
+  (`lru_cache`, 65 536 entries) replacing per-call md5 + hex parsing;
+  vectors are **bit-identical** to the previous `hotmem-hash-v1` output
+  (golden equivalence test over ASCII/unicode/random corpora). Measured
+  ~5.9x faster per embedding on realistic text — directly attacks the
+  37–78% of `parse_bundle` time the spike attributed to `embed_text`.
+
+### Changed — derived vector index polish (#92)
+- `search_by_ids` binds candidate ids in chunks of 900, so any configured
+  `oversample` stays under SQLite's legacy 999-variable cap; chunk results
+  are merged into the identical canonical order.
+- The accelerated search path runs a single FTS pass — candidate unioning
+  and BM25 normalization share one query (also true of the fallback path).
+- Swallowed Chroma delete/clear failures are now logged at warn level; the
+  derived index reuses `embed.unpack_embedding` for the blob format.
+
+### Performance follow-ups from the native helper spike (#87–#92)
+
+Work in progress — see PR for the unified acceptance criteria covering:
+single-read verified hydration (#87), streaming range hash (#88), JSONL
+inspection validation policy (#89), `embed_text` trigram batching (#90), and
+derived-vector-index polish (#92).
+
 ## [0.2.4] - 2026-08-28
 
 ### Added — Optional derived vector index (#49)
diff --git a/docs/architecture.md b/docs/architecture.md
index eb7ee09..8b42579 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -34,6 +34,14 @@ HotMem combines deterministic text embeddings, keyword overlap, and importance
 to rank local memories. Read-only inspectors provide lightweight metadata for
 CSV, JSONL, and Parquet files without turning the runtime into a query engine.
 
+Inspection is **advisory**: it never authorizes import, hydration, snapshot,
+or provenance decisions. JSONL validation is sampled by default (lines inside
+the declared sample window only) and the result declares its assurance level
+via `metadata.validation` (`sampled` | `full`); pass `--full-validation`
+(CLI) or `validation="full"` to parse every line. Authoritative verification
+of canonical memory content always happens through provenance checksums, not
+through inspection.
+
 An optional derived vector index can accelerate candidate retrieval. The index
 is disposable and rebuildable from SQLite, sits in front of the canonical
 hybrid ranker rather than replacing it, and search falls back to the
diff --git a/src/hotmem/cli.py b/src/hotmem/cli.py
index bf82053..fcb1f80 100644
--- a/src/hotmem/cli.py
+++ b/src/hotmem/cli.py
@@ -342,25 +342,39 @@ def openapi(output: str | None, fmt: str):
     type=int,
     help="Max sample rows to preview (CSV/JSONL).",
 )
+@click.option(
+    "--full-validation",
+    "full_validation",
+    is_flag=True,
+    help="JSONL: validate every line instead of only the sampled window. "
+    "Inspection is advisory; full validation costs ~5x on large files.",
+)
 @click.option(
     "--json",
     "as_json",
     is_flag=True,
     help="Emit raw JSON (bypasses the renderer, for scripting).",
 )
-def inspect(uri: str, count_rows: bool, sample_size: int, as_json: bool):
+def inspect(uri: str, count_rows: bool, sample_size: int, full_validation: bool, as_json: bool):
     """Inspect a local file's structure and provenance without ingesting it.
 
     Lightweight metadata-only inspection for CSV, JSONL, and Parquet files.
     Never copies file contents into the database — returns URI, size, checksum,
     columns, and an optional bounded sample. Unsupported formats and remote
-    schemes fail with a clear error.
+    schemes fail with a clear error. Inspection is advisory: JSONL validation
+    covers the sampled window unless --full-validation is passed, and the
+    result declares its assurance level.
     """
     from hotmem.inspectors import UnsupportedFormatError, inspect_file
     from hotmem.storage import UnsupportedSchemeError
 
     try:
-        inspection = inspect_file(uri, count_rows=count_rows, sample_size=sample_size)
+        inspection = inspect_file(
+            uri,
+            count_rows=count_rows,
+            sample_size=sample_size,
+            validation="full" if full_validation else "sampled",
+        )
     except UnsupportedFormatError as err:
         raise click.ClickException(str(err)) from err
     except UnsupportedSchemeError as err:
@@ -380,6 +394,7 @@ def inspect(uri: str, count_rows: bool, sample_size: int, as_json: bool):
         rows=data["row_count"],
         checksum=str(data["checksum"])[:12] + "…",
     )
+    click.echo(f"validation: {data['metadata'].get('validation', 'sampled')} (advisory)")
     if data["columns"]:
         click.echo(f"columns: {', '.join(data['columns'])}")
     if data["delimiter"]:
diff --git a/src/hotmem/db.py b/src/hotmem/db.py
index 78eabb7..a6815b3 100644
--- a/src/hotmem/db.py
+++ b/src/hotmem/db.py
@@ -34,6 +34,10 @@
 
 _trace = get_tracer("db")
 
+# Ids bound per query in search_by_ids: stays under SQLite's legacy
+# 999-variable limit even with the embedding parameter plus headroom (#92).
+_SEARCH_BIND_CHUNK = 900
+
 # Single source of truth for the memories table column order. Drives INSERT
 # statement generation and SQLite-to-SQLite import projection so the three
 # write paths cannot drift.
@@ -871,22 +875,39 @@ def search_by_ids(
         ``search_with_cosine``, so ids that expired, were archived, or were
         deleted since indexing are filtered out here. Used by the accelerated
         search path (#49) to re-score vector-index candidates in SQLite.
+
+        Ids are bound in chunks of ``_SEARCH_BIND_CHUNK`` so any configured
+        ``oversample`` stays under SQLite's legacy 999-variable cap; chunk
+        results are merged and re-sorted into the same canonical order.
         """
         if not memory_ids:
             return []
         archived_clause = "" if include_archived else " AND promotion_state != 'ARCHIVED'"
-        placeholders = ", ".join("?" for _ in memory_ids)
-        rows = self._conn.execute(
-            f"""SELECT id, identifier, fact_text, fact_summary, importance,
-                       metadata_json, source, created_at,
-                       cosine_sim(embedding, ?) AS cosine_score
-                FROM memories
-                WHERE id IN ({placeholders})
-                  AND {_ttl_live()}{archived_clause}
-                ORDER BY cosine_score DESC, id ASC""",
-            (query_embedding, *memory_ids),
-        ).fetchall()
-        return [dict(r) for r in rows]
+        results: list[dict[str, Any]] = []
+        for start in range(0, len(memory_ids), _SEARCH_BIND_CHUNK):
+            chunk = memory_ids[start : start + _SEARCH_BIND_CHUNK]
+            placeholders = ", ".join("?" for _ in chunk)
+            rows = self._conn.execute(
+                f"""SELECT id, identifier, fact_text, fact_summary, importance,
+                           metadata_json, source, created_at,
+                           cosine_sim(embedding, ?) AS cosine_score
+                    FROM memories
+                    WHERE id IN ({placeholders})
+                      AND {_ttl_live()}{archived_clause}""",
+                (query_embedding, *chunk),
+            ).fetchall()
+            results.extend(dict(r) for r in rows)
+        # Same total order as the per-chunk SQL ORDER BY: ids are unique, so
+        # (cosine DESC, id ASC) is deterministic. NULL cosine (no embedding)
+        # sorts last, matching SQLite's DESC default.
+        results.sort(
+            key=lambda r: (
+                r["cosine_score"] is None,
+                -(r["cosine_score"] or 0.0),
+                r["id"],
+            )
+        )
+        return results
 
     def all_rows(self, *, include_embedding: bool = False) -> list[dict[str, Any]]:
         """Return all memory rows as dicts (for snapshot export)."""
diff --git a/src/hotmem/embed.py b/src/hotmem/embed.py
index d2dd282..a2a93a2 100644
--- a/src/hotmem/embed.py
+++ b/src/hotmem/embed.py
@@ -21,6 +21,7 @@
 import hashlib
 import math
 import struct
+from functools import lru_cache
 
 from hotmem.trace import Timer, get_tracer
 
@@ -30,6 +31,18 @@
 EMBEDDING_MODEL = "hotmem-hash-v1"
 
 
+# Bounded cache of trigram -> (bucket, sign). Text reuses a small vocabulary
+# of character trigrams heavily (bundle corpora share word pools), so the
+# cache removes nearly all md5 calls while keeping vectors bit-identical:
+# same gram bytes -> same digest -> same bucket/sign (#90).
+@lru_cache(maxsize=65536)
+def _gram_bucket_sign(gram: str) -> tuple[int, float]:
+    h = int.from_bytes(hashlib.md5(gram.encode(), usedforsecurity=False).digest(), "big")
+    bucket = h % EMBEDDING_DIM
+    sign = 1.0 if (h >> 64) % 2 == 0 else -1.0
+    return bucket, sign
+
+
 def embed_text(text: str) -> list[float]:
     """Produce a deterministic embedding vector from text.
 
@@ -42,11 +55,7 @@ def embed_text(text: str) -> list[float]:
 
         # Hash overlapping trigrams into embedding buckets
         for i in range(max(1, len(text_lower) - 2)):
-            gram = text_lower[i : i + 3]
-            h = int(hashlib.md5(gram.encode(), usedforsecurity=False).hexdigest(), 16)
-            bucket = h % EMBEDDING_DIM
-            # Use upper bits for sign/magnitude
-            sign = 1.0 if (h >> 64) % 2 == 0 else -1.0
+            bucket, sign = _gram_bucket_sign(text_lower[i : i + 3])
             vec[bucket] += sign
 
         # L2 normalize
diff --git a/src/hotmem/inspectors/__init__.py b/src/hotmem/inspectors/__init__.py
index d4462fc..682fbc8 100644
--- a/src/hotmem/inspectors/__init__.py
+++ b/src/hotmem/inspectors/__init__.py
@@ -7,7 +7,7 @@
     contents into SQLite.
 
 Interface:
-    inspect_file(uri, *, count_rows=False, sample_size=5) -> FileInspection
+    inspect_file(uri, *, count_rows=False, sample_size=5, validation="sampled") -> FileInspection
     get_inspector(uri) -> FileInspector
     UnsupportedFormatError — raised when no inspector handles a format
 
@@ -78,6 +78,7 @@ def inspect_file(
     *,
     count_rows: bool = False,
     sample_size: int = 5,
+    validation: str = "sampled",
 ) -> FileInspection:
     """Inspect a backing file and return provenance + light metadata.
 
@@ -86,7 +87,15 @@ def inspect_file(
     raised for remote schemes; ``UnsupportedFormatError`` for unknown
     formats; otherwise a FileInspection (which may carry
     ``unsupported_reason`` for a recognized-but-malformed file).
+
+    ``validation`` is advisory-assurance control for JSONL (#89):
+    "sampled" (default) parses only the sample window and declares the
+    level via ``FileInspection.metadata["validation"]``; "full" parses
+    every line. Inspection is advisory — it never authorizes import,
+    hydration, snapshot, or provenance decisions.
     """
     adapter, meta = resolve_adapter(uri)
     inspector = _inspector_for_format(meta["format"])
-    return inspector.inspect(uri, adapter, meta, count_rows=count_rows, sample_size=sample_size)
+    return inspector.inspect(
+        uri, adapter, meta, count_rows=count_rows, sample_size=sample_size, validation=validation
+    )
diff --git a/src/hotmem/inspectors/base.py b/src/hotmem/inspectors/base.py
index 2c23118..2b0cf6b 100644
--- a/src/hotmem/inspectors/base.py
+++ b/src/hotmem/inspectors/base.py
@@ -81,6 +81,7 @@ def inspect(
         *,
         count_rows: bool = False,
         sample_size: int = 5,
+        validation: str = "sampled",  # JSONL-only; accepted for uniform dispatch
     ) -> FileInspection: ...
 
 
diff --git a/src/hotmem/inspectors/csv_inspector.py b/src/hotmem/inspectors/csv_inspector.py
index 4d6a12a..b587b0f 100644
--- a/src/hotmem/inspectors/csv_inspector.py
+++ b/src/hotmem/inspectors/csv_inspector.py
@@ -35,6 +35,7 @@ def inspect(
         *,
         count_rows: bool = False,
         sample_size: int = 5,
+        validation: str = "sampled",  # noqa: ARG002 — JSON validation is JSONL-only
     ) -> FileInspection:
         head = adapter.read_range(uri, 0, min(_SNIFF_BYTES, meta["size"] or 0))
         text = head.decode("utf-8", errors="replace")
diff --git a/src/hotmem/inspectors/jsonl_inspector.py b/src/hotmem/inspectors/jsonl_inspector.py
index 5fdbb6d..f01e387 100644
--- a/src/hotmem/inspectors/jsonl_inspector.py
+++ b/src/hotmem/inspectors/jsonl_inspector.py
@@ -5,6 +5,14 @@
     - Validate sampled lines are JSON; report the first malformed line offset
       in ``unsupported_reason`` instead of crashing (provenance, not outage).
     - O(file size) byte read, O(1) memory via buffered newline scanning.
+
+Validation policy (#89):
+    Inspection is ADVISORY — it never authorizes import, hydration, snapshot,
+    or provenance decisions. Default validation is "sampled": only lines
+    inside the ``sample_size`` window are parsed, and the result declares its
+    assurance via ``metadata["validation"]``. ``validation="full"`` keeps the
+    pre-#89 behavior (every line parsed) for consumers that need it.
+
 """
 
 from __future__ import annotations
@@ -18,6 +26,7 @@
 from .base import FileInspection
 
 _READ_CHUNK = 1 << 20  # 1 MiB read window — constant memory regardless of file size.
+_VALIDATION_MODES = ("sampled", "full")
 
 
 class JSONLInspector:
@@ -31,10 +40,15 @@ def inspect(
         *,
         count_rows: bool = False,
         sample_size: int = 5,
+        validation: str = "sampled",
     ) -> FileInspection:
+        if validation not in _VALIDATION_MODES:
+            raise ValueError(
+                f"unknown validation mode {validation!r}; expected one of {_VALIDATION_MODES}"
+            )
         path = _resolve(uri)
         row_count, sample_rows, byte_ranges, unsupported_reason = _stream(
-            path, count_rows=count_rows, sample_size=sample_size
+            path, count_rows=count_rows, sample_size=sample_size, validation=validation
         )
 
         columns = _infer_columns(sample_rows)
@@ -51,7 +65,7 @@ def inspect(
             has_header=None,
             sample=sample_rows or None,
             byte_ranges=byte_ranges or None,
-            metadata={},
+            metadata={"validation": validation},
             unsupported_reason=unsupported_reason,
         )
 
@@ -67,16 +81,21 @@ def _stream(
     *,
     count_rows: bool,
     sample_size: int,
+    validation: str = "sampled",
 ) -> tuple[int | None, list[dict[str, Any]], list[tuple[int, int]], str | None]:
     """One streaming pass: count lines, collect a bounded sample, validate JSON.
 
     Counts newlines in fixed-size chunks (cheap, allocation-free) and samples
     the first ``sample_size`` complete records by capturing byte offsets.
+
+    ``validation="sampled"`` parses only lines inside the sample window
+    (declared advisory assurance, #89); ``"full"`` parses every line.
     """
     row_count = 0 if count_rows else None
     sample_rows: list[dict[str, Any]] = []
     byte_ranges: list[tuple[int, int]] = []
     unsupported_reason: str | None = None
+    full_validation = validation == "full"
 
     line_index = 0
     line_start = 0
@@ -91,7 +110,7 @@ def _stream(
                 if carry:
                     if count_rows and carry.strip():
                         row_count = (row_count or 0) + 1
-                    if unsupported_reason is None:
+                    if unsupported_reason is None and (full_validation or line_index < sample_size):
                         unsupported_reason = _validate(carry, line_index, line_start)
                     _handle_line(
                         carry,
@@ -114,7 +133,7 @@ def _stream(
                 line_len = len(complete)
                 if count_rows and complete.strip():
                     row_count = (row_count or 0) + 1
-                if unsupported_reason is None:
+                if unsupported_reason is None and (full_validation or line_index < sample_size):
                     unsupported_reason = _validate(complete, line_index, line_start)
                 _handle_line(
                     complete,
diff --git a/src/hotmem/inspectors/parquet_inspector.py b/src/hotmem/inspectors/parquet_inspector.py
index 96f47ae..937fd6b 100644
--- a/src/hotmem/inspectors/parquet_inspector.py
+++ b/src/hotmem/inspectors/parquet_inspector.py
@@ -59,6 +59,7 @@ def inspect(
         *,
         count_rows: bool = False,  # noqa: ARG002 — num_rows comes from the footer
         sample_size: int = 0,  # noqa: ARG002 — no row sampling (metadata-only)
+        validation: str = "sampled",  # noqa: ARG002 — JSON validation is JSONL-only
     ) -> FileInspection:
         size = meta["size"]
         unsupported = self._validate_magic(adapter, uri, size)
diff --git a/src/hotmem/memory.py b/src/hotmem/memory.py
index 84c3771..d3a09a5 100644
--- a/src/hotmem/memory.py
+++ b/src/hotmem/memory.py
@@ -39,7 +39,7 @@
     BackingFileMissingError,
     ChecksumMismatchError,
     ProvenanceError,
-    verify_range,
+    verify_bytes,
 )
 from hotmem.storage import get_adapter
 from hotmem.trace import Timer, get_tracer
@@ -267,8 +267,10 @@ def hydrate_memory_detailed(
             raise ProvenanceError("truncated", source_uri, expected=expected_checksum)
 
         # On-demand checksum verification (skipped if no checksum stored or verify=False).
+        # Single-read verification (#87): hash the bytes already read — the
+        # digest and error semantics are identical to verify_range's re-read.
         if expected_checksum and verify:
-            verify_range(adapter, resolved_uri, offset, length, expected_checksum)
+            verify_bytes(source_uri, data, expected_checksum)
             verified = True
 
     _trace.info(
diff --git a/src/hotmem/provenance.py b/src/hotmem/provenance.py
index 2c9e82f..f4ee694 100644
--- a/src/hotmem/provenance.py
+++ b/src/hotmem/provenance.py
@@ -14,6 +14,7 @@
          .actual: str | None
      ChecksumMismatchError(ProvenanceError)
      BackingFileMissingError(ProvenanceError)
+     verify_bytes(source_uri, data, expected_checksum) -> None
      verify_range(adapter, source_uri, offset, length, expected_checksum) -> None
 
 Deps: none (stdlib only)
@@ -29,6 +30,11 @@
 
 _trace = get_tracer("provenance")
 
+# Ranges above this size are verified via chunked streaming (O(chunk) memory)
+# when the adapter exposes read_range_chunked; smaller ranges keep the simple
+# single-read path (#88). 8 MiB = 8x the inspector/embed read chunk.
+STREAM_VERIFY_THRESHOLD = 8 << 20
+
 Reason = Literal["checksum_mismatch", "missing_file", "truncated"]
 
 
@@ -76,6 +82,26 @@ def __init__(self, source_uri: str) -> None:
         super().__init__("missing_file", source_uri)
 
 
+def verify_bytes(source_uri: str, data: bytes, expected_checksum: str) -> None:
+    """Verify already-read bytes against the expected range SHA-256 (#87).
+
+    Raises ChecksumMismatchError on mismatch. Missing-file and truncated
+    conditions belong to the caller — it owns the read and can produce the
+    precise reason from read results.
+
+    The digest semantics are identical to ``verify_range``: SHA-256 over the
+    byte RANGE the caller materialized, never the whole file.
+    """
+    actual = hashlib.sha256(data).hexdigest()
+    if actual != expected_checksum:
+        _trace.warn(
+            "verify",
+            "checksum mismatch",
+            detail={"source_uri": source_uri, "expected": expected_checksum, "actual": actual},
+        )
+        raise ChecksumMismatchError(source_uri, expected=expected_checksum, actual=actual)
+
+
 def verify_range(
     adapter: Any,
     source_uri: str,
@@ -95,10 +121,20 @@ def verify_range(
     The checksum is computed as SHA-256 of the byte RANGE [offset, offset+length),
     NOT the whole file (main's adapter.checksum computes whole-file SHA-256,
     so we compute the range checksum ourselves via read_range + sha256).
+
+    Ranges above ``STREAM_VERIFY_THRESHOLD`` are hashed while streaming
+    through ``adapter.read_range_chunked`` (an optional adapter capability —
+    absent on other adapters or smaller ranges, the whole-range read runs
+    unchanged). The digest and every error are identical either way (#88).
     """
     if expected_checksum is None:
         return
 
+    chunked = getattr(adapter, "read_range_chunked", None)
+    if length > STREAM_VERIFY_THRESHOLD and callable(chunked):
+        _verify_streaming(source_uri, chunked, offset, length, expected_checksum)
+        return
+
     try:
         data = adapter.read_range(source_uri, offset, length)
     except FileNotFoundError as err:
@@ -113,7 +149,41 @@ def verify_range(
         )
         raise ProvenanceError("truncated", source_uri, expected=expected_checksum)
 
-    actual = hashlib.sha256(data).hexdigest()
+    verify_bytes(source_uri, data, expected_checksum)
+
+
+def _verify_streaming(
+    source_uri: str,
+    chunked: Any,
+    offset: int,
+    length: int,
+    expected_checksum: str,
+) -> None:
+    """Stream-hash a large range with O(chunk) memory; identical errors (#88)."""
+    h = hashlib.sha256()
+    received = 0
+    try:
+        stream = chunked(source_uri, offset, length)
+    except FileNotFoundError as err:
+        _trace.warn("verify", "missing backing file", detail={"source_uri": source_uri})
+        raise BackingFileMissingError(source_uri) from err
+    try:
+        for chunk in stream:
+            h.update(chunk)
+            received += len(chunk)
+    except FileNotFoundError as err:
+        _trace.warn("verify", "missing backing file", detail={"source_uri": source_uri})
+        raise BackingFileMissingError(source_uri) from err
+
+    if received < length:
+        _trace.warn(
+            "verify",
+            "truncated backing file",
+            detail={"source_uri": source_uri, "expected": length, "got": received},
+        )
+        raise ProvenanceError("truncated", source_uri, expected=expected_checksum)
+
+    actual = h.hexdigest()
     if actual != expected_checksum:
         _trace.warn(
             "verify",
diff --git a/src/hotmem/search.py b/src/hotmem/search.py
index f2b0444..f535d64 100644
--- a/src/hotmem/search.py
+++ b/src/hotmem/search.py
@@ -65,9 +65,9 @@ def _search_text(row: dict[str, Any]) -> str:
 
 def _fetch_candidates(
     db: MemoryDB,
-    query: str,
     query_vec: list[float],
     query_blob: bytes,
+    fts_rows: list[dict[str, Any]],
     include_archived: bool,
     vector_index: VectorIndex | None,
 ) -> list[dict[str, Any]]:
@@ -79,9 +79,10 @@ def _fetch_candidates(
 
     Accelerated (fresh index): the index supplies oversampled cosine
     candidate ids; those ids are re-fetched and re-scored in SQLite with the
-    same TTL-live/archived predicates, unioned with FTS match ids so text-only
-    matches are never lost. Ranking is recomputed downstream either way, so
-    both paths produce identical results.
+    same TTL-live/archived predicates, unioned with the FTS match ids
+    (already fetched once by the caller) so text-only matches are never
+    lost. Ranking is recomputed downstream either way, so both paths produce
+    identical results.
     """
     if (
         vector_index is not None
@@ -98,7 +99,6 @@ def _fetch_candidates(
             hits = []
         if hits:
             candidate_ids = [h["id"] for h in hits]
-            fts_rows = db.fts_search(query, include_archived=include_archived)
             candidate_ids += [r["id"] for r in fts_rows]
             # Dedupe, preserving order (index ranking first, FTS additions after).
             seen: set[str] = set()
@@ -137,10 +137,12 @@ def search_memories(
         query_vec = embed_text(query)
         query_blob = pack_embedding(query_vec)
 
+        # One FTS pass serves both candidate unioning and BM25 scoring (#92).
+        fts_rows = db.fts_search(query, include_archived=include_archived)
         candidates = _fetch_candidates(
-            db, query, query_vec, query_blob, include_archived, vector_index
+            db, query_vec, query_blob, fts_rows, include_archived, vector_index
         )
-        fts_scores = _normalize_bm25(db.fts_search(query, include_archived=include_archived))
+        fts_scores = _normalize_bm25(fts_rows)
 
         # Apply hybrid scoring
         scored = []
diff --git a/src/hotmem/server.py b/src/hotmem/server.py
index 90af51e..aa198ec 100644
--- a/src/hotmem/server.py
+++ b/src/hotmem/server.py
@@ -963,7 +963,7 @@ async def rebuild_index():
         """Full rebuild of the derived vector index from canonical storage.
 
         Reads only SQLite rows (metadata + embeddings) — never touches a
-        backing file.         Returns 400 ``vector_index_disabled`` when no backend
+        backing file. Returns 400 ``vector_index_disabled`` when no backend
         is configured (the default). This is an admin operation, never on
         the search hot path; stale indexes fall back to the SQLite scan.
         """
diff --git a/src/hotmem/storage/local.py b/src/hotmem/storage/local.py
index 7c38155..ad55a25 100644
--- a/src/hotmem/storage/local.py
+++ b/src/hotmem/storage/local.py
@@ -10,6 +10,7 @@
 
 import hashlib
 import mmap
+from collections.abc import Iterator
 from functools import lru_cache
 from pathlib import Path
 
@@ -54,6 +55,31 @@ def read_range(self, uri: str, offset: int, length: int) -> bytes:
             f.seek(offset)
             return f.read(length)
 
+    def read_range_chunked(
+        self, uri: str, offset: int, length: int, chunk_size: int = 1 << 20
+    ) -> Iterator[bytes]:
+        """Yield [offset, offset+length) in bounded chunks (O(chunk) memory).
+
+        Optional capability consumed by provenance.verify_range for
+        large-range streaming verification (#88); adapters without it fall
+        back to whole-range reads. Raises the same FileNotFoundError as
+        read_range; a short stream is the caller's truncation signal.
+        """
+        if offset < 0:
+            raise ValueError(f"offset must be non-negative, got {offset}")
+        if length < 0:
+            raise ValueError(f"length must be non-negative, got {length}")
+        path = _to_path(uri)
+        remaining = length
+        with open(path, "rb") as f:
+            f.seek(offset)
+            while remaining > 0:
+                chunk = f.read(min(chunk_size, remaining))
+                if not chunk:
+                    return
+                remaining -= len(chunk)
+                yield chunk
+
     def exists(self, uri: str) -> bool:
         return _to_path(uri).exists()
 
diff --git a/src/hotmem/vector_index.py b/src/hotmem/vector_index.py
index ed9ba9b..fa01013 100644
--- a/src/hotmem/vector_index.py
+++ b/src/hotmem/vector_index.py
@@ -32,6 +32,7 @@
 from pathlib import Path
 from typing import Any, Protocol, runtime_checkable
 
+from hotmem.embed import unpack_embedding
 from hotmem.trace import Timer, get_tracer
 
 _trace = get_tracer("vector_index")
@@ -282,8 +283,14 @@ def search(self, query_embedding: list[float], top_k: int) -> list[dict[str, Any
     def delete(self, memory_ids: list[str]) -> None:
         if not memory_ids:
             return
-        with _suppress_exception(Exception):
+        try:
             self._collection.delete(ids=list(memory_ids))
+        except Exception as err:  # derived index; deletion failure is non-fatal
+            _trace.warn(
+                "delete",
+                "chroma delete failed; index stays disposable and stale",
+                detail={"error": str(err)},
+            )
 
     def count(self) -> int:
         return int(self._collection.count())
@@ -310,8 +317,14 @@ def status(self, db: Any) -> dict[str, Any]:
         }
 
     def clear(self) -> None:
-        with _suppress_exception(Exception):
+        try:
             self._client.delete_collection("hotmem_memories")
+        except Exception as err:  # missing collection is the normal clear case
+            _trace.warn(
+                "clear",
+                "chroma collection delete failed; recreating",
+                detail={"error": str(err)},
+            )
         self._collection = self._client.get_or_create_collection(
             name="hotmem_memories",
             metadata={"hnsw:space": "cosine"},
@@ -343,19 +356,6 @@ def __exit__(self, exc_type, exc, tb) -> bool:
         return exc_type is not None and issubclass(exc_type, OSError)
 
 
-class _suppress_exception:
-    """Context manager that swallows a given exception type."""
-
-    def __init__(self, exc_type: type[BaseException]) -> None:
-        self.exc_type = exc_type
-
-    def __enter__(self) -> _suppress_exception:
-        return self
-
-    def __exit__(self, exc_type, exc, tb) -> bool:
-        return exc_type is not None and issubclass(exc_type, self.exc_type)
-
-
 def chroma_available() -> bool:
     """True when the optional chromadb package is importable."""
     try:
@@ -447,7 +447,7 @@ def rebuild_vector_index(
             blob = row.get("embedding")
             if not blob:
                 continue
-            embedding = _unpack_blob(blob)
+            embedding = unpack_embedding(blob)
             if not embedding:
                 continue
             records.append(
@@ -490,15 +490,6 @@ def rebuild_vector_index(
     return result
 
 
-def _unpack_blob(blob: bytes) -> list[float]:
-    import struct
-
-    count = len(blob) // 4
-    if count == 0:
-        return []
-    return list(struct.unpack(f"{count}f", blob))
-
-
 def remove_index_directory(base_dir: str | Path) -> bool:
     """Delete the entire derived index directory (files only, never SQLite).
 
diff --git a/tests/spy.py b/tests/spy.py
index 78174a8..eeaadf2 100644
--- a/tests/spy.py
+++ b/tests/spy.py
@@ -15,6 +15,7 @@ class SpyAdapter:
     def __init__(self, inner: Any) -> None:
         self.inner = inner
         self.read_range_calls = 0
+        self.read_range_chunked_calls = 0
         self.checksum_calls = 0
         self.exists_calls = 0
         self.read_calls = 0
@@ -23,6 +24,10 @@ def read_range(self, uri: str, offset: int, length: int) -> bytes:
         self.read_range_calls += 1
         return self.inner.read_range(uri, offset, length)
 
+    def read_range_chunked(self, uri: str, offset: int, length: int, chunk_size: int = 1 << 20):
+        self.read_range_chunked_calls += 1
+        return self.inner.read_range_chunked(uri, offset, length, chunk_size)
+
     def read(self, uri: str) -> bytes:
         self.read_calls += 1
         return self.inner.read(uri)
@@ -41,4 +46,9 @@ def metadata(self, uri: str) -> Any:
     @property
     def total_file_reads(self) -> int:
         """Count of methods that open/read the backing file (excludes exists)."""
-        return self.read_range_calls + self.checksum_calls + self.read_calls
+        return (
+            self.read_range_calls
+            + self.read_range_chunked_calls
+            + self.checksum_calls
+            + self.read_calls
+        )
diff --git a/tests/test_embed.py b/tests/test_embed.py
index 443b9d9..4671db8 100644
--- a/tests/test_embed.py
+++ b/tests/test_embed.py
@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import hashlib
 import math
+import random
 
 from hotmem.embed import EMBEDDING_DIM, embed_text, pack_embedding, unpack_embedding
 
@@ -50,3 +52,54 @@ def test_pack_unpack_roundtrip():
     recovered = unpack_embedding(blob)
     for a, b in zip(vec, recovered, strict=True):
         assert abs(a - b) < 1e-6
+
+
+def _reference_embed(text: str) -> list[float]:
+    """The pre-#90 algorithm, verbatim — the bit-compatibility oracle."""
+    vec = [0.0] * EMBEDDING_DIM
+    text_lower = text.lower()
+    for i in range(max(1, len(text_lower) - 2)):
+        gram = text_lower[i : i + 3]
+        h = int(hashlib.md5(gram.encode(), usedforsecurity=False).hexdigest(), 16)
+        bucket = h % EMBEDDING_DIM
+        sign = 1.0 if (h >> 64) % 2 == 0 else -1.0
+        vec[bucket] += sign
+    norm = math.sqrt(sum(x * x for x in vec))
+    return [x / norm for x in vec] if norm > 0 else vec
+
+
+def test_embed_text_bit_exact_against_reference():
+    """#90: the cached implementation must produce bit-identical vectors —
+    hotmem-hash-v1 is a compatibility contract, not an implementation detail."""
+    cases = [
+        "",
+        "a",
+        "ab",
+        "abc",
+        "hello world",
+        "Héllo Wörld — über café",
+        "日本語のテキストと絵文字🎉",
+        "invoice validation rules for vendor x",
+        "  mixed\tCASE and   spacing  ",
+    ]
+    rng = random.Random(90)
+    alphabet = "abcdefghijklmnopqrstuvwxyz "
+    cases += ["".join(rng.choice(alphabet) for _ in range(rng.randint(0, 300))) for _ in range(100)]
+    unicode_alphabet = "aäöü日本語🎉éè "
+    cases += [
+        "".join(rng.choice(unicode_alphabet) for _ in range(rng.randint(0, 120))) for _ in range(50)
+    ]
+    for text in cases:
+        assert embed_text(text) == _reference_embed(text), f"vector drifted for {text!r}"
+
+
+def test_embed_text_trigram_cache_engages():
+    """#90: repeated trigrams hit the cache instead of re-hashing md5."""
+    from hotmem import embed
+
+    text = "invoice validation rules " * 100
+    embed._gram_bucket_sign.cache_clear()
+    embed.embed_text(text)
+    info = embed._gram_bucket_sign.cache_info()
+    assert info.hits > 0
+    assert info.misses < info.hits  # repeated vocabulary dominates on real text
diff --git a/tests/test_file_backed.py b/tests/test_file_backed.py
index 89c8f2d..1d1b468 100644
--- a/tests/test_file_backed.py
+++ b/tests/test_file_backed.py
@@ -20,7 +20,12 @@
 
 from hotmem.db import MemoryDB
 from hotmem.memory import FileRef, add_file_backed, get_memory_metadata, hydrate_memory
-from hotmem.provenance import ProvenanceError
+from hotmem.provenance import (
+    STREAM_VERIFY_THRESHOLD,
+    ChecksumMismatchError,
+    ProvenanceError,
+    verify_range,
+)
 from hotmem.storage.local import LocalFilesystemAdapter
 
 # ── 1. add file-backed -> hydrate returns exact byte range ───────────────────
@@ -107,6 +112,38 @@ def test_checksum_mismatch_raises_http_409(app_client: TestClient):
     assert body["error"] == "provenance_mismatch"
 
 
+def test_verify_range_streams_large_ranges(tmp_path: Path):
+    """#88: ranges above STREAM_VERIFY_THRESHOLD hash while streaming —
+    same digest, identical mismatch/truncation errors, O(chunk) memory;
+    small ranges keep the simple single-read path."""
+    from spy import SpyAdapter
+
+    adapter = SpyAdapter(LocalFilesystemAdapter())
+    data = bytes(range(256)) * ((STREAM_VERIFY_THRESHOLD + 4096) // 256)
+    big = tmp_path / "big.bin"
+    big.write_bytes(data)
+
+    ok = hashlib.sha256(data[100 : 100 + STREAM_VERIFY_THRESHOLD + 1]).hexdigest()
+    verify_range(adapter, str(big), 100, STREAM_VERIFY_THRESHOLD + 1, ok)
+    assert adapter.read_range_chunked_calls == 1
+    assert adapter.read_range_calls == 0
+
+    # At exactly the threshold: still the simple single-read path (boundary).
+    ok_at = hashlib.sha256(data[100 : 100 + STREAM_VERIFY_THRESHOLD]).hexdigest()
+    verify_range(adapter, str(big), 100, STREAM_VERIFY_THRESHOLD, ok_at)
+    assert adapter.read_range_calls == 1
+
+    ok_small = hashlib.sha256(data[100:132]).hexdigest()
+    verify_range(adapter, str(big), 100, 32, ok_small)
+    assert adapter.read_range_calls == 2
+
+    with pytest.raises(ChecksumMismatchError):
+        verify_range(adapter, str(big), 100, STREAM_VERIFY_THRESHOLD, "0" * 64)
+    with pytest.raises(ProvenanceError) as exc:
+        verify_range(adapter, str(big), 0, len(data) + 1, "0" * 64)
+    assert exc.value.reason == "truncated"
+
+
 def test_missing_file_raises_provenance_error(tmp_db: MemoryDB, fixture_file: Path):
     expected = hashlib.sha256(fixture_file.read_bytes()[0:20]).hexdigest()
     ref = FileRef(
@@ -148,6 +185,34 @@ def test_missing_file_raises_http_409(app_client: TestClient):
 # ── 3. metadata access performs no file read (spy on adapter) ────────────────
 
 
+def test_verified_hydrate_reads_range_once(tmp_db: MemoryDB, fixture_file: Path):
+    """#87: verified hydration reads the range exactly once — the checksum is
+    computed over the already-read bytes instead of a second read."""
+    from spy import SpyAdapter
+
+    spy = SpyAdapter(LocalFilesystemAdapter())
+    import hotmem.memory as mem_mod
+
+    orig = mem_mod.get_adapter
+    mem_mod.get_adapter = lambda uri: spy
+    try:
+        expected = hashlib.sha256(fixture_file.read_bytes()[10:30]).hexdigest()
+        ref = FileRef(
+            source_uri=str(fixture_file),
+            byte_offset=10,
+            byte_length=20,
+            source_format="bin",
+            source_checksum=expected,
+        )
+        mid, _ = add_file_backed(tmp_db, identifier="ds", file_ref=ref, summary="v")
+
+        content = hydrate_memory(tmp_db, mid)
+        assert content == fixture_file.read_bytes()[10:30]
+        assert spy.read_range_calls == 1, "verified hydration must not re-read the range"
+    finally:
+        mem_mod.get_adapter = orig
+
+
 def test_metadata_access_no_file_read(tmp_db: MemoryDB, fixture_file: Path):
     from spy import SpyAdapter
 
diff --git a/tests/test_inspectors.py b/tests/test_inspectors.py
index 59a89d5..0f83720 100644
--- a/tests/test_inspectors.py
+++ b/tests/test_inspectors.py
@@ -149,6 +149,48 @@ def test_jsonl_chunk_spanning_line_offsets_are_exact(tmp_path):
     assert insp.byte_ranges == expected
 
 
+def test_jsonl_sampled_validation_reports_window_and_declares_assurance(tmp_path):
+    """#89: a malformed line inside the sample window is reported in both
+    modes; the result always declares its assurance level via metadata."""
+    path = tmp_path / "window.jsonl"
+    path.write_text('{"a": 1}\n{broken}\n{"a": 3}\n')
+
+    sampled = inspect_file(str(path), count_rows=True, sample_size=5)
+    assert sampled.unsupported_reason is not None
+    assert "line 1" in sampled.unsupported_reason
+    assert sampled.metadata["validation"] == "sampled"
+
+    full = inspect_file(str(path), count_rows=True, sample_size=5, validation="full")
+    assert full.unsupported_reason == sampled.unsupported_reason
+    assert full.metadata["validation"] == "full"
+
+
+def test_jsonl_sampled_validation_skips_lines_beyond_window(tmp_path):
+    """#89: a malformed line beyond the sampled window is only reported with
+    validation='full' — sampled inspection declares reduced assurance, and
+    row_count (validation-independent) still counts every non-blank line."""
+    path = tmp_path / "late.jsonl"
+    lines = [json.dumps({"i": i}) for i in range(5)] + ["{broken late}", json.dumps({"i": 9})]
+    path.write_text("\n".join(lines) + "\n")
+
+    sampled = inspect_file(str(path), count_rows=True, sample_size=5)
+    assert sampled.unsupported_reason is None
+    assert sampled.metadata["validation"] == "sampled"
+    assert sampled.row_count == 7
+
+    full = inspect_file(str(path), count_rows=True, sample_size=5, validation="full")
+    assert full.unsupported_reason is not None
+    assert "line 5" in full.unsupported_reason
+    assert full.metadata["validation"] == "full"
+
+
+def test_jsonl_validation_mode_is_validated(tmp_path):
+    path = tmp_path / "v.jsonl"
+    path.write_text('{"a": 1}\n')
+    with pytest.raises(ValueError):
+        inspect_file(str(path), validation="bogus")
+
+
 def test_jsonl_handles_no_trailing_newline(tmp_path):
     path = tmp_path / "notrail.jsonl"
     path.write_text('{"a": 1}\n{"a": 2}')  # no trailing newline
diff --git a/tests/test_vector_index.py b/tests/test_vector_index.py
index ccf21c8..0ede945 100644
--- a/tests/test_vector_index.py
+++ b/tests/test_vector_index.py
@@ -259,6 +259,44 @@ def test_rebuild_parity_includes_archived_when_requested(tmp_db: MemoryDB):
     assert "arch" in [m[0] for m in _messages(accelerated)]
 
 
+def test_search_by_ids_chunking_beyond_legacy_bind_limit(tmp_db: MemoryDB):
+    """#92: more candidate ids than SQLite's legacy 999-variable cap must
+    still work and stay order-identical to the deterministic full scan."""
+    ids = [f"m{i:04d}" for i in range(1100)]
+    for i, mid in enumerate(ids):
+        _add_fact(tmp_db, mid, f"chunk bind fact {i} invoice-{i % 7}")
+    query_vec = embed_text("invoice")
+    query_blob = pack_embedding(query_vec)
+
+    by_ids = tmp_db.search_by_ids(query_blob, ids)
+    full_scan = tmp_db.search_with_cosine(query_blob)
+    assert [r["id"] for r in by_ids] == [r["id"] for r in full_scan]
+    assert len(by_ids) == 1100
+
+
+def test_accelerated_search_runs_single_fts_pass(tmp_db: MemoryDB, monkeypatch: pytest.MonkeyPatch):
+    """#92: candidate unioning and BM25 normalization share one FTS query."""
+    _seed_store(tmp_db)
+    index = FakeVectorIndex()
+    rebuild_vector_index(tmp_db, index)
+
+    calls = {"n": 0}
+    original = tmp_db.fts_search
+
+    def counting(*args: Any, **kwargs: Any):
+        calls["n"] += 1
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(tmp_db, "fts_search", counting)
+
+    search_memories(tmp_db, "invoice", top_k=3, vector_index=index)
+    assert calls["n"] == 1
+
+    calls["n"] = 0
+    search_memories(tmp_db, "invoice", top_k=3)  # fallback path: also one pass
+    assert calls["n"] == 1
+
+
 def test_rebuild_on_null_index_is_noop(tmp_db: MemoryDB, tmp_path: Path):
     _add_fact(tmp_db, "1", "hello")
     index = get_vector_index(None, base_dir=tmp_path)