feat(node): progress names the root under way and the roots waiting

`IndexProgress` said "scanning, this many bytes of that many" and nothing
more. A group's roots are walked one after another, so a second directory
added during a large scan showed as the bar jumping back to 0 %. It now also
carries the root being walked and its position in the roots table, the kind
of walk (scan, rescan, reconcile, watch), file counts, and the roots
waiting for the scan lock in order: queued by the initial scan, by a
retarget, and by a plug; dropped when a root is removed.

`GET /api/index-status` answers for every group at once, including a group
still in its initial scan, so a client can show indexing on any page. It
names roots: loopback only, like `current_dir`.

`index_progress` and the handshake ack gain the same counters, still naming
nothing (decision D3): the root is a position in the roots table the member
already opened from the sealed index, and the queue is a count. The pusher
keeps speaking while a root only waits for the lock.

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

1 files changed, 207 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_index_jobs_are_described.py b/packages/meshbay-node/tests/test_index_jobs_are_described.py
new file mode 100644
index 0000000..dee6e39
--- /dev/null
+++ b/packages/meshbay-node/tests/test_index_jobs_are_described.py
@@ -0,0 +1,207 @@
+"""
+What the operator's progress band is told about the indexing under way.
+
+`progress` said "scanning, this many bytes of that many" and nothing more. That
+is one bar with no name on it. A node asked to add a second directory while the
+first is still hashing does them one after the other — one scan lock, one
+hashing thread — and an operator looking at a bar that jumps back to 0 % cannot
+tell a second directory from a scan that started over. So it now says which
+root, what kind of walk, how many files, and which roots wait their turn.
+"""
+
+import asyncio
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+def _set(*specs) -> RootSet:
+    return RootSet.build([s if isinstance(s, dict) else {"path": str(s), "name": s.name}
+                          for s in specs])
+
+
+def _names(idx) -> list[str]:
+    return sorted(e.name for e in idx.index.entries)
+
+
+async def _until(predicate, timeout: float = 3.0) -> bool:
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not predicate():
+        if asyncio.get_running_loop().time() > deadline:
+            return False
+        await asyncio.sleep(0.02)
+    return True
+
+
+class _Held(DirectoryIndexer):
+    """Stops before hashing the first file under `hold_under`, and records walks."""
+    hold_under: Path | None = None
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.gate = asyncio.Event()
+        self.at_gate = asyncio.Event()
+        self.walked: list[str] = []
+
+    async def _scan_root(self, root, *args, **kwargs):
+        self.walked.append(root.name)
+        return await super()._scan_root(root, *args, **kwargs)
+
+    async def _hash_or_cached(self, root, file_path):
+        if self.hold_under is not None and file_path.is_relative_to(self.hold_under):
+            self.at_gate.set()
+            await self.gate.wait()
+        return await super()._hash_or_cached(root, file_path)
+
+
+def _dirs(tmp_path: Path) -> tuple[Path, Path, Path]:
+    one, r1, r2 = tmp_path / "one", tmp_path / "r1", tmp_path / "r2"
+    for d in (one, r1, r2):
+        d.mkdir()
+    (one / "a.txt").write_bytes(b"first root")
+    (r1 / "b.bin").write_bytes(os.urandom(2000))
+    (r1 / "c.bin").write_bytes(os.urandom(3000))
+    (r2 / "d.bin").write_bytes(os.urandom(4000))
+    return one, r1, r2
+
+
+async def _drain(idx) -> None:
+    await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
+
+
+async def test_the_root_under_way_and_the_ones_waiting_are_named(tmp_path):
+    one, r1, r2 = _dirs(tmp_path)
+    idx = _Held(roots=_set(one), group_id=GROUP,
+                sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    idx.walked.clear()
+    idx.hold_under = r1
+    try:
+        await idx.retarget(_set(one, r1), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 5)
+        p = idx.progress
+        assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "scan", "r1", 1)
+        assert (p.files_done, p.files_total, p.total_bytes) == (0, 2, 5000)
+        assert p.queued == []
+
+        await idx.retarget(_set(one, r1, r2), wait=False)
+        assert p.queued == ["r2"]
+
+        idx.gate.set()
+        await _drain(idx)
+        assert idx.walked == ["r1", "r2"], "the second root did not wait for the first"
+        assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", [])
+        assert _names(idx) == ["a.txt", "b.bin", "c.bin", "d.bin"]
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_a_root_removed_while_it_waited_leaves_the_queue(tmp_path):
+    one, r1, r2 = _dirs(tmp_path)
+    idx = _Held(roots=_set(one), group_id=GROUP,
+                sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    idx.walked.clear()
+    idx.hold_under = r1
+    try:
+        await idx.retarget(_set(one, r1), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 5)
+        await idx.retarget(_set(one, r1, r2), wait=False)
+        await idx.retarget(_set(one, r1), wait=False)
+        assert idx.progress.queued == [], "a removed root is still announced as next"
+
+        idx.gate.set()
+        await _drain(idx)
+        assert idx.walked == ["r1"]
+        assert idx.progress.queued == []
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_the_initial_scan_names_the_roots_still_to_come(tmp_path):
+    _, r1, r2 = _dirs(tmp_path)
+    idx = _Held(roots=_set(r1, r2), group_id=GROUP,
+                sk_node=Ed25519PrivateKey.generate(), gek=None)
+    idx.hold_under = r1
+    scan = asyncio.create_task(idx.initial_scan())
+    try:
+        await asyncio.wait_for(idx.at_gate.wait(), 5)
+        assert (idx.progress.root, idx.progress.queued) == ("r1", ["r2"])
+        idx.gate.set()
+        await asyncio.wait_for(scan, 5)
+        assert idx.progress.queued == []
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_a_failed_initial_scan_leaves_nothing_announced(tmp_path):
+    _, r1, r2 = _dirs(tmp_path)
+
+    class _Broken(DirectoryIndexer):
+        async def _hash_or_cached(self, root, file_path):
+            raise RuntimeError("simulated failure mid-scan")
+
+    idx = _Broken(roots=_set(r1, r2), group_id=GROUP,
+                  sk_node=Ed25519PrivateKey.generate(), gek=None)
+    with pytest.raises(RuntimeError):
+        await idx.initial_scan()
+    p = idx.progress
+    assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", [])
+
+
+async def test_a_plug_waiting_its_turn_is_announced(tmp_path):
+    one, r1, _ = _dirs(tmp_path)
+    two = tmp_path / "two"
+    two.mkdir()
+    (two / "e.txt").write_bytes(b"removable")
+    idx = _Held(roots=_set(one, two), group_id=GROUP,
+                sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    idx.eject_root("two")
+    idx.hold_under = r1
+    try:
+        await idx.retarget(
+            _set(one, {"path": str(two), "name": "two", "ejected": True}, r1), wait=False)
+        await asyncio.wait_for(idx.at_gate.wait(), 5)
+        plug = asyncio.create_task(idx.plug_root("two"))
+        assert await _until(lambda: idx.progress.queued == ["two"])
+
+        idx.gate.set()
+        await asyncio.wait_for(plug, 5)
+        assert idx.progress.queued == []
+        assert idx.walked[-1] == "two"
+    finally:
+        idx.gate.set()
+        await idx.stop()
+
+
+async def test_a_burst_is_described_as_watching(tmp_path):
+    one, _, _ = _dirs(tmp_path)
+    idx = DirectoryIndexer(roots=_set(one), group_id=GROUP,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None,
+                           debounce_secs=0.01)
+    await idx.initial_scan()
+    idx._loop = asyncio.get_running_loop()
+    for i in range(2):
+        f = one / f"new{i}.bin"
+        f.write_bytes(os.urandom(1000))
+        idx._schedule_update(f)
+    await asyncio.sleep(0)
+
+    p = idx.progress
+    assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "watch", "", -1)
+    assert (p.files_done, p.files_total) == (0, 2)
+
+    assert await _until(lambda: not p.scanning)
+    assert (p.kind, p.files_done, p.files_total) == ("", 2, 2)