feat(node): indexing v2 — partial-read hashing for files above 40 MB

Files above 40 MB are no longer read in full. Instead, blake3 hashes
45 MB of samples (first 20 MB + last 20 MB + 5 MB at 50% offset).
Files at or below 40 MB are unchanged (full read, hash_version 1).

A new `hash_version` field on IndexEntry (default 1) travels on the
wire and through the cache so both versions coexist without breaking
existing nodes or clients.

The IndexCache auto-migrates its schema on open (ALTER TABLE), so no
manual step is required on upgrade. A standalone migration script is
available in QE/migration/ for operators who want to preview or force
a full re-hash.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

6 files changed, 619 insertions, 40 deletions

diff --git a/docs/indexing-v2.md b/docs/indexing-v2.md
new file mode 100644
index 0000000..6fc7990
--- /dev/null
+++ b/docs/indexing-v2.md
@@ -0,0 +1,312 @@
+# Indexing v2 — Partial-read hashing for large files
+
+> Status: **plan, not built.** Decision record and implementation checklist.
+
+---
+
+## 0. Problem
+
+The current indexer reads every file in full to compute its blake3 content hash (`id`).
+For a large media library (multi-terabyte, thousands of files), this means:
+
+- **Time**: initial indexing takes tens of minutes to hours.
+- **Disk I/O**: every byte of every file is read, which wears SSDs and saturates spinning
+  drives for the entire duration. A USB hard drive serving a 4 TB library is pegged for
+  over an hour.
+- **Blocking**: no connected peer receives a usable index until the full scan finishes.
+
+The hash exists for **content identity** (deduplication, cross-group search, file
+requests). A 4 GB film does not need 4 GB of I/O to be identified with overwhelming
+probability — 45 MB of well-chosen samples suffice.
+
+---
+
+## 1. Design
+
+### 1.1 Hashing rules
+
+| File size          | Method                                              | `hash_version` |
+|--------------------|-----------------------------------------------------|-----------------|
+| <= 40 MB           | Full read, blake3 of entire content (unchanged)     | `1`             |
+| > 40 MB            | Partial read, blake3 of 45 MB sampled (see below)   | `2`             |
+
+**Partial-read algorithm (hash_version 2):**
+
+Given a file of `S` bytes where `S > 40 MB`:
+
+1. Read the first **20 MB** (bytes `[0, 20 MB)`).
+2. Append the last **20 MB** (bytes `[S - 20 MB, S)`).
+3. Append **5 MB** starting at **50% of the file** (bytes `[S // 2, S // 2 + 5 MB)`).
+4. Compute `blake3(concatenation of the three regions)`.
+
+The three regions may overlap for files just above 40 MB. This is fine — the concatenation
+is deterministic for a given file, which is the only property that matters.
+
+**Why these offsets.** Head and tail catch container headers, trailers, and the common case
+of files that differ only at one end (re-encoded, re-muxed, appended). The mid-sample
+catches files that share a header and trailer but differ in content (same container,
+different media stream).
+
+**Why 40 MB threshold.** Below 40 MB the partial read would sample the entire file anyway
+(head + tail >= file size), so the full-read path is both simpler and produces the same
+result. The boundary is inclusive: a 40 MB file is read in full.
+
+### 1.2 `hash_version` field
+
+A new field on `IndexEntry`:
+
+```
+hash_version: int = 1
+```
+
+- `1` — the `id` is blake3 of the full file content. This is the only value any existing
+  node has ever produced.
+- `2` — the `id` is blake3 of the 45 MB partial sample described above.
+
+**For files <= 40 MB on a v2 node, `hash_version` stays `1`.** The hash is identical to
+what a v1 node produces, because both read the file in full. This preserves cross-group
+search compatibility for small files across v1 and v2 nodes.
+
+**For files > 40 MB on a v2 node, `hash_version` is `2`.** The hash is different from
+what a v1 node would produce for the same file. This is the accepted side effect.
+
+### 1.3 Backward compatibility
+
+| Scenario | Behaviour |
+|---|---|
+| v2 node sends `hash_version` to v1 client | Client ignores unknown field (JS objects are open) |
+| v1 node sends entries without `hash_version` | Client/consumer treats it as `1` (dataclass default) |
+| v2 `IndexEntry(**e)` where `e` lacks `hash_version` | Uses default `1` — existing serialized indexes deserialize correctly |
+| Cross-group search: same file, one node v1, one node v2 | Different `id` for files > 40 MB — not merged. Accepted |
+| Cross-group search: same small file, mixed nodes | Same `id` (both `hash_version=1`) — merged correctly |
+| Hub tables (`swarm_sources`, `content_blocklist`, `content_reports`) | Store `content_hash` as an opaque string. No change needed |
+| `GroupIndex.serialize()` / `deserialize()` | `asdict(e)` includes `hash_version`; `IndexEntry(**e)` with default handles missing field |
+
+**Nothing breaks.** A v1 node's data remains valid. A v2 node produces correct new hashes.
+Mixed v1/v2 environments work, with the documented search side effect.
+
+---
+
+## 2. Affected components
+
+### 2.1 `meshbay_common` — `protocol.py`
+
+| Change | Detail |
+|---|---|
+| `IndexEntry` dataclass | Add `hash_version: int = 1` field |
+| `index_entry_wire()` | Add `"hash_version": e.hash_version` to the wire dict |
+
+### 2.2 `meshbay_node` — `indexer/indexer.py`
+
+| Change | Detail |
+|---|---|
+| Constants | `_PARTIAL_THRESHOLD = 40 * 1024 * 1024`, `_PARTIAL_HEAD = 20 * 1024 * 1024`, `_PARTIAL_TAIL = 20 * 1024 * 1024`, `_PARTIAL_MID = 5 * 1024 * 1024` |
+| `_scan_file()` | After stat, if `size > _PARTIAL_THRESHOLD`: use partial-read blake3. Set `hash_version=2` on the returned `IndexEntry`. Otherwise: unchanged (full read, `hash_version=1`) |
+| `_hash_or_cached()` | Compute expected `hash_version` from file size. Pass it to cache `lookup()`. Store it in cache `put()`. Set it on the returned `IndexEntry` |
+
+**`_scan_file` partial-read implementation:**
+
+```python
+def _partial_hash(file_path: Path, size: int) -> str:
+    hasher = blake3.blake3()
+    with open(long_path(file_path), "rb") as f:
+        _feed(hasher, f, _PARTIAL_HEAD)
+        f.seek(size - _PARTIAL_TAIL)
+        _feed(hasher, f, _PARTIAL_TAIL)
+        f.seek(size // 2)
+        _feed(hasher, f, _PARTIAL_MID)
+    return hasher.hexdigest()
+
+def _feed(hasher, f, nbytes: int) -> None:
+    remaining = nbytes
+    while remaining > 0:
+        chunk = f.read(min(_HASH_CHUNK, remaining))
+        if not chunk:
+            break
+        hasher.update(chunk)
+        remaining -= len(chunk)
+```
+
+### 2.3 `meshbay_node` — `indexer/cache.py`
+
+| Change | Detail |
+|---|---|
+| Schema | Add `hash_version INTEGER NOT NULL DEFAULT 1` column to `files` table |
+| `_SCHEMA` | New databases get the column via `CREATE TABLE` |
+| `_MIGRATE_V2` | `ALTER TABLE files ADD COLUMN hash_version INTEGER NOT NULL DEFAULT 1` — applied on `open()` if the column does not exist |
+| `lookup()` | Add `hash_version` parameter. WHERE clause becomes `path = ? AND size = ? AND mtime = ? AND hash_version = ?` |
+| `put()` | Add `hash_version` parameter. INSERT includes `hash_version` |
+| `CachedEntry` | Add `hash_version: int` field |
+
+**Auto-migration on open:** `IndexCache.open()` runs the ALTER TABLE inside a try/except
+(column already exists → no-op). This way a node upgrade just works — no manual step
+needed for the cache.
+
+### 2.4 `meshbay_node` — `indexer/group_index.py`
+
+No code change needed. `serialize()` calls `asdict(e)` which includes `hash_version`.
+`deserialize()` calls `IndexEntry(**e)` which uses the default `1` for entries written
+before v2.
+
+### 2.5 `meshbay_node` — `transport/wire.py`
+
+No code change needed. `index_sync_message()` and `index_delta_message()` call
+`index_entry_wire()` which is updated in §2.1.
+
+### 2.6 Client side (browser / desktop)
+
+**No code changes required.** Entries are JavaScript objects; the extra `hash_version`
+field is carried through without needing explicit handling:
+
+- `transport.js` — `_applyIndexMessage()` passes the opened payload through. `hash_version`
+  rides along on each entry object.
+- `group-page.js` — `applyIndex()` / `applyIndexDelta()` store entries as-is in state
+  and in IndexedDB.
+- `hub-client.js` — IndexedDB cache stores entry objects verbatim.
+- `source-merge.js` — merges by `id`. Different hashes naturally don't merge.
+- `search-page.js` — aggregates entries from cached indexes. No change.
+- `files-app.js` — renders entries. `hash_version` is ignored.
+
+### 2.7 Hub side
+
+**No code changes required.** `swarm_sources.content_hash`, `content_reports.content_hash`,
+`content_blocklist.content_hash` are opaque `String(64)` columns. They store whatever
+blake3 hex the node provides. No hub migration needed.
+
+### 2.8 Tests
+
+| Test | What it verifies |
+|---|---|
+| `test_indexer.py` — new cases | Partial hash for file > 40 MB produces `hash_version=2`. File <= 40 MB produces `hash_version=1`. Partial hash is deterministic. Partial hash differs from full hash for the same large file |
+| `test_indexer.py` — existing cases | All existing tests still pass (small files, type detection, enrichment, etc.) |
+| `test_index_cache.py` — new cases | Cache lookup with `hash_version` match. Cache miss when `hash_version` differs. Schema migration from v1 cache |
+| `test_index_cache.py` — existing cases | Unchanged behaviour for v1 entries |
+| `test_indexer.py` — round-trip | `GroupIndex.serialize()` → `deserialize()` preserves `hash_version` on entries |
+| `test_index_seal_client.py` | Wire format includes `hash_version`, old entries without it deserialize as v1 |
+
+---
+
+## 3. Migration
+
+### 3.1 What actually needs migrating
+
+The node has two relevant stores:
+
+| Store | Location | Content | Migration |
+|---|---|---|---|
+| `index_cache.db` | `data_dir/index_cache.db` | `(path, size, mtime) → hash` accelerator | Add `hash_version` column |
+| In-memory `GroupIndex` | rebuilt from disk on every startup | Current file listing | No migration — rebuilt on next scan |
+
+**The IndexCache is the only persistent store that needs a schema change.** The GroupIndex
+is rebuilt by scanning the filesystem on every daemon start. Once the code uses v2 hashing,
+the next startup produces v2 hashes for large files automatically.
+
+**The cache auto-migrates.** `IndexCache.open()` adds the `hash_version` column if missing.
+Existing rows get `DEFAULT 1`. When the v2 indexer looks up a large file with
+`hash_version=2`, the cached v1 entry won't match (different hash_version in WHERE), so
+the file is re-hashed with the partial algorithm and the new entry is written with
+`hash_version=2`.
+
+This means: **large files are re-hashed lazily on first scan after upgrade.** The first
+scan after upgrading to v2 re-reads 45 MB per large file instead of the full content —
+already much faster than v1's full read.
+
+### 3.2 Migration script — `QE/migration/migrate_index_v2.sh`
+
+A standalone bash script (not in git — QE/ is gitignored) that:
+
+1. Detects the node's `data_dir` from `node.toml` (default `~/.local/share/meshbay-node/`)
+2. Checks that `index_cache.db` exists
+3. Runs `ALTER TABLE files ADD COLUMN hash_version INTEGER NOT NULL DEFAULT 1`
+4. Optionally (`--purge-large`) deletes cache entries for files > 40 MB, forcing immediate
+   re-hash on next scan instead of lazy migration
+5. Reports what it did
+
+**Works on Ubuntu and Fedora** — uses only `sqlite3` (present by default on both) and
+standard bash.
+
+**Not strictly required** if the code's auto-migration in `IndexCache.open()` is
+implemented. The script exists for operators who want to:
+- Verify the schema change before restarting
+- Force a clean re-hash of all large files in one pass
+- Run the migration on a machine where the node isn't installed yet (preparing a data dir)
+
+### 3.3 What the operator does
+
+1. Update the node package (or `pip install -e` in dev)
+2. (Optional) Run `QE/migration/migrate_index_v2.sh` to preview or force the migration
+3. Restart the node daemon
+4. The first scan re-hashes files > 40 MB with partial reads — much faster than before
+
+No client-side action needed. No hub-side action needed.
+
+---
+
+## 4. What does NOT change
+
+- **Files <= 40 MB** — identical hashing, identical `id`, `hash_version=1`.
+- **Enrichment** (thumbnails, duration, metadata) — unchanged, still runs after hashing.
+- **File downloads** — `file_req` uses the current `id` from the index. After re-indexing,
+  clients get the new index with new hashes and request accordingly.
+- **Watchdog / reconciliation** — filesystem events trigger the same code paths. New or
+  modified files are hashed with the appropriate method based on size.
+- **Index encryption / signing** — `GroupIndex.serialize()` and the GEK-sealed wire
+  messages are unchanged. `hash_version` rides inside each entry via `asdict()`.
+- **Index delta computation** — `GroupIndex.diff()` compares entries by all fields
+  (dataclass `__eq__`). A re-indexed file whose hash changed (v1 → v2) appears as a
+  deletion of the old id + addition of the new id, which is correct.
+- **Hub tables** — opaque content_hash storage, untouched.
+- **MNP version** — this is an additive field on index entries. No protocol version bump
+  needed. An older node that doesn't send `hash_version` is handled by the default.
+
+---
+
+## 5. Side effects — documented and accepted
+
+1. **Cross-group search**: a file > 40 MB indexed on a v1 node and a v2 node produces
+   different `id`s. The search page will not merge them as the same file. The user sees
+   two entries instead of one, each from its own group. This resolves itself when both
+   nodes upgrade.
+
+2. **First scan after upgrade**: files > 40 MB are re-hashed. With v2 this reads 45 MB
+   per file (not the full content), so the re-index is fast — but it is not instant. A
+   4 TB library with 1000 large files reads ~44 GB instead of 4 TB.
+
+3. **`content_hash` drift on hub tables**: if a public group's node upgrades, the hashes
+   it registers in `swarm_sources` change for large files. Old entries with v1 hashes
+   become stale. The swarm registration mechanism's `last_seen` update handles this — stale
+   entries age out. No explicit cleanup needed.
+
+4. **Index version bump**: re-indexing sets `GroupIndex.version = int(time.time())`,
+   which triggers a full `index_sync` to all connected peers. This is the normal path for
+   any index change — it is not new load.
+
+---
+
+## 6. Indexing trigger points — verified safe
+
+| Trigger | Location | Impact of v2 |
+|---|---|---|
+| Daemon startup — initial scan | `daemon.py:648-674` | Uses `_hash_or_cached()` which applies v2 rules. Safe |
+| Create Group wizard — Step 3 | `create-group-page.js:244-248` polls `index-status` | No change — polls progress, doesn't control hashing |
+| Settings — add root | `group-settings.js:909-927` | Triggers `retarget()` → scan → `_hash_or_cached()`. Safe |
+| Watchdog — file created/modified | `indexer.py:797-816` | Calls `_update_entry()` → `_hash_or_cached()`. Safe |
+| Reconciliation loop | `indexer.py:513-650` | Calls `_sweep_available_roots()` → `_hash_or_cached()`. Safe |
+| Hot reload — `_reload_config()` | `daemon.py:790-829` | Creates new `DirectoryIndexer` with v2 code. Safe |
+
+---
+
+## 7. Implementation order
+
+1. **`protocol.py`** — add `hash_version` field to `IndexEntry` and `index_entry_wire()`
+2. **`cache.py`** — add `hash_version` to schema, auto-migrate on open, update lookup/put
+3. **`indexer.py`** — implement `_partial_hash()`, update `_scan_file()` and
+   `_hash_or_cached()`
+4. **Tests** — new test cases for partial hashing, cache versioning, wire round-trip
+5. **`QE/migration/migrate_index_v2.sh`** — standalone migration script
+6. **Manual test** — run a node with a mixed library (small + large files), verify:
+   - Small files: same hash as before, `hash_version=1`
+   - Large files: different hash, `hash_version=2`, 45 MB read
+   - Cross-group search: small files merge, large files don't (across v1/v2 nodes)
+   - Cache hit on second scan: no re-read
+   - Index sync to connected peers: entries carry `hash_version`
diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py
index 4d2dd9d..e502379 100644
--- a/packages/meshbay-common/src/meshbay_common/protocol.py
+++ b/packages/meshbay-common/src/meshbay_common/protocol.py
@@ -204,6 +204,7 @@ class IndexEntry:
     track_no:   int | None = None  # tag or parsed, Music app
     taken_at:   int | None = None  # unix timestamp, EXIF DateTimeOriginal — Photos app
     camera:     str | None = None  # "Make Model", when both present — Photos app
+    hash_version: int = 1          # 1 = full-file blake3, 2 = partial-read (45 MB sample)
 
 
 def index_entry_wire(e: IndexEntry) -> dict:
@@ -224,6 +225,7 @@ def index_entry_wire(e: IndexEntry) -> dict:
         "season": e.season, "episode": e.episode,
         "artist": e.artist, "album": e.album, "track_no": e.track_no,
         "taken_at": e.taken_at, "camera": e.camera,
+        "hash_version": e.hash_version,
     }
 
 
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/cache.py b/packages/meshbay-node/src/meshbay_node/indexer/cache.py
index 31b9b15..c167db9 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/cache.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/cache.py
@@ -32,21 +32,25 @@ log = logging.getLogger(__name__)
 
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS files (
-    path      TEXT PRIMARY KEY,
-    mtime     REAL NOT NULL,
-    size      INTEGER NOT NULL,
-    hash      TEXT NOT NULL,
-    type      TEXT NOT NULL,
-    added_at  INTEGER NOT NULL
+    path         TEXT PRIMARY KEY,
+    mtime        REAL NOT NULL,
+    size         INTEGER NOT NULL,
+    hash         TEXT NOT NULL,
+    type         TEXT NOT NULL,
+    added_at     INTEGER NOT NULL,
+    hash_version INTEGER NOT NULL DEFAULT 1
 );
 """
 
+_MIGRATE_V2 = "ALTER TABLE files ADD COLUMN hash_version INTEGER NOT NULL DEFAULT 1"
+
 
 @dataclass
 class CachedEntry:
     hash: str
     type: str
     added_at: int
+    hash_version: int = 1
 
 
 class IndexCache:
@@ -60,6 +64,10 @@ class IndexCache:
         self._db_path.parent.mkdir(parents=True, exist_ok=True)
         self._db = await aiosqlite.connect(str(self._db_path))
         await self._db.executescript(_SCHEMA)
+        try:
+            await self._db.execute(_MIGRATE_V2)
+        except Exception:
+            pass  # column already exists
         await self._db.commit()
 
     async def close(self) -> None:
@@ -74,34 +82,36 @@ class IndexCache:
     async def __aexit__(self, *_):
         await self.close()
 
-    async def lookup(self, path: str, size: int, mtime: float) -> CachedEntry | None:
+    async def lookup(self, path: str, size: int, mtime: float,
+                     hash_version: int = 1) -> CachedEntry | None:
         """
-        A cache hit requires an EXACT match on both size and mtime. A mtime
-        touched without a content change is a false negative (an unnecessary
-        rehash) — accepted, since the alternative (trusting a stale hash) is
-        a silent wrong answer instead of an occasional wasted read.
+        A cache hit requires an EXACT match on size, mtime AND hash_version.
+        A v1 cached hash won't serve a v2 lookup for the same path — the file
+        is re-hashed with the new algorithm instead.
         """
         async with self._db.execute(
-                "SELECT hash, type, added_at FROM files "
-                "WHERE path = ? AND size = ? AND mtime = ?",
-                (path, size, mtime)) as cur:
+                "SELECT hash, type, added_at, hash_version FROM files "
+                "WHERE path = ? AND size = ? AND mtime = ? AND hash_version = ?",
+                (path, size, mtime, hash_version)) as cur:
             row = await cur.fetchone()
-        return CachedEntry(hash=row[0], type=row[1], added_at=row[2]) if row else None
+        return CachedEntry(hash=row[0], type=row[1], added_at=row[2],
+                           hash_version=row[3]) if row else None
 
     async def put(self, path: str, size: int, mtime: float, hash: str,
-                  type: str, added_at: int) -> None:
+                  type: str, added_at: int, hash_version: int = 1) -> None:
         """
-        Written only once a file has been hashed in full — never partway
-        through — so a crash mid-hash leaves no stale/partial row behind: the
-        next scan simply finds no cache entry and hashes the file again.
+        Written only once a file has been hashed — never partway through — so
+        a crash mid-hash leaves no stale/partial row behind: the next scan
+        simply finds no cache entry and hashes the file again.
         """
         await self._db.execute(
-            "INSERT INTO files (path, mtime, size, hash, type, added_at) "
-            "VALUES (?, ?, ?, ?, ?, ?) "
+            "INSERT INTO files (path, mtime, size, hash, type, added_at, hash_version) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?) "
             "ON CONFLICT(path) DO UPDATE SET "
             "mtime = excluded.mtime, size = excluded.size, hash = excluded.hash, "
-            "type = excluded.type, added_at = excluded.added_at",
-            (path, mtime, size, hash, type, added_at))
+            "type = excluded.type, added_at = excluded.added_at, "
+            "hash_version = excluded.hash_version",
+            (path, mtime, size, hash, type, added_at, hash_version))
         await self._db.commit()
 
     # ── Maintenance (node admin UI "prune index cache") ──────────────────────
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index f78465b..8376c23 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -94,6 +94,32 @@ def _is_indexable_size(path: Path, size: int) -> bool:
 
 _HASH_CHUNK = 8 * 1024 * 1024   # 8 MB streaming hash chunks
 
+_PARTIAL_THRESHOLD = 40 * 1024 * 1024   # files above this use partial-read hashing
+_PARTIAL_HEAD      = 20 * 1024 * 1024
+_PARTIAL_TAIL      = 20 * 1024 * 1024
+_PARTIAL_MID       =  5 * 1024 * 1024
+
+
+def _feed(hasher, f, nbytes: int) -> None:
+    remaining = nbytes
+    while remaining > 0:
+        chunk = f.read(min(_HASH_CHUNK, remaining))
+        if not chunk:
+            break
+        hasher.update(chunk)
+        remaining -= len(chunk)
+
+
+def _partial_hash(file_path: Path, size: int) -> str:
+    hasher = blake3.blake3()
+    with open(long_path(file_path), "rb") as f:
+        _feed(hasher, f, _PARTIAL_HEAD)
+        f.seek(size - _PARTIAL_TAIL)
+        _feed(hasher, f, _PARTIAL_TAIL)
+        f.seek(size // 2)
+        _feed(hasher, f, _PARTIAL_MID)
+    return hasher.hexdigest()
+
 
 @dataclass
 class IndexProgress:
@@ -162,29 +188,33 @@ def _size_files(files: list[Path]) -> list[tuple[Path, int]]:
 
 def _scan_file(root: Root, file_path: Path) -> IndexEntry | None:
     """Compute IndexEntry for a file. Blocking — run in executor.
-    Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.)
-    don't require loading the whole file into memory."""
+    Files <= 40 MB are hashed in full (hash_version 1). Files > 40 MB use a
+    45 MB partial read — first 20 MB, last 20 MB, 5 MB at 50% — for
+    hash_version 2."""
     if not _is_indexable(file_path):
         return None
     try:
         stat = file_path.stat()
         if not _is_indexable_size(file_path, stat.st_size):
             return None
-        hasher = blake3.blake3()
-        # long_path is a no-op off Windows; there it is what lets a deep media
-        # library past MAX_PATH.
-        with open(long_path(file_path), "rb") as f:
-            while chunk := f.read(_HASH_CHUNK):
-                hasher.update(chunk)
+        if stat.st_size > _PARTIAL_THRESHOLD:
+            hex_hash = _partial_hash(file_path, stat.st_size)
+            hv = 2
+        else:
+            hasher = blake3.blake3()
+            with open(long_path(file_path), "rb") as f:
+                while chunk := f.read(_HASH_CHUNK):
+                    hasher.update(chunk)
+            hex_hash = hasher.hexdigest()
+            hv = 1
         return IndexEntry(
-            id=hasher.hexdigest(),
-            # Stored exactly as the filesystem gave it: this is the string that
-            # opens the file. Normalization is for comparison only.
+            id=hex_hash,
             name=file_path.name,
             path=_virtual_dir(root, file_path),
             size=stat.st_size,
             type=_detect_type(file_path),
             added_at=int(stat.st_mtime),
+            hash_version=hv,
         )
     except (OSError, PermissionError, ValueError) as e:
         log.warning("Cannot index %s: %s", file_path, e)
@@ -375,10 +405,8 @@ class DirectoryIndexer:
     async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None:
         """
         Cache-aware replacement for a bare _scan_file() call: skips the
-        content read entirely when this path's (size, mtime) still match
-        what was hashed last time — the difference between a redundant full
-        rehash of a 100+ GB library on every restart and a stat()-only pass.
-        The only place that decides to actually read a file's bytes.
+        content read entirely when this path's (size, mtime, hash_version)
+        still match what was hashed last time.
         """
         if not _is_indexable(file_path):
             return None
@@ -389,8 +417,11 @@ class DirectoryIndexer:
         if not _is_indexable_size(file_path, st.st_size):
             return None
 
+        expected_hv = 2 if st.st_size > _PARTIAL_THRESHOLD else 1
+
         if self._cache is not None:
-            cached = await self._cache.lookup(str(file_path), st.st_size, st.st_mtime)
+            cached = await self._cache.lookup(
+                str(file_path), st.st_size, st.st_mtime, expected_hv)
             if cached is not None:
                 return IndexEntry(
                     id=cached.hash,
@@ -399,13 +430,15 @@ class DirectoryIndexer:
                     size=st.st_size,
                     type=cached.type,
                     added_at=cached.added_at,
+                    hash_version=cached.hash_version,
                 )
 
         loop = asyncio.get_event_loop()
         entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path)
         if entry and self._cache is not None:
             await self._cache.put(str(file_path), st.st_size, st.st_mtime,
-                                   entry.id, entry.type, entry.added_at)
+                                   entry.id, entry.type, entry.added_at,
+                                   entry.hash_version)
         return entry
 
     def _report_collisions(self) -> None:
diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py
index 24e2f76..036b4d7 100644
--- a/packages/meshbay-node/tests/test_index_cache.py
+++ b/packages/meshbay-node/tests/test_index_cache.py
@@ -122,3 +122,80 @@ async def test_cache_survives_reopen(tmp_path):
 
     assert hit is not None
     assert hit.hash == "abc123"
+
+
+# ── hash_version support (indexing v2) ─────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_put_v2_then_lookup_hits(cache):
+    await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                     hash="partial_abc", type="video", added_at=42,
+                     hash_version=2)
+
+    hit = await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                              hash_version=2)
+    assert hit is not None
+    assert hit.hash == "partial_abc"
+    assert hit.hash_version == 2
+
+
+@pytest.mark.asyncio
+async def test_lookup_misses_on_wrong_hash_version(cache):
+    await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                     hash="full_hash", type="video", added_at=42,
+                     hash_version=1)
+
+    assert await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                               hash_version=2) is None
+
+
+@pytest.mark.asyncio
+async def test_put_v2_overwrites_v1_for_same_path(cache):
+    await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                     hash="full_hash", type="video", added_at=42,
+                     hash_version=1)
+    await cache.put("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                     hash="partial_hash", type="video", added_at=42,
+                     hash_version=2)
+
+    assert await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                               hash_version=1) is None
+    hit = await cache.lookup("/lib/big.mkv", size=50_000_000, mtime=111.0,
+                              hash_version=2)
+    assert hit is not None
+    assert hit.hash == "partial_hash"
+
+
+@pytest.mark.asyncio
+async def test_v1_schema_auto_migrates(tmp_path):
+    """An index_cache.db created by old code (no hash_version column) gains
+    the column on the next open(), and existing rows default to hash_version=1."""
+    import aiosqlite
+    db_path = tmp_path / "old_cache.db"
+    async with aiosqlite.connect(str(db_path)) as db:
+        await db.executescript("""
+            CREATE TABLE files (
+                path      TEXT PRIMARY KEY,
+                mtime     REAL NOT NULL,
+                size      INTEGER NOT NULL,
+                hash      TEXT NOT NULL,
+                type      TEXT NOT NULL,
+                added_at  INTEGER NOT NULL
+            );
+        """)
+        await db.execute(
+            "INSERT INTO files (path, mtime, size, hash, type, added_at) "
+            "VALUES (?, ?, ?, ?, ?, ?)",
+            ("/lib/old.mkv", 111.0, 1000, "oldhash", "video", 42))
+        await db.commit()
+
+    cache = IndexCache(db_path=db_path)
+    await cache.open()
+    hit = await cache.lookup("/lib/old.mkv", size=1000, mtime=111.0,
+                              hash_version=1)
+    await cache.close()
+
+    assert hit is not None
+    assert hit.hash == "oldhash"
+    assert hit.hash_version == 1
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index 729dade..6aee1b5 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -723,3 +723,148 @@ async def test_reconcile_backoff_resets_when_something_actually_changes(
             await task
         except asyncio.CancelledError:
             pass
+
+
+# ── Indexing v2 — partial-read hashing ────────────────────────────────────────
+
+
+def test_small_file_gets_hash_version_1(tmp_path):
+    from meshbay_node.indexer.indexer import _scan_file
+    d = tmp_path / "root"
+    d.mkdir()
+    f = d / "small.mp4"
+    f.write_bytes(os.urandom(1024))
+
+    entry = _scan_file(one_root(d).roots[0], f)
+    assert entry is not None
+    assert entry.hash_version == 1
+
+
+def test_large_file_gets_hash_version_2(tmp_path):
+    from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD
+    d = tmp_path / "root"
+    d.mkdir()
+    f = d / "big.mkv"
+    size = _PARTIAL_THRESHOLD + 1
+    f.write_bytes(os.urandom(size))
+
+    entry = _scan_file(one_root(d).roots[0], f)
+    assert entry is not None
+    assert entry.hash_version == 2
+    assert entry.size == size
+
+
+def test_file_at_threshold_gets_hash_version_1(tmp_path):
+    from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD
+    d = tmp_path / "root"
+    d.mkdir()
+    f = d / "exact.mkv"
+    f.write_bytes(os.urandom(_PARTIAL_THRESHOLD))
+
+    entry = _scan_file(one_root(d).roots[0], f)
+    assert entry is not None
+    assert entry.hash_version == 1
+
+
+def test_partial_hash_differs_from_full_hash(tmp_path):
+    """For a file above the threshold, the partial hash must differ from what
+    a full-file blake3 would produce (they read different bytes)."""
+    import blake3 as b3
+    from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD
+    d = tmp_path / "root"
+    d.mkdir()
+    f = d / "big.mkv"
+    content = os.urandom(_PARTIAL_THRESHOLD + 1024 * 1024)
+    f.write_bytes(content)
+
+    entry = _scan_file(one_root(d).roots[0], f)
+    full_hash = b3.blake3(content).hexdigest()
+
+    assert entry.id != full_hash
+    assert entry.hash_version == 2
+
+
+def test_partial_hash_is_deterministic(tmp_path):
+    from meshbay_node.indexer.indexer import _scan_file, _PARTIAL_THRESHOLD
+    d = tmp_path / "root"
+    d.mkdir()
+    f = d / "big.mkv"
+    f.write_bytes(os.urandom(_PARTIAL_THRESHOLD + 1))
+
+    e1 = _scan_file(one_root(d).roots[0], f)
+    e2 = _scan_file(one_root(d).roots[0], f)
+    assert e1.id == e2.id
+
+
+def test_group_index_roundtrip_preserves_hash_version(sk_node, gek):
+    from meshbay_common.protocol import IndexEntry
+    idx = GroupIndex(group_id="hv-test", sk_node=sk_node, gek=gek)
+    idx.add_entry(IndexEntry(
+        id="aaa", name="small.mp4", path="root", size=1024,
+        type="video", added_at=100, hash_version=1))
+    idx.add_entry(IndexEntry(
+        id="bbb", name="big.mkv", path="root", size=50_000_000,
+        type="video", added_at=200, hash_version=2))
+
+    wire = idx.serialize()
+    recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
+
+    by_id = {e.id: e for e in recovered.entries}
+    assert by_id["aaa"].hash_version == 1
+    assert by_id["bbb"].hash_version == 2
+
+
+def test_deserialize_without_hash_version_defaults_to_1(sk_node, gek):
+    """Entries serialized by old code (no hash_version field) must deserialize
+    as hash_version=1."""
+    from meshbay_common.protocol import IndexEntry
+    idx = GroupIndex(group_id="compat", sk_node=sk_node, gek=gek)
+    idx.add_entry(IndexEntry(
+        id="old", name="f.mp4", path="root", size=1024,
+        type="video", added_at=100))
+    wire = idx.serialize()
+    recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
+    assert recovered.entries[0].hash_version == 1
+
+
+def test_index_entry_wire_includes_hash_version():
+    from meshbay_common.protocol import IndexEntry, index_entry_wire
+    e = IndexEntry(id="x", name="f.mp4", path="root", size=1,
+                   type="video", added_at=0, hash_version=2)
+    w = index_entry_wire(e)
+    assert w["hash_version"] == 2
+
+
+@pytest.mark.asyncio
+async def test_cache_aware_scan_uses_hash_version(tmp_path, sk_node, gek):
+    from meshbay_node.indexer.indexer import _PARTIAL_THRESHOLD
+    d = tmp_path / "root"
+    d.mkdir()
+    small = d / "small.mp4"
+    small.write_bytes(os.urandom(1024))
+    big = d / "big.mkv"
+    big.write_bytes(os.urandom(_PARTIAL_THRESHOLD + 1))
+
+    cache = IndexCache(db_path=tmp_path / "cache.db")
+    await cache.open()
+
+    indexer = DirectoryIndexer(
+        roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek,
+        cache=cache)
+    await indexer.initial_scan()
+
+    by_name = {e.name: e for e in indexer.index.entries}
+    assert by_name["small.mp4"].hash_version == 1
+    assert by_name["big.mkv"].hash_version == 2
+
+    hit_small = await cache.lookup(
+        str(small), small.stat().st_size, small.stat().st_mtime,
+        hash_version=1)
+    assert hit_small is not None
+
+    hit_big = await cache.lookup(
+        str(big), big.stat().st_size, big.stat().st_mtime,
+        hash_version=2)
+    assert hit_big is not None
+
+    await cache.close()