feat(node): share the (path,size,mtime)->hash index cache across every group

An operator routinely shares the same physical folder into more than one
group (a music library, a Séries drive) — IndexCache used to be opened
once per group (data_dir/{group_id}/index_cache.db), so the second group
to reference an already-fully-hashed multi-terabyte folder paid the same
full content read the first one did. IndexCache itself carried no
group_id in its schema; only daemon.py's wiring did. Now one instance,
opened once at startup (data_dir/index_cache.db), shared by every group's
DirectoryIndexer.

Confirmed against a real deployment (2026-08-25/26): a group sharing an
already-indexed folder with an existing group indexes it instantly, with
zero rehashing.

Also fixes a related cross-group correctness gap found during this work:
media_cache.db (thumbnails, TMDB/MusicBrainz metadata — already node-wide,
untouched by this change) was pruned for a file the moment it left *one*
group's index, even if another group's index still held the same content
hash — forcing a redundant re-fetch/re-probe/re-thumbnail for a group that
never actually lost anything. Prune now runs only once no group's index
references the file_id any more.

Adds a node admin UI action ("Maintenance" card, prune-index-cache) to
drop cache rows that no longer belong to any group's roots — skips
anything under a root that is merely temporarily unavailable (indexer.py's
"a root that goes away freezes, never empties" rule extends to this
cache too, or a reconnected drive would pay a full rehash for no reason).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013XSohfUQQiaE77qyFLgSv3

4 files changed, 288 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index b367a20..71aae78 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -440,6 +440,71 @@ async def test_delta_reflects_additions_and_deletions(tmp_path, shared_dir, gek)
 
 
 @pytest.mark.asyncio
+async def test_media_cache_not_pruned_when_another_group_still_has_the_content(
+        tmp_path, shared_dir, gek):
+    """
+    media_cache.db is node-wide, keyed by content hash — a file shared into
+    two groups is one row there. Removing it from ONE group's index (root
+    unshared, group left) must not wipe the thumbnail/tmdb/mbid mapping the
+    OTHER group's copy still needs, or that surviving group pays for a
+    redundant re-fetch/re-probe/re-thumbnail for content it never lost.
+    """
+    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32)
+    daemon._media_cache = AsyncMock()
+    daemon._webrtc = MagicMock()
+    daemon._webrtc._sessions = {}
+
+    indexer_a = DirectoryIndexer(roots=one_root(shared_dir), group_id="a" * 32,
+                                 sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer_a.initial_scan()
+    shared_id = indexer_a.index.entries[0].id
+
+    indexer_b = DirectoryIndexer(roots=one_root(shared_dir), group_id="b" * 32,
+                                 sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer_b.initial_scan()
+    assert indexer_b.index.get_entry(shared_id) is not None
+
+    daemon._indexers = [indexer_a, indexer_b]
+
+    await daemon._on_index_change(indexer_a)  # establishes the snapshot
+    await asyncio.sleep(0.05)
+
+    indexer_a.index.remove_entry(shared_id)
+    await daemon._on_index_change(indexer_a)
+    await asyncio.sleep(0.05)
+
+    daemon._media_cache.prune_file.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_media_cache_pruned_once_no_group_has_the_content_left(
+        tmp_path, shared_dir, gek):
+    """Counterpart of the test above: with only one group ever having held
+    the content, its removal must still prune media_cache as before — the
+    fix only withholds pruning when the content genuinely survives
+    elsewhere, it must not make pruning stop happening altogether."""
+    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32)
+    daemon._media_cache = AsyncMock()
+    daemon._webrtc = MagicMock()
+    daemon._webrtc._sessions = {}
+
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="a" * 32,
+                               sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer.initial_scan()
+    removed_id = indexer.index.entries[0].id
+    daemon._indexers = [indexer]
+
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+    indexer.index.remove_entry(removed_id)
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+    daemon._media_cache.prune_file.assert_called_once_with(removed_id)
+
+
+@pytest.mark.asyncio
 async def test_a_burst_of_changes_produces_one_broadcast(tmp_path, shared_dir, gek):
     """Coalescing: several _on_index_change calls in quick succession (one
     per debounced watchdog event) must collapse into a single push."""
diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py
index db0e37e..24e2f76 100644
--- a/packages/meshbay-node/tests/test_index_cache.py
+++ b/packages/meshbay-node/tests/test_index_cache.py
@@ -60,6 +60,52 @@ async def test_put_overwrites_previous_row_for_same_path(cache):
 
 
 @pytest.mark.asyncio
+async def test_count_reflects_number_of_rows(cache):
+    assert await cache.count() == 0
+
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+                     type="video", added_at=1)
+    await cache.put("/lib/b.mkv", size=2000, mtime=222.0, hash="b",
+                     type="video", added_at=2)
+
+    assert await cache.count() == 2
+
+
+@pytest.mark.asyncio
+async def test_all_paths_returns_every_row(cache):
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+                     type="video", added_at=1)
+    await cache.put("/lib/b.mkv", size=2000, mtime=222.0, hash="b",
+                     type="video", added_at=2)
+
+    assert set(await cache.all_paths()) == {"/lib/a.mkv", "/lib/b.mkv"}
+
+
+@pytest.mark.asyncio
+async def test_remove_many_drops_only_the_given_paths(cache):
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+                     type="video", added_at=1)
+    await cache.put("/lib/b.mkv", size=2000, mtime=222.0, hash="b",
+                     type="video", added_at=2)
+
+    await cache.remove_many(["/lib/a.mkv"])
+
+    assert await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) is None
+    assert await cache.lookup("/lib/b.mkv", size=2000, mtime=222.0) is not None
+    assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_remove_many_with_empty_list_is_a_no_op(cache):
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="a",
+                     type="video", added_at=1)
+
+    await cache.remove_many([])
+
+    assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
 async def test_cache_survives_reopen(tmp_path):
     db_path = tmp_path / "index_cache.db"
 
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index 9aa9fb9..729dade 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -284,6 +284,44 @@ async def test_second_scan_with_same_cache_hashes_nothing(
 
 
 @pytest.mark.asyncio
+async def test_second_group_sharing_the_same_folder_hashes_nothing(
+        shared_dir, sk_node, gek, index_cache):
+    """
+    The scenario the cache was made node-wide for (2026-08-25): an operator
+    shares the same physical folder into a second group. `IndexCache` is
+    keyed purely by absolute path, with no notion of group_id at all — a
+    *different* group_id scanning the same folder through the same shared
+    cache instance must hit exactly as hard as a same-group restart does
+    (the test right above this one). Before this cache was shared node-wide,
+    each group got its own on-disk cache file and this scan would have
+    rehashed every byte again.
+    """
+    first = DirectoryIndexer(roots=one_root(shared_dir), group_id="group-a",
+                             sk_node=sk_node, gek=gek, cache=index_cache)
+    await first.initial_scan()
+    assert first.index.count == 4
+
+    calls = []
+    real_scan_file = indexer_mod._scan_file
+
+    def spy(root, path):
+        calls.append(path)
+        return real_scan_file(root, path)
+
+    indexer_mod._scan_file = spy
+    try:
+        second = DirectoryIndexer(roots=one_root(shared_dir), group_id="group-b",
+                                  sk_node=sk_node, gek=gek, cache=index_cache)
+        await second.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert calls == [], (
+        f"a second group scanning the same folder must not rehash it, got {calls}")
+    assert {e.id for e in second.index.entries} == {e.id for e in first.index.entries}
+
+
+@pytest.mark.asyncio
 async def test_modified_file_is_rehashed(tmp_path, sk_node, gek, index_cache):
     d = tmp_path / "shared"
     d.mkdir()
diff --git a/packages/meshbay-node/tests/test_ops_index_cache.py b/packages/meshbay-node/tests/test_ops_index_cache.py
new file mode 100644
index 0000000..0b28d69
--- /dev/null
+++ b/packages/meshbay-node/tests/test_ops_index_cache.py
@@ -0,0 +1,139 @@
+"""Tests for ops.index_cache_stats / ops.prune_index_cache."""
+
+import types
+
+import pytest
+from meshbay_node import ops
+from meshbay_node.indexer.cache import IndexCache
+from meshbay_node.roots import Root
+
+
+@pytest.fixture
+async def cache(tmp_path):
+    c = IndexCache(db_path=tmp_path / "index_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+def _indexer(roots):
+    """A stand-in for DirectoryIndexer — prune_index_cache only ever reads
+    `.roots`, so a real one (with its watchdog Observer, executor, etc.)
+    would be pure overhead here."""
+    return types.SimpleNamespace(roots=roots)
+
+
+@pytest.mark.asyncio
+async def test_stats_with_no_cache_reports_zero():
+    assert await ops.index_cache_stats({}) == {"count": 0}
+
+
+@pytest.mark.asyncio
+async def test_stats_reports_row_count(cache):
+    await cache.put("/lib/a.mkv", size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+    assert await ops.index_cache_stats({"index_cache": cache}) == {"count": 1}
+
+
+@pytest.mark.asyncio
+async def test_prune_without_a_cache_raises():
+    with pytest.raises(ops.OpError):
+        await ops.prune_index_cache({})
+
+
+@pytest.mark.asyncio
+async def test_prune_removes_a_path_no_group_claims_any_more(cache):
+    """A folder no group's roots reference at all any more — every group
+    that once shared it was deleted, or the root was removed everywhere."""
+    await cache.put("/gone/orphan.mkv", size=1, mtime=1.0, hash="a",
+                     type="video", added_at=1)
+    state = {"index_cache": cache, "indexers": {}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 1, "kept": 0}
+    assert await cache.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_prune_keeps_a_path_whose_file_still_exists_under_an_available_root(
+        tmp_path, cache):
+    root_dir = tmp_path / "lib"
+    root_dir.mkdir()
+    f = root_dir / "still-here.mkv"
+    f.write_bytes(b"x")
+    await cache.put(str(f), size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+    root = Root(name="lib", path=root_dir, available=True)
+    state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 0, "kept": 1}
+    assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_prune_removes_a_path_deleted_from_an_available_root(tmp_path, cache):
+    root_dir = tmp_path / "lib"
+    root_dir.mkdir()
+    gone = root_dir / "deleted.mkv"
+    # Never written to disk — simulates a file that was there when cached
+    # and has since been deleted.
+    await cache.put(str(gone), size=1, mtime=1.0, hash="a", type="video", added_at=1)
+
+    root = Root(name="lib", path=root_dir, available=True)
+    state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 1, "kept": 0}
+    assert await cache.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_prune_never_touches_a_path_under_an_unavailable_root(tmp_path, cache):
+    """
+    indexer.py's own rule: a root that goes away freezes, it never empties.
+    A disconnected drive must not have its cache wiped just because the
+    files can't be verified right now — that would force a full rehash of
+    the whole drive the moment it comes back, exactly the cost this cache
+    exists to avoid. Checked with a root directory that does not even exist
+    on disk (the strongest form of "can't verify") to prove availability,
+    not on-disk state, is what decides this.
+    """
+    root_dir = tmp_path / "usb"  # never created — simulates "unplugged"
+    unreachable = root_dir / "movie.mkv"
+    await cache.put(str(unreachable), size=1, mtime=1.0, hash="a",
+                     type="video", added_at=1)
+
+    root = Root(name="usb", path=root_dir, available=False)
+    state = {"index_cache": cache, "indexers": {"g": _indexer([root])}}
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 0, "kept": 1}
+    assert await cache.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_prune_across_two_groups_sharing_one_root_path(tmp_path, cache):
+    """The scenario this whole cache redesign is for: the same folder is a
+    root of two groups. A path under it must be kept as long as *either*
+    group's root still resolves it, available."""
+    root_dir = tmp_path / "shared"
+    root_dir.mkdir()
+    f = root_dir / "track.mp3"
+    f.write_bytes(b"x")
+    await cache.put(str(f), size=1, mtime=1.0, hash="a", type="audio", added_at=1)
+
+    root_a = Root(name="shared", path=root_dir, available=True)
+    root_b = Root(name="shared", path=root_dir, available=True)
+    state = {
+        "index_cache": cache,
+        "indexers": {"g1": _indexer([root_a]), "g2": _indexer([root_b])},
+    }
+
+    result = await ops.prune_index_cache(state)
+
+    assert result == {"status": "pruned", "removed": 0, "kept": 1}