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

7 files changed, 1239 insertions, 1 deletions

diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index ab1613d..b367a20 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -237,6 +237,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu
         data_dir=tmp_path / "data",
     )
     daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01  # real value would make this test wait 0.5s
     daemon._hub = AsyncMock()
     daemon._hub.register_swarm = AsyncMock(return_value=2)
     daemon._state["endpoint_hint"] = "node123"
@@ -256,6 +257,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu
     daemon._webrtc = mock_webrtc
 
     await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)  # let the coalescing timer fire
 
     mock_session._send.assert_called_once()
     msg = mock_session._send.call_args[0][0]
@@ -288,6 +290,7 @@ async def test_daemon_index_change_registers_swarm_for_public_group(
         data_dir=tmp_path / "data",
     )
     daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01
     daemon._hub = AsyncMock()
     daemon._hub.register_swarm = AsyncMock(return_value=2)
     daemon._state["endpoint_hint"] = "node123"
@@ -317,6 +320,7 @@ async def test_daemon_index_change_skips_other_group_peers(
         data_dir=tmp_path / "data",
     )
     daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01
     daemon._hub = AsyncMock()
     daemon._hub.register_swarm = AsyncMock(return_value=0)
     daemon._state["endpoint_hint"] = "node123"
@@ -340,6 +344,145 @@ async def test_daemon_index_change_skips_other_group_peers(
     daemon._webrtc = mock_webrtc
 
     await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
 
     same_group._send.assert_called_once()
     other_group._send.assert_not_called()
+
+
+# ── INDEX_DELTA (phase 4) ────────────────────────────────────────────────────
+
+def _new_daemon_for_group(tmp_path, shared_dir, gek, group_id="a" * 32,
+                          visibility="private"):
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id=group_id, name="test-group", shared_dir=str(shared_dir),
+            visibility=visibility, quic_port=29010,
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01
+    daemon._hub = AsyncMock()
+    daemon._hub.register_swarm = AsyncMock(return_value=0)
+    daemon._state["endpoint_hint"] = "node123"
+    return daemon
+
+
+@pytest.mark.asyncio
+async def test_first_broadcast_is_full_sync_second_is_delta(tmp_path, shared_dir, gek):
+    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek)
+    indexer = DirectoryIndexer(
+        roots=one_root(shared_dir), group_id="a" * 32,
+        sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer.initial_scan()
+
+    session = MagicMock()
+    session._group_id = "a" * 32
+    session._send = MagicMock()
+    mock_webrtc = MagicMock()
+    mock_webrtc._sessions = {"p1": session}
+    daemon._webrtc = mock_webrtc
+
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+    first = session._send.call_args_list[0].args[0]
+    assert first["type"] == "index_sync"
+    assert len(first["entries"]) == indexer.index.count
+
+    # Nothing actually changed in the index between the two calls, but
+    # _on_index_change does not know or care why it was called — the
+    # SECOND broadcast must still be a delta, now that there is a
+    # previous snapshot to diff against.
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+    second = session._send.call_args_list[1].args[0]
+    assert second["type"] == "index_delta"
+    assert second["additions"] == []
+    assert second["deletions"] == []
+
+
+@pytest.mark.asyncio
+async def test_delta_reflects_additions_and_deletions(tmp_path, shared_dir, gek):
+    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek)
+    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
+
+    session = MagicMock()
+    session._group_id = "a" * 32
+    session._send = MagicMock()
+    daemon._webrtc = MagicMock()
+    daemon._webrtc._sessions = {"p1": session}
+
+    await daemon._on_index_change(indexer)  # first: full sync, establishes the snapshot
+    await asyncio.sleep(0.05)
+
+    # A real change: one entry removed, one added.
+    indexer.index.remove_entry(removed_id)
+    from meshbay_common.protocol import IndexEntry
+    new_entry = IndexEntry(id="new-file-id", name="new.mp4", path="shared",
+                           size=10, type="video", added_at=0)
+    indexer.index.add_entry(new_entry)
+
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+    delta_msg = session._send.call_args_list[1].args[0]
+    assert delta_msg["type"] == "index_delta"
+    assert delta_msg["deletions"] == [removed_id]
+    assert [a["id"] for a in delta_msg["additions"]] == ["new-file-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."""
+    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek)
+    indexer = DirectoryIndexer(
+        roots=one_root(shared_dir), group_id="a" * 32,
+        sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer.initial_scan()
+
+    session = MagicMock()
+    session._group_id = "a" * 32
+    session._send = MagicMock()
+    daemon._webrtc = MagicMock()
+    daemon._webrtc._sessions = {"p1": session}
+
+    for _ in range(5):
+        await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+    session._send.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_swarm_registration_only_sends_new_hashes_after_the_first(
+        tmp_path, shared_dir, gek):
+    daemon = _new_daemon_for_group(tmp_path, shared_dir, gek, visibility="public")
+    indexer = DirectoryIndexer(
+        roots=one_root(shared_dir), group_id="a" * 32,
+        sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer.initial_scan()
+    total_files = indexer.index.count
+
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+    assert len(daemon._hub.register_swarm.call_args_list[0].args[0]) == total_files
+
+    from meshbay_common.protocol import IndexEntry
+    indexer.index.add_entry(IndexEntry(id="new-file-id", name="new.mp4",
+                                       path="shared", size=10, type="video",
+                                       added_at=0))
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+    assert daemon._hub.register_swarm.call_count == 2
+    assert daemon._hub.register_swarm.call_args_list[1].args[0] == ["new-file-id"], \
+        "only the newly added hash must be (re-)registered, not the whole library"
diff --git a/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py
new file mode 100644
index 0000000..7cb74cb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py
@@ -0,0 +1,329 @@
+"""
+Adding a group (Create Group wizard, or "add a directory" to an existing
+one) fires `_reload_config()` without awaiting it (`asyncio.ensure_future`,
+ui/app.py) — the request handler, and whatever browser tab triggered it,
+return immediately. This is deliberate: the initial scan behind it can take
+a very long time (measured at 23 minutes for a 114 GB library on a slow
+disk), and none of that work belongs to the HTTP request or the WebRTC
+session that happened to start it.
+
+This test proves the scan is genuinely independent of its caller: it starts
+the reload the same way the real endpoint does — schedules it and does not
+await it, standing in for "the browser tab that made the call was closed" —
+then does something else, and only afterwards checks that the reload
+finished and the new group became available on its own.
+"""
+
+import asyncio
+import base64
+import os
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from meshbay_node import ops
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+import meshbay_node.indexer.indexer as indexer_mod
+
+
+def _free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _toml(data_dir, first_group_dir, second_group_id=None, second_group_dir=None) -> str:
+    # data_dir MUST come before any [section] header — TOML has no notion of
+    # "back to top-level" once a table is open, so a bare `key = value` line
+    # placed after [node] becomes node.data_dir, not the top-level data_dir
+    # load_config() actually reads. Silently falls back to the real default
+    # (~/.local/share/meshbay) instead of erroring, which is exactly how this
+    # test once ran a whole daemon — including _shutdown()'s unlink of
+    # ui-token — against the developer's real, already-running node.
+    text = f"""
+data_dir = "{data_dir}"
+
+[hub]
+url = "http://localhost:9999"
+username = "testuser"
+
+[node]
+quic_port = {_free_port()}
+ui_port = {_free_port()}
+
+[[groups]]
+id = "{"a" * 32}"
+name = "first"
+shared_dir = "{first_group_dir}"
+visibility = "private"
+"""
+    if second_group_id:
+        text += f"""
+[[groups]]
+id = "{second_group_id}"
+name = "slow-new-group"
+shared_dir = "{second_group_dir}"
+visibility = "private"
+"""
+    return text
+
+
+def _mock_keystore_keys(sk_ed):
+    sk_x = X25519PrivateKey.generate()
+    pk_x_raw = sk_x.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    mock_keys = MagicMock()
+    mock_keys.sk_ed25519 = sk_ed
+    mock_keys.pk_ed25519_b64 = "test"
+    mock_keys.sk_x25519 = sk_x
+    mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode()
+    return mock_keys
+
+
+@pytest.mark.asyncio
+async def test_hot_loaded_group_finishes_scanning_without_anyone_awaiting_the_reload(
+        tmp_path):
+    first_dir = tmp_path / "first"
+    first_dir.mkdir()
+    (first_dir / "readme.txt").write_bytes(b"hello")
+
+    second_dir = tmp_path / "second"
+    second_dir.mkdir()
+    for i in range(3):
+        (second_dir / f"file{i}.bin").write_bytes(os.urandom(64))
+    second_group_id = "b" * 32
+
+    data_dir = tmp_path / "data"
+    config_path = tmp_path / "node.toml"
+    config_path.write_text(_toml(data_dir, first_dir))
+
+    from meshbay_node.config import load_config
+    daemon = NodeDaemon(load_config(config_path), config_path=config_path)
+
+    sk_node = Ed25519PrivateKey.generate()
+    mock_keys = _mock_keystore_keys(sk_node)
+    mock_session = MagicMock()
+    mock_session.node_id = "node123"
+    mock_session.user_id = "user123"
+    mock_session.hub_pk_pem = sk_node.public_key().public_bytes(
+        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+    # Slow the second group's hashing down (a stand-in for a large/slow
+    # library) so there is a real window in which "nobody is awaiting this"
+    # actually means something, without needing a genuinely huge file.
+    real_scan_file = indexer_mod._scan_file
+
+    def slow_scan_file(root, path):
+        import time
+        time.sleep(0.15)
+        return real_scan_file(root, path)
+
+    with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
+         patch("meshbay_node.daemon.HubClient") as MockHub, \
+         patch.object(indexer_mod, "_scan_file", slow_scan_file):
+
+        hub_instance = AsyncMock()
+        hub_instance.startup = AsyncMock(return_value=mock_session)
+        hub_instance.send_ws = AsyncMock()
+        hub_instance._ws = None
+        hub_instance.close = AsyncMock()
+        hub_instance.__aenter__ = AsyncMock(return_value=hub_instance)
+        hub_instance.__aexit__ = AsyncMock(return_value=False)
+        MockHub.return_value = hub_instance
+
+        shutdown_event = asyncio.Event()
+
+        async def mock_maintain_ws(**kwargs):
+            await shutdown_event.wait()
+        hub_instance.maintain_ws = mock_maintain_ws
+
+        async def run_daemon():
+            with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15):
+                try:
+                    await asyncio.wait_for(daemon.run(), timeout=15)
+                except (asyncio.TimeoutError, Exception):
+                    pass
+
+        run_task = asyncio.create_task(run_daemon())
+        try:
+            for _ in range(50):
+                if daemon._state.get("status") == "running":
+                    break
+                await asyncio.sleep(0.05)
+            assert daemon._state["status"] == "running"
+            assert second_group_id not in daemon._state.get("groups_ctx", {})
+
+            # Add the second group to the config on disk, the way the wizard's
+            # attach + /api/reload would leave it, then fire the reload exactly
+            # as ui/app.py does: scheduled, NOT awaited.
+            config_path.write_text(_toml(data_dir, first_dir,
+                                         second_group_id, second_dir))
+            reload_task = asyncio.ensure_future(daemon._reload_config())
+
+            # Stand in for "the browser tab is gone": do something completely
+            # unrelated to the reload, and explicitly do not await it here.
+            await asyncio.sleep(0.01)
+            assert second_group_id not in daemon._state.get("groups_ctx", {}), \
+                "the scan (3 files x 0.15s) cannot have finished yet"
+
+            # Only now catch up with the background work, from a place that
+            # has no relationship to whatever originally triggered it.
+            await asyncio.wait_for(reload_task, timeout=5)
+
+            assert second_group_id in daemon._state["groups_ctx"], \
+                "the new group must be usable once its scan finishes, " \
+                "regardless of whether anything was still watching the reload"
+            new_indexer = daemon._state["indexers"][second_group_id]
+            assert new_indexer.index.count == 3
+            assert new_indexer.progress.scanning is False
+        finally:
+            shutdown_event.set()
+            await daemon._shutdown()
+            run_task.cancel()
+            try:
+                await run_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
+
+@pytest.mark.asyncio
+async def test_group_scoped_ops_404_until_listed_then_succeed(tmp_path):
+    """
+    The wizard's own sequence, reproduced against the real ops layer: attach
+    a brand-new group, fire the reload the way /api/reload now does
+    (ops.start_reload — scheduled, not awaited), and hit the group-scoped
+    calls that come right after in the UI (add a root, init the GEK) while
+    the scan is still running.
+
+    Found live: "Attaching to node" no longer times out (ops.start_reload
+    returns immediately), but the very next wizard step then failed with
+    "Group not configured on this node" / "Group not hosted on this node" —
+    the group is not in daemon._state["config"].groups or ["groups_ctx"]
+    until _reload_config_inner() finishes, scan included, which is *after*
+    ops.start_reload has already returned. This locks in both halves: the
+    404 while the scan runs, and success once ops.list_groups() actually
+    lists the group — the exact condition the wizard's own wait
+    (platform.waitForGroupHosted, app.js) polls for.
+    """
+    first_dir = tmp_path / "first"
+    first_dir.mkdir()
+    (first_dir / "readme.txt").write_bytes(b"hello")
+
+    second_dir = tmp_path / "second"
+    second_dir.mkdir()
+    for i in range(3):
+        (second_dir / f"file{i}.bin").write_bytes(os.urandom(64))
+    second_group_id = "c" * 32
+    extra_root_dir = tmp_path / "extra_root"
+    extra_root_dir.mkdir()
+
+    data_dir = tmp_path / "data2"
+    config_path = tmp_path / "node2.toml"
+    config_path.write_text(_toml(data_dir, first_dir))
+
+    from meshbay_node.config import load_config
+    daemon = NodeDaemon(load_config(config_path), config_path=config_path)
+
+    sk_node = Ed25519PrivateKey.generate()
+    mock_keys = _mock_keystore_keys(sk_node)
+    mock_session = MagicMock()
+    mock_session.node_id = "node123"
+    mock_session.user_id = "user123"
+    mock_session.hub_pk_pem = sk_node.public_key().public_bytes(
+        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+    real_scan_file = indexer_mod._scan_file
+
+    def slow_scan_file(root, path):
+        import time
+        time.sleep(0.15)
+        return real_scan_file(root, path)
+
+    with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
+         patch("meshbay_node.daemon.HubClient") as MockHub, \
+         patch.object(indexer_mod, "_scan_file", slow_scan_file):
+
+        hub_instance = AsyncMock()
+        hub_instance.startup = AsyncMock(return_value=mock_session)
+        hub_instance.send_ws = AsyncMock()
+        hub_instance._ws = None
+        hub_instance.close = AsyncMock()
+        hub_instance.__aenter__ = AsyncMock(return_value=hub_instance)
+        hub_instance.__aexit__ = AsyncMock(return_value=False)
+        MockHub.return_value = hub_instance
+
+        shutdown_event = asyncio.Event()
+
+        async def mock_maintain_ws(**kwargs):
+            await shutdown_event.wait()
+        hub_instance.maintain_ws = mock_maintain_ws
+
+        async def run_daemon():
+            with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15):
+                try:
+                    await asyncio.wait_for(daemon.run(), timeout=15)
+                except (asyncio.TimeoutError, Exception):
+                    pass
+
+        run_task = asyncio.create_task(run_daemon())
+        try:
+            for _ in range(50):
+                if daemon._state.get("status") == "running":
+                    break
+                await asyncio.sleep(0.05)
+            assert daemon._state["status"] == "running"
+
+            # The same file write ops.attach_group does (a raw text append),
+            # then the same fire-and-forget reload /api/reload now does.
+            config_path.write_text(_toml(data_dir, first_dir,
+                                         second_group_id, second_dir))
+            reload_task = asyncio.ensure_future(ops.start_reload(daemon._state))
+
+            await asyncio.sleep(0.01)
+            listing = await ops.list_groups(daemon._state)
+            assert second_group_id not in [g["id"] for g in listing["groups"]], \
+                "the scan (3 files x 0.15s) cannot have finished this fast"
+
+            # Exactly the wizard's next two steps, hit mid-scan.
+            with pytest.raises(ops.OpError) as add_root_exc:
+                await ops.add_root(daemon._state, second_group_id,
+                                   str(extra_root_dir))
+            assert add_root_exc.value.status == 404
+
+            with pytest.raises(ops.OpError) as gek_exc:
+                await ops.set_gek(daemon._state, second_group_id)
+            assert gek_exc.value.status == 404
+
+            # Now wait the way platform.waitForGroupHosted (app.js) does:
+            # poll list_groups(), not index-status, until the group is
+            # actually there.
+            for _ in range(100):
+                listing = await ops.list_groups(daemon._state)
+                if second_group_id in [g["id"] for g in listing["groups"]]:
+                    break
+                await asyncio.sleep(0.05)
+            else:
+                pytest.fail("group never appeared in list_groups()")
+
+            await asyncio.wait_for(reload_task, timeout=5)
+
+            # Both calls that 404'd above must now succeed.
+            add_result = await ops.add_root(daemon._state, second_group_id,
+                                            str(extra_root_dir))
+            assert add_result["status"] == "added"
+
+            gek_result = await ops.set_gek(daemon._state, second_group_id)
+            assert gek_result["status"] == "ok"
+        finally:
+            shutdown_event.set()
+            await daemon._shutdown()
+            run_task.cancel()
+            try:
+                await run_task
+            except (asyncio.CancelledError, Exception):
+                pass
diff --git a/packages/meshbay-node/tests/test_index_cache.py b/packages/meshbay-node/tests/test_index_cache.py
new file mode 100644
index 0000000..db0e37e
--- /dev/null
+++ b/packages/meshbay-node/tests/test_index_cache.py
@@ -0,0 +1,78 @@
+"""Tests for the (path, size, mtime) -> hash cache (indexer/cache.py)."""
+
+import pytest
+
+from meshbay_node.indexer.cache import IndexCache
+
+
+@pytest.fixture
+async def cache(tmp_path):
+    c = IndexCache(db_path=tmp_path / "index_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+@pytest.mark.asyncio
+async def test_put_then_lookup_hits(cache):
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+                     type="video", added_at=42)
+
+    hit = await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0)
+
+    assert hit is not None
+    assert hit.hash == "abc123"
+    assert hit.type == "video"
+    assert hit.added_at == 42
+
+
+@pytest.mark.asyncio
+async def test_lookup_misses_on_unknown_path(cache):
+    assert await cache.lookup("/lib/never-seen.mkv", size=1, mtime=1.0) is None
+
+
+@pytest.mark.asyncio
+async def test_lookup_misses_on_different_mtime(cache):
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+                     type="video", added_at=42)
+
+    assert await cache.lookup("/lib/a.mkv", size=1000, mtime=222.0) is None
+
+
+@pytest.mark.asyncio
+async def test_lookup_misses_on_different_size(cache):
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+                     type="video", added_at=42)
+
+    assert await cache.lookup("/lib/a.mkv", size=2000, mtime=111.0) is None
+
+
+@pytest.mark.asyncio
+async def test_put_overwrites_previous_row_for_same_path(cache):
+    await cache.put("/lib/a.mkv", size=1000, mtime=111.0, hash="old",
+                     type="video", added_at=1)
+    await cache.put("/lib/a.mkv", size=2000, mtime=222.0, hash="new",
+                     type="video", added_at=2)
+
+    assert await cache.lookup("/lib/a.mkv", size=1000, mtime=111.0) is None
+    hit = await cache.lookup("/lib/a.mkv", size=2000, mtime=222.0)
+    assert hit.hash == "new"
+
+
+@pytest.mark.asyncio
+async def test_cache_survives_reopen(tmp_path):
+    db_path = tmp_path / "index_cache.db"
+
+    c1 = IndexCache(db_path=db_path)
+    await c1.open()
+    await c1.put("/lib/a.mkv", size=1000, mtime=111.0, hash="abc123",
+                 type="video", added_at=42)
+    await c1.close()
+
+    c2 = IndexCache(db_path=db_path)
+    await c2.open()
+    hit = await c2.lookup("/lib/a.mkv", size=1000, mtime=111.0)
+    await c2.close()
+
+    assert hit is not None
+    assert hit.hash == "abc123"
diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py
new file mode 100644
index 0000000..52cd78b
--- /dev/null
+++ b/packages/meshbay-node/tests/test_index_progress.py
@@ -0,0 +1,165 @@
+"""
+Indexing status visible node -> client: handshake ack field, the loopback
+status route for the Create Group wizard / "add a directory", and the
+periodic INDEX_PROGRESS push to already-connected peers. Never the index
+itself (see test_daemon.py for that) and never anything sent to the hub.
+"""
+
+import asyncio
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from unittest.mock import MagicMock
+from fastapi.testclient import TestClient
+
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer.indexer import IndexProgress
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+from meshbay_node.ui.app import create_ui_app
+
+
+def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * 32):
+    index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate())
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    group_ctx = {"index": index}
+    if progress is not None:
+        group_ctx["progress"] = progress
+    session._ctx = {"groups": {group_id: group_ctx}}
+    session._group_id = group_id
+    return session
+
+
+# ── _indexing_status() ───────────────────────────────────────────────────────
+
+def test_indexing_status_defaults_idle_when_no_progress_tracked():
+    session = _session_with_progress(None)
+    assert session._indexing_status() == {
+        "scanning": False, "scanned_bytes": 0, "total_bytes": 0}
+
+
+def test_indexing_status_reflects_live_progress():
+    progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000,
+                             current_dir="StarWars")
+    session = _session_with_progress(progress)
+
+    status = session._indexing_status()
+
+    assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000}
+    assert "current_dir" not in status, \
+        "the directory name is operator-local detail, never sent to a member"
+
+
+# ── /api/groups/{id}/index-status (loopback) ────────────────────────────────
+
+def _ui_client(state: dict) -> TestClient:
+    return TestClient(create_ui_app({"status": "running", "groups_ctx": {},
+                                     "indexes": {}, **state}))
+
+
+def test_index_status_route_idle_for_unknown_group():
+    client = _ui_client({"indexers": {}})
+    resp = client.get("/api/groups/unknown-group/index-status")
+    assert resp.status_code == 200
+    assert resp.json() == {"scanning": False, "scanned_bytes": 0,
+                           "total_bytes": 0, "current_dir": ""}
+
+
+def test_index_status_route_reflects_indexer_progress():
+    fake_indexer = MagicMock()
+    fake_indexer.progress = IndexProgress(
+        scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000,
+        current_dir="2024")
+    client = _ui_client({"indexers": {"g" * 32: fake_indexer}})
+
+    resp = client.get(f"/api/groups/{'g' * 32}/index-status")
+
+    assert resp.json() == {
+        "scanning": True, "scanned_bytes": 1_000_000,
+        "total_bytes": 4_000_000_000, "current_dir": "2024",
+    }
+
+
+# ── _push_index_progress / _progress_pusher ─────────────────────────────────
+
+def _daemon(tmp_path) -> NodeDaemon:
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(),
+        groups=[],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    return NodeDaemon(config)
+
+
+@pytest.mark.asyncio
+async def test_push_index_progress_only_reaches_same_group_peers(tmp_path):
+    daemon = _daemon(tmp_path)
+
+    same_group = MagicMock()
+    same_group._group_id = "a" * 32
+    same_group._send = MagicMock()
+    other_group = MagicMock()
+    other_group._group_id = "b" * 32
+    other_group._send = MagicMock()
+
+    mock_webrtc = MagicMock()
+    mock_webrtc._sessions = {"p1": same_group, "p2": other_group}
+    daemon._webrtc = mock_webrtc
+
+    progress = IndexProgress(scanning=True, scanned_bytes=10, total_bytes=100)
+    daemon._push_index_progress("a" * 32, progress)
+
+    same_group._send.assert_called_once()
+    msg = same_group._send.call_args[0][0]
+    assert msg["type"] == "index_progress"
+    assert msg["group_id"] == "a" * 32
+    assert msg["scanning"] is True
+    assert msg["scanned_bytes"] == 10
+    assert msg["total_bytes"] == 100
+    other_group._send.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path):
+    daemon = _daemon(tmp_path)
+
+    session = MagicMock()
+    session._group_id = "a" * 32
+    session._send = MagicMock()
+    mock_webrtc = MagicMock()
+    mock_webrtc._sessions = {"p1": session}
+    daemon._webrtc = mock_webrtc
+
+    indexer = MagicMock()
+    indexer.group_id = "a" * 32
+    indexer.progress = IndexProgress(scanning=True, scanned_bytes=0, total_bytes=100)
+
+    task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05))
+    try:
+        # Two ticks while still scanning.
+        await asyncio.sleep(0.12)
+        assert session._send.call_count >= 2
+        assert all(c.args[0]["scanning"] is True for c in session._send.call_args_list)
+
+        # Scan finishes between ticks.
+        indexer.progress.scanning = False
+        calls_before = session._send.call_count
+        await asyncio.sleep(0.07)
+        assert session._send.call_count == calls_before + 1, \
+            "exactly one final push must follow the False transition"
+        assert session._send.call_args.args[0]["scanning"] is False
+
+        # Nothing further once idle.
+        calls_after_final = session._send.call_count
+        await asyncio.sleep(0.15)
+        assert session._send.call_count == calls_after_final, \
+            "no more pushes once idle and already reported"
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index c304361..68ccc4c 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -9,7 +9,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 
 from meshbay_common.crypto import generate_gek
-from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache
+import meshbay_node.indexer.indexer as indexer_mod
 from conftest import one_root
 from meshbay_node.keystore import NodeKeys
 
@@ -190,3 +191,290 @@ async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek):
     wire = indexer.index.serialize()
     recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
     assert recovered.count == indexer.index.count
+
+
+# ── Cache-aware scanning ───────────────────────────────────────────────────────
+
+@pytest.fixture
+async def index_cache(tmp_path):
+    c = IndexCache(db_path=tmp_path / "index_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+@pytest.mark.asyncio
+async def test_second_scan_with_same_cache_hashes_nothing(
+        shared_dir, sk_node, gek, index_cache):
+    """
+    The whole point of the cache: a "restart" (a fresh DirectoryIndexer, same
+    on-disk cache) that finds every file's (size, mtime) unchanged must not
+    read a single byte of file content.
+    """
+    first = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                             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="g",
+                                  sk_node=sk_node, gek=gek, cache=index_cache)
+        await second.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert calls == [], f"expected zero hash calls on a fully-cached rescan, got {calls}"
+    assert second.index.count == first.index.count
+    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()
+    f = d / "movie.mkv"
+    f.write_bytes(b"original content")
+
+    first = DirectoryIndexer(roots=one_root(d), group_id="g",
+                             sk_node=sk_node, gek=gek, cache=index_cache)
+    await first.initial_scan()
+    old_id = first.index.entries[0].id
+
+    # Change both content and mtime, as any real edit would.
+    f.write_bytes(b"a completely different, longer payload")
+    os.utime(f, (time.time() + 5, time.time() + 5))
+
+    second = DirectoryIndexer(roots=one_root(d), group_id="g",
+                              sk_node=sk_node, gek=gek, cache=index_cache)
+    await second.initial_scan()
+
+    assert second.index.count == 1
+    assert second.index.entries[0].id != old_id
+
+
+@pytest.mark.asyncio
+async def test_scan_interrupted_partway_leaves_only_completed_files_cached(
+        tmp_path, sk_node, gek, index_cache):
+    """
+    A cache row is only ever written after a file is fully hashed (cache.py),
+    so a crash mid-scan cannot leave a stale/partial row — the next scan just
+    treats the not-yet-cached files as new, and finishes the job.
+    """
+    d = tmp_path / "shared"
+    d.mkdir()
+    names = [f"file{i}.bin" for i in range(5)]
+    for i, name in enumerate(names):
+        (d / name).write_bytes(os.urandom(64) * (i + 1))
+
+    real_scan_file = indexer_mod._scan_file
+    hashed_before_crash = []
+
+    def crash_after_three(root, path):
+        if len(hashed_before_crash) >= 3:
+            raise RuntimeError("simulated crash mid-scan")
+        entry = real_scan_file(root, path)
+        hashed_before_crash.append(path)
+        return entry
+
+    indexer_mod._scan_file = crash_after_three
+    try:
+        crashing = DirectoryIndexer(roots=one_root(d), group_id="g",
+                                    sk_node=sk_node, gek=gek, cache=index_cache)
+        with pytest.raises(RuntimeError):
+            await crashing.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert len(hashed_before_crash) == 3
+
+    # A normal rescan (same cache) must still end up with all 5 files
+    # correctly indexed, hashing only the ones the crash never got to.
+    calls = []
+
+    def spy(root, path):
+        calls.append(path)
+        return real_scan_file(root, path)
+
+    indexer_mod._scan_file = spy
+    try:
+        resumed = DirectoryIndexer(roots=one_root(d), group_id="g",
+                                   sk_node=sk_node, gek=gek, cache=index_cache)
+        await resumed.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert resumed.index.count == 5
+    assert len(calls) == 2, f"expected only the 2 not-yet-cached files to be hashed, got {len(calls)}"
+
+
+# ── Progress state ───────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_progress_reflects_bytes_scanned(shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek)
+    assert indexer.progress.scanning is False
+
+    await indexer.initial_scan()
+
+    total_size = sum(f.stat().st_size for f in shared_dir.rglob("*") if f.is_file())
+    assert indexer.progress.scanning is False, "must end idle, not stuck scanning"
+    assert indexer.progress.scanned_bytes == total_size
+    assert indexer.progress.total_bytes == total_size
+
+
+@pytest.mark.asyncio
+async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek):
+    d = tmp_path / "shared"
+    d.mkdir()
+    (d / "a.bin").write_bytes(os.urandom(64))
+    (d / "b.bin").write_bytes(os.urandom(64))
+
+    real_scan_file = indexer_mod._scan_file
+
+    def boom(root, path):
+        raise RuntimeError("simulated failure mid-scan")
+
+    indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
+                               sk_node=sk_node, gek=gek)
+    indexer_mod._scan_file = boom
+    try:
+        with pytest.raises(RuntimeError):
+            await indexer.initial_scan()
+    finally:
+        indexer_mod._scan_file = real_scan_file
+
+    assert indexer.progress.scanning is False, \
+        "an exception mid-scan must not leave the scanning flag stuck on"
+
+
+# ── Off-loop directory walks, reconcile backoff ─────────────────────────────
+
+@pytest.mark.asyncio
+async def test_walk_root_does_not_stall_the_event_loop(tmp_path, sk_node, gek):
+    d = tmp_path / "shared"
+    d.mkdir()
+    (d / "f.bin").write_bytes(b"x")
+
+    real_walk = indexer_mod._walk_root
+
+    def slow_walk(root):
+        time.sleep(0.2)
+        return real_walk(root)
+
+    indexer_mod._walk_root = slow_walk
+    ticks = 0
+
+    async def ticker():
+        nonlocal ticks
+        while True:
+            await asyncio.sleep(0.01)
+            ticks += 1
+
+    ticker_task = asyncio.create_task(ticker())
+    try:
+        indexer = DirectoryIndexer(roots=one_root(d), group_id="g",
+                                   sk_node=sk_node, gek=gek)
+        await indexer.initial_scan()
+    finally:
+        indexer_mod._walk_root = real_walk
+        ticker_task.cancel()
+        try:
+            await ticker_task
+        except asyncio.CancelledError:
+            pass
+
+    assert ticks >= 5, (
+        "the event loop must keep running other tasks while a directory "
+        f"walk is in progress in the executor — only {ticks} ticks happened "
+        "during a 0.2s walk")
+
+
+@pytest.mark.asyncio
+async def test_reconcile_backoff_grows_with_no_changes_then_caps(shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek, reconcile_secs=0.01)
+    await indexer.initial_scan()
+    assert indexer._reconcile_delay == 0.01
+
+    task = asyncio.create_task(indexer._reconcile_loop())
+    try:
+        await asyncio.sleep(0.2)
+        assert indexer._reconcile_delay > 0.01, \
+            "several no-change ticks must have grown the delay"
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    # A low, instance-only cap so the clamp is observable without waiting
+    # through dozens of real doublings up to the real 7200s ceiling.
+    indexer.RECONCILE_BACKOFF_CAP = 0.05
+    indexer._reconcile_delay = 0.04
+    task = asyncio.create_task(indexer._reconcile_loop())
+    try:
+        await asyncio.sleep(0.15)
+        assert indexer._reconcile_delay <= 0.05, "delay must never exceed the cap"
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+
+@pytest.mark.asyncio
+async def test_note_activity_resets_backoff(shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek, reconcile_secs=10.0)
+    await indexer.initial_scan()
+    indexer._reconcile_delay = 5000.0  # simulate a long-idle backoff
+
+    indexer.note_activity()
+
+    assert indexer._reconcile_delay == 10.0
+
+
+@pytest.mark.asyncio
+async def test_reconcile_backoff_resets_when_something_actually_changes(
+        shared_dir, sk_node, gek):
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek, reconcile_secs=0.02)
+    await indexer.initial_scan()
+    indexer._reconcile_delay = 5.0  # pretend it had already backed off a lot
+
+    # A fake reconcile() rather than a real filesystem change: the real
+    # sweep's timing (disk I/O, the executor round trip) would race against
+    # this test's own sleeps. What matters here is only _reconcile_loop's
+    # reaction to "something changed", not reconcile()'s own detection logic
+    # — that is covered separately (test_root_availability.py).
+    reconciled_once = asyncio.Event()
+
+    async def fake_reconcile():
+        reconciled_once.set()
+        return True
+
+    indexer._reconcile_delay = 0.01
+    indexer.reconcile = fake_reconcile
+
+    task = asyncio.create_task(indexer._reconcile_loop())
+    try:
+        await asyncio.wait_for(reconciled_once.wait(), timeout=2.0)
+        assert indexer._reconcile_delay == 0.02, \
+            "a real change must reset the delay back to the base interval"
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py
index d2ccc0d..83758ae 100644
--- a/packages/meshbay-node/tests/test_ops.py
+++ b/packages/meshbay-node/tests/test_ops.py
@@ -9,6 +9,7 @@ appeared: the adapters must be thin, and the operations must not decide who may
 call them.
 """
 
+import asyncio
 import inspect
 from pathlib import Path
 
@@ -221,3 +222,32 @@ async def test_reload_config_without_fn_is_refused(tmp_path):
     state = _state(tmp_path)
     with pytest.raises(ops.OpError, match="Reload not available"):
         await ops.reload_config(state)
+
+
+async def test_start_reload_returns_before_reload_fn_finishes(tmp_path):
+    """The loopback route uses this one: a brand-new group's initial scan
+    can take minutes, and the Electron bridge caps every loopback call at
+    30s (main.js node:call) — start_reload must not block on it."""
+    state = _state(tmp_path)
+    release = asyncio.Event()
+    called = []
+
+    async def slow_reload():
+        await release.wait()
+        called.append(True)
+    state["reload_fn"] = slow_reload
+
+    out = await asyncio.wait_for(ops.start_reload(state), timeout=1.0)
+
+    assert out["status"] == "reloading"
+    assert not called, "start_reload must return before reload_fn finishes"
+
+    release.set()
+    await asyncio.sleep(0)  # let the still-running reload_fn task complete
+    assert called, "reload_fn must still actually run, just not be waited on"
+
+
+async def test_start_reload_without_fn_is_refused(tmp_path):
+    state = _state(tmp_path)
+    with pytest.raises(ops.OpError, match="Reload not available"):
+        await ops.start_reload(state)
diff --git a/packages/meshbay-node/tests/test_scan_settings_policy.py b/packages/meshbay-node/tests/test_scan_settings_policy.py
new file mode 100644
index 0000000..719b988
--- /dev/null
+++ b/packages/meshbay-node/tests/test_scan_settings_policy.py
@@ -0,0 +1,205 @@
+"""
+The operator can tune how often the indexer's reconciliation backstop runs,
+and how long it waits after a file's last write before hashing it.
+
+Same shape as test_apps_enabled_policy.py / test_member_upload_policy.py:
+changed by a signed operator instruction, stored on the node rather than the
+hub. Unlike those two, there is also a *live* DirectoryIndexer object to
+update — see test_set_scan_settings_updates_the_live_indexer below.
+"""
+
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_SET_SCAN_SETTINGS
+from meshbay_common.crypto import generate_gek
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture
+def gek():
+    return generate_gek()
+
+
+@pytest.fixture
+def shared_dir(tmp_path):
+    d = tmp_path / "shared"
+    d.mkdir()
+    (d / "video.mkv").write_bytes(os.urandom(256))
+    return d
+
+
+def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": one_root(shared_root),
+        "index": index,
+        "sk_node": index.sk_node,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = None
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+# ── Refused before a challenge is even issued ───────────────────────────────
+
+async def test_out_of_range_reconcile_interval_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 1.0, "debounce_secs": 2.0})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_out_of_range_debounce_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 600.0, "debounce_secs": 99999.0})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_non_numeric_values_are_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": "not-a-number", "debounce_secs": 2.0})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    session = _session(tmp_path, "member-1", operator="the-operator")
+    session._has_admin_authority = lambda: False
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Who may change it ───────────────────────────────────────────────────────
+
+async def test_changing_it_needs_a_signature(tmp_path):
+    """The request only ever produces a challenge — nothing is applied
+    until a signature over the transcript verifies."""
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_set_scan_settings(
+        {"reconcile_interval_secs": 600.0, "debounce_secs": 2.0})
+
+    assert issued == [(OP_SET_SCAN_SETTINGS, "600,2")]
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        defaults = await roster.scan_settings("g1")
+        assert defaults == {
+            "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+            "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+        }, "unset must mean the indexer's own defaults, or an upgrade " \
+           "changes behaviour for every existing group"
+
+        await roster.set_scan_settings("g1", 1200.0, 5.0, set_by="op")
+        assert await roster.scan_settings("g1") == {
+            "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0}
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.scan_settings("g1") == {
+            "reconcile_interval_secs": 1200.0, "debounce_secs": 5.0}
+        assert await reopened.scan_settings("g2") == {
+            "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
+            "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
+        }, "one group's setting must not answer for another"
+    finally:
+        await reopened.close()
+
+
+# ── Applying it to the live indexer ─────────────────────────────────────────
+
+async def test_set_scan_settings_updates_the_live_indexer(tmp_path, shared_dir, gek):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    indexer = DirectoryIndexer(
+        roots=one_root(shared_dir), group_id="g1",
+        sk_node=Ed25519PrivateKey.generate(), gek=gek)
+    await indexer.initial_scan()
+    indexer._reconcile_delay = 5000.0  # simulate a long-idle backoff
+    state = {"roster": roster, "indexers": {"g1": indexer}}
+
+    try:
+        result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0)
+
+        assert result == {"reconcile_interval_secs": 1800.0, "debounce_secs": 3.0,
+                          "group_id": "g1"}
+        assert indexer.reconcile_secs == 1800.0
+        assert indexer.debounce_secs == 3.0
+        assert indexer._reconcile_delay == 1800.0, \
+            "the new interval must apply right away, not after whatever " \
+            "backoff had already stretched the wait to"
+        assert await roster.scan_settings("g1") == {
+            "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0}
+    finally:
+        await roster.close()
+
+
+async def test_set_scan_settings_without_a_live_indexer_still_persists(tmp_path):
+    """A group hosted on the node but with no running indexer in this
+    process (e.g. a test, or a group not yet hot-loaded) must not crash —
+    the setting still lands in roster.db for whenever it is."""
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state = {"roster": roster, "indexers": {}}
+
+    try:
+        result = await ops.set_scan_settings(state, "g1", 1800.0, 3.0)
+        assert result["reconcile_interval_secs"] == 1800.0
+        assert await roster.scan_settings("g1") == {
+            "reconcile_interval_secs": 1800.0, "debounce_secs": 3.0}
+    finally:
+        await roster.close()