feat(node): persistent index cache, visible scan progress, adaptive reconcile, and delta sync

Indexer performance work, in four parts:

- Persistent (path, size, mtime) -> hash cache (indexer/cache.py) so a node
  restart no longer re-hashes every file — measured at 23 minutes for a
  114 GB library on a slow disk before this, near-instant after. Hashing
  is deliberately kept sequential (max_workers=1): it was never actually
  concurrent despite the pool size, and two interleaved reads seek-thrash
  a spinning disk instead of going faster.

- Byte-based scan progress (IndexProgress), surfaced via the loopback
  index-status route, the handshake ack, and a periodic INDEX_PROGRESS
  push to connected peers — drives a progress bar in the Create Group
  wizard and "add a directory" in Settings, and an animated presence dot.
  Guaranteed to settle back to idle via try/finally and a final push on
  the scanning->false transition.

- The reconcile backstop's directory walks now run in the executor
  instead of blocking the daemon's event loop; its interval defaults to
  10 min (was 60s) with adaptive backoff to 2h when nothing changes,
  reset on a real change or a peer connecting, and is now a per-group
  operator setting (signed op + group Settings UI).

- INDEX_DELTA wired up (protocol support existed, nothing called it):
  _on_index_change now sends additions/deletions instead of rebuilding
  the full entries list, coalesced over a short window so a burst of
  file events produces one push, and the hub swarm registration for
  public groups only (re-)registers newly added hashes.

Also fixes several bugs found while testing the above against real
libraries (a 114 GB and a 100+ GB group on a USB HDD):

- /api/reload blocked until the reload — including a brand-new group's
  full initial scan — finished, which the Electron bridge's fixed 30s
  call timeout turned into a hard failure on any real library. The route
  now fires the reload without waiting (ops.start_reload), matching
  add_root/remove_root's existing pattern; the wizard's own step order
  was fixed to wait for the group to actually appear hosted before the
  steps that need it (extra roots, GEK), with retries for the residual
  race between that and the daemon's own bookkeeping.
- transport.js's hand-rolled msgpack codec had no case for uint64/int64
  (0xcf/0xd3) and crashed decoding any message containing one — hit by
  IndexProgress.scanned_bytes/total_bytes for any group over ~4.3 GB.
  Verified against real msgpack-encoded bytes from the Python side.
- chat_hist_resp, and this change's own index_progress and
  set_scan_settings_ack pushes, were not routed by message type and
  could be handed to an unrelated pending request by the transport's
  "oldest pending" fallback, stalling it until its own 30s timeout and
  corrupting whatever received the wrong reply in its place.

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

9 files changed, 797 insertions, 72 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index c68f999..af85340 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -45,7 +45,7 @@ from meshbay_node.chat.store import ChatStore
 from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
 from meshbay_node.roots import RootSet, RootError
 from meshbay_node.hub_client import HubClient, HubConfig
-from meshbay_node.indexer import DirectoryIndexer
+from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex
 from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
 from meshbay_node.roster import Roster
 from meshbay_node.transport import (
@@ -119,6 +119,17 @@ class NodeDaemon:
         self._denylist = (
             Denylist(path=config.data_dir / "denylist.json") if Denylist else None)
         self._chat_stores: dict[str, ChatStore] = {}
+        self._index_caches: dict[str, IndexCache] = {}
+        # Coalesces a burst of index changes (one per debounced watchdog
+        # event) into a single broadcast — see _on_index_change. 0.5s is
+        # short enough nobody notices the wait, long enough that dropping a
+        # few hundred files into a watched folder produces one push instead
+        # of one per file.
+        self._broadcast_coalesce_secs = 0.5
+        self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {}
+        # group_id -> (version, {id: entry}) as of the last thing actually
+        # broadcast — the comparison point for the next delta.
+        self._last_broadcast_snapshot: dict[str, tuple] = {}
         self._audit_store: AuditStore | None = None
         self._bundle_store: BundleStore | None = None
         self._roster: Roster | None = None
@@ -234,12 +245,31 @@ class NodeDaemon:
                     log.info("No GEK yet for group %s — will accept first setup",
                              group_cfg.name)
 
+                index_cache = IndexCache(
+                    db_path=data_dir / group_cfg.id[:16] / "index_cache.db")
+                await index_cache.open()
+                self._index_caches[group_cfg.id] = index_cache
+
+                # Read once at load, like member_upload/enabled_apps below —
+                # kept current in place afterwards by set_scan_settings
+                # (ops.py), which updates both this indexer object directly
+                # and roster.db, so a restart picks up the same values.
+                scan_settings = (
+                    await self._roster.scan_settings(group_cfg.id)
+                    if self._roster else {
+                        "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+                        "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+                    })
+
                 indexer = DirectoryIndexer(
                     roots=roots,
                     group_id=group_cfg.id,
                     sk_node=keys.sk_ed25519,
                     gek=gek,
                     on_change=self._on_index_change,
+                    cache=index_cache,
+                    reconcile_secs=scan_settings["reconcile_interval_secs"],
+                    debounce_secs=scan_settings["debounce_secs"],
                 )
                 await indexer.start(defer_scan=True)
                 self._indexers.append(indexer)
@@ -253,6 +283,20 @@ class NodeDaemon:
                     "gek": gek,
                     "roots": roots,
                     "index": indexer.index,
+                    # Live reference, mutated in place by the indexer itself
+                    # (see IndexProgress in indexer.py) — read, never copied,
+                    # by the handshake ack and the periodic progress pusher.
+                    "progress": indexer.progress,
+                    # Bound method, called when a peer completes the
+                    # handshake — resets reconcile's backoff (indexer.py
+                    # _reconcile_loop) so the backstop is prompt again now
+                    # that someone is actually looking.
+                    "note_activity": indexer.note_activity,
+                    # Shown to the operator in Settings, and kept current in
+                    # place by set_scan_settings (ops.py) — same reasoning as
+                    # member_upload below.
+                    "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
+                    "debounce_secs": scan_settings["debounce_secs"],
                     "visibility": group_cfg.visibility,
                     # Admission policy comes from node.toml, never from the hub:
                     # a hub that could declare a group open would be handed its key.
@@ -456,6 +500,8 @@ class NodeDaemon:
                 if gctx:
                     self._tasks.append(asyncio.create_task(
                         _bg_scan(idx, group_cfg.name, gctx)))
+                    self._tasks.append(asyncio.create_task(
+                        self._progress_pusher(idx)))
 
             # 12. Wait for shutdown
             stop_event = asyncio.Event()
@@ -561,18 +607,40 @@ class NodeDaemon:
                 if gek:
                     log.info("GEK loaded for new group %s", group_cfg.id[:8])
 
+            index_cache = IndexCache(
+                db_path=data_dir / group_cfg.id[:16] / "index_cache.db")
+            await index_cache.open()
+            self._index_caches[group_cfg.id] = index_cache
+
+            scan_settings = (
+                await self._roster.scan_settings(group_cfg.id)
+                if self._roster else {
+                    "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+                    "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+                })
+
             indexer = DirectoryIndexer(
                 roots=roots,
                 group_id=group_cfg.id,
                 sk_node=sk_ed,
                 gek=gek,
                 on_change=self._on_index_change,
+                cache=index_cache,
+                reconcile_secs=scan_settings["reconcile_interval_secs"],
+                debounce_secs=scan_settings["debounce_secs"],
             )
-            await indexer.start()
+            # Registered *before* start() runs its (blocking, possibly very
+            # long — see the StarWars benchmark) initial scan, specifically
+            # so /api/groups/{id}/index-status can see indexer.progress
+            # while a brand-new group is still scanning — this is the one
+            # group state that must stay visible during the very window the
+            # group is not yet authorized for member connections (below).
             self._indexers.append(indexer)
             self._state["indexes"][group_cfg.id] = indexer.index
             self._state["indexers"][group_cfg.id] = indexer
 
+            await indexer.start()
+
             data_dir.mkdir(parents=True, exist_ok=True)
             chat_db = data_dir / group_cfg.id[:16] / "chat.db"
             store = ChatStore(db_path=chat_db)
@@ -583,6 +651,10 @@ class NodeDaemon:
                 "gek": gek,
                 "roots": roots,
                 "index": indexer.index,
+                "progress": indexer.progress,
+                "note_activity": indexer.note_activity,
+                "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
+                "debounce_secs": scan_settings["debounce_secs"],
                 "visibility": group_cfg.visibility,
                 "join_policy": group_cfg.join_policy,
                 "member_upload": (
@@ -600,6 +672,11 @@ class NodeDaemon:
             log.info("Hot-loaded group %s (%s, %d roots)",
                      group_cfg.name, group_cfg.id[:8], len(roots))
             added_names.append(group_cfg.name)
+            # The initial scan above already ran to completion (indexer.start()
+            # is not deferred here), so this only matters for whatever scans
+            # this group as time goes on — a root added later, reconcile
+            # picking one back up.
+            self._tasks.append(asyncio.create_task(self._progress_pusher(indexer)))
 
         # ── Tear down removed groups ─────────────────────────────────────
         removed_names = []
@@ -618,6 +695,16 @@ class NodeDaemon:
                     await store.close()
                 except Exception:
                     pass
+            cache = self._index_caches.pop(gid, None)
+            if cache:
+                try:
+                    await cache.close()
+                except Exception:
+                    pass
+            pending = self._pending_broadcasts.pop(gid, None)
+            if pending:
+                pending.cancel()
+            self._last_broadcast_snapshot.pop(gid, None)
             self._state["indexes"].pop(gid, None)
             self._state["indexers"].pop(gid, None)
             old_name = gid[:8]
@@ -708,46 +795,149 @@ class NodeDaemon:
         log.warning("No unwrappable GEK bundle found for group %s", group_id[:8])
         return None
 
+    async def _progress_pusher(self, indexer: DirectoryIndexer,
+                               interval: float = 2.0) -> None:
+        """
+        Watches indexer.progress and pushes a light INDEX_PROGRESS message to
+        this group's connected peers — never the index itself, that stays
+        _on_index_change's job. Runs for the node's whole lifetime: a scan
+        can start from several places (initial scan, a root added later,
+        reconcile picking a root back up), and this only needs to notice the
+        flag, not why it changed.
+
+        The final push at the False transition is what lets a presence dot
+        reliably turn back off on an already-connected client — the
+        handshake ack only covers the moment of connecting. `interval` is a
+        parameter (not a bare constant) only so a test can drive this loop
+        without waiting on the real 2s cadence.
+        """
+        was_scanning = False
+        while True:
+            await asyncio.sleep(interval)
+            progress = indexer.progress
+            now_scanning = progress.scanning
+            if now_scanning or was_scanning:
+                self._push_index_progress(indexer.group_id, progress)
+            was_scanning = now_scanning
+
+    def _push_index_progress(self, group_id: str, progress) -> None:
+        if not self._webrtc:
+            return
+        msg = {
+            "type": MNP.INDEX_PROGRESS,
+            "v": MNP_VERSION,
+            "group_id": group_id,
+            "scanning": progress.scanning,
+            "scanned_bytes": progress.scanned_bytes,
+            "total_bytes": progress.total_bytes,
+        }
+        pushed = 0
+        for session in list(self._webrtc._sessions.values()):
+            if session._group_id == group_id:
+                try:
+                    session._send(msg)
+                    pushed += 1
+                except Exception:
+                    pass
+        if pushed:
+            log.debug("Index progress pushed to %d peer(s) for group %s",
+                      pushed, group_id[:8])
+
     async def _on_index_change(self, indexer: DirectoryIndexer) -> None:
-        """Called when a DirectoryIndexer detects file changes."""
+        """
+        Called when a DirectoryIndexer detects file changes — once per
+        debounced watchdog event, so dropping N files into a watched folder
+        calls this N times in quick succession. Coalesces those into one
+        broadcast (_broadcast_index_change) rather than one push per file:
+        the timer is reset on every call and only fires once calls stop
+        arriving for _broadcast_coalesce_secs.
+        """
+        group_id = indexer.group_id
+        loop = asyncio.get_event_loop()
+        pending = self._pending_broadcasts.pop(group_id, None)
+        if pending:
+            pending.cancel()
+
+        def fire() -> None:
+            self._pending_broadcasts.pop(group_id, None)
+            asyncio.ensure_future(self._broadcast_index_change(indexer))
+
+        self._pending_broadcasts[group_id] = loop.call_later(
+            self._broadcast_coalesce_secs, fire)
+
+    async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None:
+        """
+        The actual push, run once per coalesced burst. Sends a full
+        INDEX_SYNC the first time a group is ever broadcast (no previous
+        snapshot to diff against — the client's own first fetchIndex() call
+        already covers that case) and an INDEX_DELTA every time after,
+        computed against the last thing this method actually sent.
+        """
         group_id = indexer.group_id
         idx = indexer.index
         log.info("Index changed for group %s: %d files (v%d)",
                  group_id[:8], idx.count, idx.version)
 
-        # 11.5 — Push updated index to connected WebRTC peers in this group
+        prev = self._last_broadcast_snapshot.get(group_id)
+        delta = None
+        if prev is not None:
+            prev_version, prev_entries = prev
+            previous = GroupIndex._snapshot(
+                idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries)
+            delta = idx.diff(previous)
+        self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id())
+
+        # 11.5 — Push to connected WebRTC peers in this group
         if self._webrtc:
-            entries = [
-                {
-                    "id": e.id, "name": e.name, "path": e.path,
-                    "size": e.size, "type": e.type, "added_at": e.added_at,
+            if delta is not None:
+                msg = {
+                    "type": MNP.INDEX_DELTA,
+                    "v": MNP_VERSION,
+                    "group_id": idx.group_id,
+                    "base_version": delta.base_version,
+                    "version": delta.version,
+                    "additions": [
+                        {"id": e.id, "name": e.name, "path": e.path,
+                         "size": e.size, "type": e.type, "added_at": e.added_at}
+                        for e in delta.additions
+                    ],
+                    "deletions": delta.deletions,
+                }
+            else:
+                msg = {
+                    "type": MNP.INDEX_SYNC,
+                    "v": MNP_VERSION,
+                    "group_id": idx.group_id,
+                    "version": idx.version,
+                    "entries": [
+                        {"id": e.id, "name": e.name, "path": e.path,
+                         "size": e.size, "type": e.type, "added_at": e.added_at}
+                        for e in idx.entries
+                    ],
                 }
-                for e in idx.entries
-            ]
-            sync_msg = {
-                "type": MNP.INDEX_SYNC,
-                "v": MNP_VERSION,
-                "group_id": idx.group_id,
-                "version": idx.version,
-                "entries": entries,
-            }
             pushed = 0
             for session in list(self._webrtc._sessions.values()):
                 if session._group_id == group_id:
                     try:
-                        session._send(sync_msg)
+                        session._send(msg)
                         pushed += 1
                     except Exception:
                         pass
             if pushed:
-                log.info("Index pushed to %d WebRTC peers", pushed)
+                log.info("Index %s pushed to %d WebRTC peers",
+                         "delta" if delta is not None else "sync", pushed)
 
         # 11.9 — Register file hashes with hub swarm table (public groups only, H7)
         group_cfg = next(
             (g for g in self._config.groups if g.id == group_id), None)
         if (self._hub and self._state.get("endpoint_hint")
                 and group_cfg and group_cfg.visibility == "public"):
-            hashes = [e.id for e in idx.entries]
+            # Only the newly added hashes once there is a delta to know them
+            # from — registering the whole library again on every change is
+            # the same O(changes x library size) cost the delta above exists
+            # to avoid.
+            hashes = ([e.id for e in delta.additions] if delta is not None
+                     else [e.id for e in idx.entries])
             if hashes:
                 endpoint = f"webrtc:{self._config.node.quic_port}"
                 asyncio.ensure_future(self._register_swarm(hashes, endpoint))
@@ -772,6 +962,10 @@ class NodeDaemon:
         log.info("Shutting down...")
         self._state["status"] = "stopping"
 
+        for handle in self._pending_broadcasts.values():
+            handle.cancel()
+        self._pending_broadcasts.clear()
+
         for task in self._tasks:
             task.cancel()
         for task in self._tasks:
@@ -795,6 +989,9 @@ class NodeDaemon:
         for store in self._chat_stores.values():
             await store.close()
 
+        for cache in self._index_caches.values():
+            await cache.close()
+
         for indexer in self._indexers:
             await indexer.stop()
 
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/__init__.py b/packages/meshbay-node/src/meshbay_node/indexer/__init__.py
index bb6231b..c92c730 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/__init__.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/__init__.py
@@ -1,5 +1,6 @@
 """Directory indexer and Mesh Group Index."""
 from .indexer import DirectoryIndexer
 from .group_index import GroupIndex
+from .cache import IndexCache
 
-__all__ = ["DirectoryIndexer", "GroupIndex"]
+__all__ = ["DirectoryIndexer", "GroupIndex", "IndexCache"]
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/cache.py b/packages/meshbay-node/src/meshbay_node/indexer/cache.py
new file mode 100644
index 0000000..c26ddf1
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/indexer/cache.py
@@ -0,0 +1,95 @@
+"""
+MeshBay Node — persistent (path, size, mtime) -> hash cache, one per group.
+
+Without this, every node restart re-reads and re-hashes every file in every
+root, even when nothing changed — measured at 23 minutes for a 114 GB library
+on a USB hard drive. This cache lets a scan skip the read entirely for a file
+whose size and mtime still match what was hashed last time.
+
+It is a path-keyed accelerator only. The GroupIndex itself stays keyed by
+content hash (see indexer.py's note on why two identical files are one
+entry) — this cache never changes that, it only avoids recomputing a hash
+that has not changed.
+"""
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+
+import aiosqlite
+
+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
+);
+"""
+
+
+@dataclass
+class CachedEntry:
+    hash: str
+    type: str
+    added_at: int
+
+
+class IndexCache:
+    """Async SQLite (path, size, mtime) -> hash cache for one group."""
+
+    def __init__(self, db_path: Path):
+        self._db_path = db_path
+        self._db: aiosqlite.Connection | None = None
+
+    async def open(self) -> None:
+        self._db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._db = await aiosqlite.connect(str(self._db_path))
+        await self._db.executescript(_SCHEMA)
+        await self._db.commit()
+
+    async def close(self) -> None:
+        if self._db:
+            await self._db.close()
+            self._db = None
+
+    async def __aenter__(self):
+        await self.open()
+        return self
+
+    async def __aexit__(self, *_):
+        await self.close()
+
+    async def lookup(self, path: str, size: int, mtime: float) -> 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.
+        """
+        async with self._db.execute(
+                "SELECT hash, type, added_at FROM files "
+                "WHERE path = ? AND size = ? AND mtime = ?",
+                (path, size, mtime)) as cur:
+            row = await cur.fetchone()
+        return CachedEntry(hash=row[0], type=row[1], added_at=row[2]) if row else None
+
+    async def put(self, path: str, size: int, mtime: float, hash: str,
+                  type: str, added_at: int) -> 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.
+        """
+        await self._db.execute(
+            "INSERT INTO files (path, mtime, size, hash, type, added_at) "
+            "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))
+        await self._db.commit()
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
index 5dbdc5d..ec98667 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
@@ -86,6 +86,23 @@ class GroupIndex:
     def count(self) -> int:
         return len(self._entries)
 
+    def entries_by_id(self) -> dict:
+        """A snapshot copy, for diff() to compare a later version against —
+        see daemon.py _on_index_change, the only caller."""
+        return dict(self._entries)
+
+    @classmethod
+    def _snapshot(cls, group_id: str, sk_node: Ed25519PrivateKey, gek: bytes | None,
+                  version: int, entries_by_id: dict) -> "GroupIndex":
+        """
+        A lightweight stand-in for diff()'s `previous` argument — never
+        serialized or sent anywhere, just a comparison point built from an
+        earlier entries_by_id() snapshot rather than a live GroupIndex.
+        """
+        idx = cls(group_id=group_id, sk_node=sk_node, gek=gek, version=version)
+        idx._entries = dict(entries_by_id)
+        return idx
+
     # ── Serialisation ─────────────────────────────────────────────────────────
 
     def serialize(self) -> bytes:
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index 482b556..b479f7e 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -26,6 +26,7 @@ import asyncio
 import logging
 import time
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable, Awaitable
 
@@ -36,6 +37,7 @@ from watchdog.observers import Observer
 
 from meshbay_common.paths import fold, find_fold_collisions, long_path
 from meshbay_common.protocol import IndexEntry
+from meshbay_node.indexer.cache import IndexCache
 from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.roots import Root, RootSet
 
@@ -75,6 +77,27 @@ def _is_indexable(path: Path) -> bool:
 _HASH_CHUNK = 8 * 1024 * 1024   # 8 MB streaming hash chunks
 
 
+@dataclass
+class IndexProgress:
+    """
+    A snapshot of "is this indexer mid-scan right now", for the status shown
+    to the operator (Create Group wizard, adding a directory) and pushed to
+    connected members (a presence dot, never anything more specific — see
+    daemon.py/webrtc_server.py). Reset per _scan_root() call rather than
+    accumulated across a group's roots: the consumers that matter always
+    watch exactly one root being scanned.
+
+    Mutated only from the asyncio loop thread (the hashing itself runs in an
+    executor thread, but never touches this), so no lock is needed.
+    """
+    scanning:      bool = False
+    scanned_bytes: int  = 0
+    total_bytes:   int  = 0
+    # Basename only, deliberately not the full path — enough to show progress
+    # without broadcasting the operator's directory structure.
+    current_dir:   str  = ""
+
+
 def _virtual_dir(root: Root, file_path: Path) -> str:
     """
     The directory a file appears in, as members see it: `"Films/2024"`.
@@ -87,6 +110,17 @@ def _virtual_dir(root: Root, file_path: Path) -> str:
     return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}"
 
 
+def _walk_root(root: Root) -> list[Path]:
+    """
+    Blocking directory walk — always run via an executor, never awaited
+    directly in the asyncio loop. A tree of tens of thousands of files (or
+    one on a slow network share) can take seconds; run inline, that stalls
+    every other thing the daemon is doing — WebRTC sessions, chat, the admin
+    UI — for as long as it takes.
+    """
+    return [p for p in root.path.rglob("*") if p.is_file()]
+
+
 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.)
@@ -136,8 +170,17 @@ class DirectoryIndexer:
     # How often to re-check which roots are readable and reconcile the index
     # against what is actually on disk. Not a poll for changes — a backstop for
     # the events the OS did not deliver, and the way a re-plugged drive is
-    # noticed.
-    RECONCILE_SECS = 60.0
+    # noticed. Watchdog already covers the common case in real time, so this
+    # does not need to run often to do its job; it backs off further still
+    # (see _reconcile_loop) when nothing has changed for a while, and a
+    # per-group operator setting (roster.py SETTING_RECONCILE_INTERVAL) can
+    # override the starting point.
+    DEFAULT_RECONCILE_SECS = 600.0    # 10 min
+    RECONCILE_BACKOFF_CAP  = 7200.0   # 2 h — never sleeps longer than this
+    # How long to wait after the last event on a given path before acting on
+    # it — several writes to the same file in quick succession (a slow copy
+    # in several passes) collapse into one hash instead of one per write.
+    DEFAULT_DEBOUNCE_SECS = 2.0
 
     def __init__(
         self,
@@ -146,20 +189,43 @@ class DirectoryIndexer:
         sk_node: Ed25519PrivateKey,
         gek: bytes | None,
         on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None,
+        cache: IndexCache | None = None,
+        reconcile_secs: float = DEFAULT_RECONCILE_SECS,
+        debounce_secs: float = DEFAULT_DEBOUNCE_SECS,
     ):
         self.roots    = roots
         self.group_id = group_id
         self.sk_node  = sk_node
         self.gek      = gek
         self.on_change = on_change
+        self.reconcile_secs = reconcile_secs
+        self.debounce_secs = debounce_secs
+        # Current backoff delay — starts at reconcile_secs, doubles on every
+        # tick that finds nothing changed (up to RECONCILE_BACKOFF_CAP), and
+        # resets the moment something real happens (a change, or a peer
+        # connecting — see note_activity()).
+        self._reconcile_delay = reconcile_secs
+        # Path -> (size, mtime, hash) accelerator, so a restart does not have
+        # to re-read a file it already hashed last time (see cache.py). None
+        # in tests that do not care about it — every hash is then a miss.
+        self._cache = cache
 
         self._index    = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek)
         self._index.roots = roots.describe()
-        self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="indexer")
+        # One worker, deliberately, not a real pool: hashing two files at once
+        # buys nothing here and can cost a lot. It only offloads the blocking
+        # read+hash off the asyncio loop; it was never used for concurrency —
+        # every call site awaits one run_in_executor before starting the next
+        # (see _hash_or_cached below) — and measured on a spinning USB drive,
+        # two interleaved multi-GB reads would seek-thrash against each other
+        # rather than go faster. Left at 1 so the number does not promise a
+        # concurrency this code never provided.
+        self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="indexer")
         self._observer: Observer | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
         self._reconciler: asyncio.Task | None = None
         self._pending_timers: dict[str, asyncio.TimerHandle] = {}
+        self.progress = IndexProgress()
 
     @property
     def index(self) -> GroupIndex:
@@ -204,21 +270,77 @@ class DirectoryIndexer:
 
     async def _scan_root(self, root: Root) -> int:
         log.info("Scanning %s (root %r) ...", root.path, root.name)
-        loop = asyncio.get_event_loop()
         count = 0
+        loop = asyncio.get_event_loop()
         try:
-            files = [p for p in root.path.rglob("*") if p.is_file()]
+            files = await loop.run_in_executor(self._executor, _walk_root, root)
         except OSError as e:
             log.warning("Cannot scan root %r: %s", root.name, e)
             return 0
-        for file_path in files:
-            entry = await loop.run_in_executor(
-                self._executor, _scan_file, root, file_path)
-            if entry:
-                self._index.add_entry(entry)
-                count += 1
+
+        # Sizes up front, off the same listing that already walked the tree —
+        # the progress bar's denominator, not a second pass over the disk.
+        sized: list[tuple[Path, int]] = []
+        for p in files:
+            try:
+                sized.append((p, p.stat().st_size))
+            except OSError:
+                continue
+
+        self.progress.scanning = True
+        self.progress.scanned_bytes = 0
+        self.progress.total_bytes = sum(size for _, size in sized)
+        self.progress.current_dir = ""
+        try:
+            for file_path, size in sized:
+                self.progress.current_dir = file_path.parent.name
+                entry = await self._hash_or_cached(root, file_path)
+                self.progress.scanned_bytes += size
+                if entry:
+                    self._index.add_entry(entry)
+                    count += 1
+        finally:
+            # Must run even if a hash/IO error propagates out of the loop
+            # above — an indexing state that never turns back off is worse
+            # than the scan itself failing.
+            self.progress.scanning = False
+            self.progress.current_dir = ""
         return count
 
+    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.
+        """
+        if not _is_indexable(file_path):
+            return None
+        try:
+            st = file_path.stat()
+        except OSError:
+            return None
+
+        if self._cache is not None:
+            cached = await self._cache.lookup(str(file_path), st.st_size, st.st_mtime)
+            if cached is not None:
+                return IndexEntry(
+                    id=cached.hash,
+                    name=file_path.name,
+                    path=_virtual_dir(root, file_path),
+                    size=st.st_size,
+                    type=cached.type,
+                    added_at=cached.added_at,
+                )
+
+        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)
+        return entry
+
     def _report_collisions(self) -> None:
         """
         Names that are the same file on a case-insensitive filesystem.
@@ -324,20 +446,38 @@ class DirectoryIndexer:
     async def _reconcile_loop(self) -> None:
         while True:
             try:
-                await asyncio.sleep(self.RECONCILE_SECS)
-                await self.reconcile()
+                await asyncio.sleep(self._reconcile_delay)
+                changed = await self.reconcile()
+                if changed:
+                    self._reconcile_delay = self.reconcile_secs
+                else:
+                    self._reconcile_delay = min(
+                        self._reconcile_delay * 2, self.RECONCILE_BACKOFF_CAP)
             except asyncio.CancelledError:
                 raise
             except Exception:
                 log.exception("Reconcile failed — continuing")
 
-    async def reconcile(self) -> None:
+    def note_activity(self) -> None:
+        """
+        Called when something makes a prompt reconcile worth having again —
+        today, a peer completing the handshake for this group
+        (webrtc_server.py). Someone is looking, so the backstop should be at
+        its normal cadence rather than however far backoff had stretched it.
+        """
+        self._reconcile_delay = self.reconcile_secs
+
+    async def reconcile(self) -> bool:
         """
         Re-check availability, and rescan roots that came back.
 
         The only place a root's entries are dropped: when the root is readable
         and the files are genuinely gone. A root that is not readable is left
         untouched, which is the whole point.
+
+        Returns whether anything actually changed — _reconcile_loop uses this
+        to back off when a pass finds nothing to do, rather than running at
+        the same cadence forever regardless of how quiet the root is.
         """
         changed = self.roots.refresh_availability()
         touched = False
@@ -367,6 +507,8 @@ class DirectoryIndexer:
             if self.on_change:
                 await self.on_change(self)
 
+        return touched
+
     async def _sweep_available_roots(self) -> bool:
         """
         Catch what the watcher missed: files gone, and files never announced.
@@ -375,24 +517,18 @@ class DirectoryIndexer:
         absent has nothing to compare against, and comparing anyway is exactly
         the mistake this module exists to avoid.
         """
-        loop = asyncio.get_event_loop()
         changed = False
+        loop = asyncio.get_event_loop()
         for root in self.roots:
             if not root.available:
                 continue
             try:
-                on_disk = {p.resolve() for p in root.path.rglob("*")
-                           if _is_indexable(p)}
+                on_disk, known = await loop.run_in_executor(
+                    self._executor, self._sweep_scan_root, root)
             except OSError as e:
                 log.warning("Cannot reconcile root %r: %s", root.name, e)
                 continue
 
-            known: dict[Path, str] = {}
-            for entry in self._entries_under(root):
-                abs_path = self._entry_path(root, entry)
-                if abs_path:
-                    known[abs_path] = entry.id
-
             for missing in set(known) - on_disk:
                 # Duplicate content is handled without a special case here: the
                 # entry goes, and the add loop below re-indexes the surviving
@@ -404,26 +540,46 @@ class DirectoryIndexer:
                 log.info("Reconcile: %s is gone", missing)
                 changed = True
 
-            for added in on_disk - set(known):
-                entry = await loop.run_in_executor(
-                    self._executor, _scan_file, root, added)
-                if not entry:
-                    continue
-                # The index is keyed by **content hash**, so two identical files
-                # at two paths are one entry and the path comparison above
-                # cannot see the second. Adding it anyway rewrites that entry's
-                # path every cycle, bumps the version, and pushes an index
-                # update to every connected peer once a minute — for ever.
-                # Measured on a live node: `clip.mp4` present at the root and in
-                # uploads/ with the same bytes.
-                if self._index.get_entry(entry.id) is not None:
-                    log.debug("Reconcile: %s duplicates content already indexed "
-                              "as %s — leaving the index alone",
-                              added, entry.id[:8])
-                    continue
-                self._index.add_entry(entry)
-                log.info("Reconcile: %s appeared (missed event)", added)
-                changed = True
+            added_paths = on_disk - set(known)
+            if not added_paths:
+                continue
+
+            added_sized: list[tuple[Path, int]] = []
+            for p in added_paths:
+                try:
+                    added_sized.append((p, p.stat().st_size))
+                except OSError:
+                    added_sized.append((p, 0))
+
+            self.progress.scanning = True
+            self.progress.scanned_bytes = 0
+            self.progress.total_bytes = sum(size for _, size in added_sized)
+            self.progress.current_dir = ""
+            try:
+                for added, size in added_sized:
+                    self.progress.current_dir = added.parent.name
+                    entry = await self._hash_or_cached(root, added)
+                    self.progress.scanned_bytes += size
+                    if not entry:
+                        continue
+                    # The index is keyed by **content hash**, so two identical
+                    # files at two paths are one entry and the path comparison
+                    # above cannot see the second. Adding it anyway rewrites
+                    # that entry's path every cycle, bumps the version, and
+                    # pushes an index update to every connected peer once a
+                    # minute — for ever. Measured on a live node: `clip.mp4`
+                    # present at the root and in uploads/ with the same bytes.
+                    if self._index.get_entry(entry.id) is not None:
+                        log.debug("Reconcile: %s duplicates content already "
+                                  "indexed as %s — leaving the index alone",
+                                  added, entry.id[:8])
+                        continue
+                    self._index.add_entry(entry)
+                    log.info("Reconcile: %s appeared (missed event)", added)
+                    changed = True
+            finally:
+                self.progress.scanning = False
+                self.progress.current_dir = ""
         return changed
 
     def _entries_under(self, root: Root) -> list[IndexEntry]:
@@ -444,6 +600,21 @@ class DirectoryIndexer:
         except OSError:
             return None
 
+    def _sweep_scan_root(self, root: Root) -> tuple[set[Path], dict[Path, str]]:
+        """
+        Blocking: the two disk-touching pieces of one reconcile pass for a
+        root, bundled so both run together in the executor rather than in
+        the asyncio loop — the tree walk, and resolving the real path of
+        every entry already known under this root (one syscall each).
+        """
+        on_disk = {p.resolve() for p in root.path.rglob("*") if _is_indexable(p)}
+        known: dict[Path, str] = {}
+        for entry in self._entries_under(root):
+            abs_path = self._entry_path(root, entry)
+            if abs_path:
+                known[abs_path] = entry.id
+        return on_disk, known
+
     def _restart_observer(self) -> None:
         """Re-schedule watches after roots appeared or disappeared."""
         if self._observer:
@@ -454,8 +625,6 @@ class DirectoryIndexer:
 
     # ── Internal update ───────────────────────────────────────────────────────
 
-    _DEBOUNCE_SECS = 2.0
-
     def _schedule_update(self, file_path: Path, deleted: bool = False) -> None:
         """Called from watchdog thread — schedule debounced async update."""
         if not self._loop:
@@ -472,7 +641,7 @@ class DirectoryIndexer:
             self._pending_timers.pop(key, None)
             asyncio.ensure_future(self._update_entry(file_path, deleted))
 
-        self._pending_timers[key] = self._loop.call_later(self._DEBOUNCE_SECS, fire)
+        self._pending_timers[key] = self._loop.call_later(self.debounce_secs, fire)
 
     def _remove_by_path(self, root: Root, file_path: Path) -> None:
         """Remove any existing entries that point at this file."""
@@ -511,9 +680,7 @@ class DirectoryIndexer:
         self._remove_by_path(root, file_path)
 
         if not deleted:
-            loop = asyncio.get_event_loop()
-            entry = await loop.run_in_executor(
-                self._executor, _scan_file, root, file_path)
+            entry = await self._hash_or_cached(root, file_path)
             if entry:
                 self._index.add_entry(entry)
                 log.debug("Indexed: %s (%s, %d bytes)",
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index c9d862a..154813d 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -22,6 +22,7 @@ in the adapter.
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import re
 from dataclasses import asdict
@@ -730,12 +731,69 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict:
     return {"apps": apps, "group_id": group_id}
 
 
+# ── Scan settings ────────────────────────────────────────────────────────────
+
+async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float,
+                            debounce_secs: float) -> dict:
+    """
+    How often the indexer's reconciliation backstop runs, and how long a
+    changed file is left alone before being hashed (indexer.py
+    DirectoryIndexer). Persisted like set_member_upload/set_enabled_apps —
+    but there is also a *live* DirectoryIndexer object to update, since it
+    reads these once at construction and runs its own background loop with
+    them rather than consulting groups_ctx on every use.
+    """
+    roster = _roster(state)
+    await roster.set_scan_settings(group_id, reconcile_interval_secs, debounce_secs,
+                                   set_by=state.get("node_user_id", ""))
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        indexer.reconcile_secs = reconcile_interval_secs
+        indexer.debounce_secs = debounce_secs
+        # Apply the new interval now rather than after whatever backoff had
+        # already stretched the wait to.
+        indexer.note_activity()
+    # Optional, unlike _group_ctx(): a group can be persisted here before it
+    # is hot-loaded (or in a test that only cares about the roster/indexer
+    # side), and that must not turn a successful write into a 404.
+    ctx = state.get("groups_ctx", {}).get(group_id)
+    if ctx is not None:
+        ctx["reconcile_interval_secs"] = reconcile_interval_secs
+        ctx["debounce_secs"] = debounce_secs
+    log.info("Scan settings for group %s: reconcile=%.0fs debounce=%.0fs",
+             group_id[:8], reconcile_interval_secs, debounce_secs)
+    return {"reconcile_interval_secs": reconcile_interval_secs,
+            "debounce_secs": debounce_secs, "group_id": group_id}
+
+
 # ── Reload ──────────────────────────────────────────────────────────────────
 
 async def reload_config(state: dict) -> dict:
-    """Hot-reload node.toml without dropping connections."""
+    """Hot-reload node.toml without dropping connections. Blocks until the
+    reload actually finishes — see start_reload for why the loopback route
+    uses that instead."""
     reload_fn = state.get("reload_fn")
     if not reload_fn:
         raise OpError("Reload not available", status=503)
     await reload_fn()
     return {"status": "reloaded"}
+
+
+async def start_reload(state: dict) -> dict:
+    """
+    Same as reload_config, but does not wait for the reload to finish.
+
+    The loopback route uses this one: the Electron bridge caps every call at
+    a fixed 30s (main.js node:call), and hot-loading a brand-new group runs
+    its full initial scan synchronously inside _reload_config_inner()
+    (daemon.py) before that coroutine returns — minutes, not seconds, on a
+    real library (found against a 45 GB group on the same slow disk the
+    StarWars benchmark used). The reload keeps running on the daemon's own
+    event loop either way; add_root/remove_root below already fire it the
+    same way for exactly this reason.
+    """
+    reload_fn = state.get("reload_fn")
+    if not reload_fn:
+        raise OpError("Reload not available", status=503)
+    asyncio.ensure_future(reload_fn())
+    return {"status": "reloading"}
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index 1cc8cae..6eb7651 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -599,6 +599,39 @@ class Roster:
                                json.dumps(sorted(apps)), set_by)
         return apps
 
+    # How often the indexer's reconciliation backstop runs, and how long it
+    # waits after the last change on a file before hashing it. Unset means
+    # the indexer's own defaults — an existing group's behaviour must not
+    # change because a node was upgraded. See indexer.py DirectoryIndexer
+    # for what these actually do and why the defaults are what they are.
+    SETTING_RECONCILE_INTERVAL = "reconcile_interval_secs"
+    SETTING_DEBOUNCE_SECS = "debounce_secs"
+
+    async def scan_settings(self, group_id: str) -> dict:
+        # Imported here, not at module load: roster.py is loaded before the
+        # indexer package during startup, and this is the only place the two
+        # need each other's names.
+        from meshbay_node.indexer.indexer import DirectoryIndexer
+
+        reconcile = await self.get_setting(group_id, self.SETTING_RECONCILE_INTERVAL)
+        debounce = await self.get_setting(group_id, self.SETTING_DEBOUNCE_SECS)
+        return {
+            "reconcile_interval_secs": (
+                float(reconcile) if reconcile is not None
+                else DirectoryIndexer.DEFAULT_RECONCILE_SECS),
+            "debounce_secs": (
+                float(debounce) if debounce is not None
+                else DirectoryIndexer.DEFAULT_DEBOUNCE_SECS),
+        }
+
+    async def set_scan_settings(self, group_id: str, reconcile_interval_secs: float,
+                                debounce_secs: float, set_by: str = "") -> dict:
+        await self.set_setting(group_id, self.SETTING_RECONCILE_INTERVAL,
+                               str(float(reconcile_interval_secs)), set_by)
+        await self.set_setting(group_id, self.SETTING_DEBOUNCE_SECS,
+                               str(float(debounce_secs)), set_by)
+        return await self.scan_settings(group_id)
+
     async def create_invite(
         self,
         group_id: str,
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index b6f572a..fa6c3e9 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -63,6 +63,7 @@ from meshbay_common.adminop import (
     OP_MEMBER_UNPIN,
     OP_MEMBER_UPLOAD,
     OP_APPS_ENABLED,
+    OP_SET_SCAN_SETTINGS,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
     OP_GROUP_ATTACH,
@@ -85,6 +86,7 @@ from meshbay_common.join import (
 from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node import ops
 from meshbay_node.roots import (
     RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name,
@@ -409,6 +411,8 @@ class WebRTCPeerSession:
                 self._do_member_upload(msg)
             elif mtype == MNP.APPS_ENABLED:
                 self._do_apps_enabled(msg)
+            elif mtype == MNP.SET_SCAN_SETTINGS:
+                self._do_set_scan_settings(msg)
             elif mtype == MNP.MEMBER_UNPIN:
                 self._do_member_unpin(msg)
             elif mtype == MNP.GEK_ROTATE:
@@ -659,6 +663,20 @@ class WebRTCPeerSession:
             # setting (or one whose context has not loaded it yet) hides
             # nothing.
             "enabled_apps": list(self._group_ctx().get("enabled_apps") or []),
+            # So a client that connects mid-scan shows the indexing state
+            # immediately, instead of waiting for the next periodic
+            # INDEX_PROGRESS push. Never a path or filename — see
+            # IndexProgress in indexer.py.
+            "indexing": self._indexing_status(),
+            # Current values only — not enforced from here, just shown to
+            # the operator in Settings so the number on screen matches what
+            # the indexer is actually doing (set_scan_settings, ops.py).
+            "scan_settings": {
+                "reconcile_interval_secs": self._group_ctx().get(
+                    "reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS),
+                "debounce_secs": self._group_ctx().get(
+                    "debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS),
+            },
         }
         if node_user_id:
             ack["node_user_id"] = node_user_id
@@ -668,6 +686,13 @@ class WebRTCPeerSession:
         self._send(ack)
         self._audit("handshake")
 
+        # Someone is here now — reconcile's backstop should be prompt again
+        # rather than however far its backoff had stretched while nobody
+        # was connected (indexer.py DirectoryIndexer.note_activity).
+        note_activity = self._group_ctx().get("note_activity")
+        if note_activity:
+            note_activity()
+
     async def _do_gek_bundle_fetch(self) -> None:
         """Serve the caller's wrapped GEK bundle during the handshake window."""
         bundle_store = self._ctx.get("bundle_store")
@@ -1648,6 +1673,69 @@ class WebRTCPeerSession:
             except Exception:
                 pass
 
+    # Reconcile's backstop and the watchdog debounce (indexer.py
+    # DirectoryIndexer) — how hard the node works on the operator's own
+    # disk, not a member-facing permission. Signed for the same reason as
+    # apps_enabled: consistency of the authorization model, not because a
+    # wrong value here is itself dangerous.
+    MIN_RECONCILE_SECS = 10.0
+    MAX_RECONCILE_SECS = 24 * 3600.0
+    MIN_DEBOUNCE_SECS = 0.0
+    MAX_DEBOUNCE_SECS = 300.0
+
+    def _do_set_scan_settings(self, msg: dict) -> None:
+        try:
+            reconcile = float(msg.get("reconcile_interval_secs"))
+            debounce = float(msg.get("debounce_secs"))
+        except (TypeError, ValueError):
+            self._send({"type": "error", "detail": "Invalid scan settings"})
+            return
+        if not (self.MIN_RECONCILE_SECS <= reconcile <= self.MAX_RECONCILE_SECS):
+            self._send({"type": "error",
+                       "detail": f"reconcile_interval_secs must be between "
+                                 f"{self.MIN_RECONCILE_SECS:.0f} and "
+                                 f"{self.MAX_RECONCILE_SECS:.0f}"})
+            return
+        if not (self.MIN_DEBOUNCE_SECS <= debounce <= self.MAX_DEBOUNCE_SECS):
+            self._send({"type": "error",
+                       "detail": f"debounce_secs must be between "
+                                 f"{self.MIN_DEBOUNCE_SECS:.0f} and "
+                                 f"{self.MAX_DEBOUNCE_SECS:.0f}"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}")
+
+    async def _admin_exec_set_scan_settings(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        try:
+            reconcile_s, debounce_s = pending["subject"].split(",")
+            reconcile, debounce = float(reconcile_s), float(debounce_s)
+        except (ValueError, KeyError):
+            self._send({"type": "error", "detail": "Invalid scan settings"})
+            return
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"set_scan_settings:{pending['subject']}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_scan_settings, self._group_id or "", reconcile, debounce)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("set_scan_settings", pending["subject"])
+
+        notice = {"type": MNP.SET_SCAN_SETTINGS_ACK, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     # ── Node management (D5) ─────────────────────────────────────────────────
 
     async def _do_node_status(self, msg: dict) -> None:
@@ -2012,6 +2100,23 @@ class WebRTCPeerSession:
             return self._ctx["groups"][self._group_id]
         return self._ctx
 
+    def _indexing_status(self) -> dict:
+        """
+        {"scanning": bool, "scanned_bytes": int, "total_bytes": int} for the
+        handshake ack and INDEX_PROGRESS pushes — never a path or filename,
+        that stays local to the operator's own admin UI. Absent "progress"
+        (context not loaded, or a group with no indexer at all) reads as
+        idle rather than erroring.
+        """
+        progress = self._group_ctx().get("progress")
+        if progress is None:
+            return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0}
+        return {
+            "scanning": progress.scanning,
+            "scanned_bytes": progress.scanned_bytes,
+            "total_bytes": progress.total_bytes,
+        }
+
     def _peer_registry(self) -> dict:
         """
         Connected peers for THIS group only.
@@ -2626,6 +2731,9 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_APPS_ENABLED:
             self._spawn(
                 self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_SET_SCAN_SETTINGS:
+            self._spawn(
+                self._admin_exec_set_scan_settings(pending, transcript, sig_bytes))
         elif pending["op"] == OP_ROOT_ADD:
             self._spawn(
                 self._admin_exec_root_add(pending, transcript, sig_bytes))
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index b505f25..d5b3f94 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -26,6 +26,7 @@ from fastapi.responses import HTMLResponse, JSONResponse
 from meshbay_node import __version__
 from meshbay_node import ops
 from meshbay_node.config import DEFAULT_CONFIG_PATH
+from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_common.crypto import generate_gek, wrap_gek_aes
 from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
 
@@ -319,6 +320,39 @@ def create_ui_app(state: dict) -> FastAPI:
             asyncio.ensure_future(reload_fn())
         return result
 
+    # asyncio.ensure_future above schedules the reload (and whatever initial
+    # scan it triggers) on the daemon's own event loop — it has no link to
+    # this HTTP request or to any browser tab. Closing the client that made
+    # this call does not cancel it: the scan is the node's own background
+    # work, not something borrowed from the request that started it.
+
+    @app.get("/api/groups/{group_id}/index-status")
+    async def index_status(group_id: str):
+        """
+        Polled by the Create Group wizard and by "add a directory" in
+        Settings — the same source either way, since both just start a scan
+        on this group's indexer. `current_dir` is a basename only, and is
+        never sent over MNP (see IndexProgress in indexer.py) — this route
+        is loopback-only, for the operator's own screen.
+
+        Reads state["indexers"] rather than groups_ctx: a brand-new group is
+        registered there before its (possibly long) initial scan runs, but
+        is only added to groups_ctx once that scan finishes (it is not yet
+        authorized for member connections either way — see _reload_config)
+        — this is precisely the window the wizard needs to watch.
+        """
+        indexer = state.get("indexers", {}).get(group_id)
+        progress = indexer.progress if indexer else None
+        if progress is None:
+            return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0,
+                    "current_dir": ""}
+        return {
+            "scanning": progress.scanning,
+            "scanned_bytes": progress.scanned_bytes,
+            "total_bytes": progress.total_bytes,
+            "current_dir": progress.current_dir,
+        }
+
     # ── Upload toggle (operator only, localhost) ─────────────────────────
 
     @app.put("/api/groups/{group_id}/member-upload")
@@ -327,11 +361,26 @@ def create_ui_app(state: dict) -> FastAPI:
             state, group_id, bool(payload.get("allowed", False)),
         ))
 
+    # ── Scan settings (operator only, localhost) ──────────────────────────
+
+    @app.put("/api/groups/{group_id}/scan-settings")
+    async def set_scan_settings(group_id: str, payload: dict):
+        return await _op(lambda: ops.set_scan_settings(
+            state, group_id,
+            float(payload.get("reconcile_interval_secs",
+                              DirectoryIndexer.DEFAULT_RECONCILE_SECS)),
+            float(payload.get("debounce_secs",
+                              DirectoryIndexer.DEFAULT_DEBOUNCE_SECS)),
+        ))
+
     # ── Reload config ────────────────────────────────────────────────────
 
     @app.post("/api/reload")
     async def reload_config():
-        return await _op(lambda: ops.reload_config(state))
+        # start_reload, not reload_config: this must return before a
+        # brand-new group's synchronous initial scan finishes (minutes, not
+        # seconds, on a real library) — see ops.start_reload for why.
+        return await _op(lambda: ops.start_reload(state))
 
     # ── Chat endpoints ───────────────────────────────────────────────────────