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

4 files changed, 151 insertions, 36 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 932a90d..a43e3a3 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -1163,14 +1163,16 @@ class NodeDaemon:
         parameter (not a bare constant) only so a test can drive this loop
         without waiting on the real 2s cadence.
         """
-        was_scanning = False
+        was_busy = False
         while True:
             await asyncio.sleep(interval)
             progress = indexer.progress
-            now_scanning = progress.scanning
-            if now_scanning or was_scanning:
+            # A root waiting for the scan lock is work the operator is owed a
+            # sight of, even in the gap where nothing is walking yet.
+            busy = progress.scanning or bool(progress.queued)
+            if busy or was_busy:
                 self._push_index_progress(indexer.group_id, progress)
-            was_scanning = now_scanning
+            was_busy = busy
 
     def _push_index_progress(self, group_id: str, progress) -> None:
         if not self._webrtc:
@@ -1181,7 +1183,9 @@ class NodeDaemon:
         # scan. Sealing it would buy an attacker's rough estimate of a library's
         # size and cost a key derivation and a decrypt per push. If a field that
         # names anything is ever added here, that trade is void and this message
-        # joins the other two.
+        # joins the other two. `kind` is one of four fixed words; the root under
+        # way is a position in the roots table the member already opened from
+        # the sealed index, and the roots waiting are a count, never names.
         msg = {
             "type": MNP.INDEX_PROGRESS,
             "v": MNP_VERSION,
@@ -1189,6 +1193,11 @@ class NodeDaemon:
             "scanning": progress.scanning,
             "scanned_bytes": progress.scanned_bytes,
             "total_bytes": progress.total_bytes,
+            "files_done": progress.files_done,
+            "files_total": progress.files_total,
+            "kind": progress.kind,
+            "root_pos": progress.root_pos,
+            "queued": len(progress.queued),
         }
         pushed = 0
         for session in list(self._webrtc._sessions.values()):
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index db1b11c..c1b151a 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -26,7 +26,7 @@ import asyncio
 import logging
 import time
 from concurrent.futures import ThreadPoolExecutor
-from dataclasses import dataclass
+from dataclasses import dataclass, field as dataclass_field
 from pathlib import Path
 from typing import Callable, Awaitable
 
@@ -141,6 +141,20 @@ class IndexProgress:
     # Basename only, deliberately not the full path — enough to show progress
     # without broadcasting the operator's directory structure.
     current_dir:   str  = ""
+    # What the operator's progress band names. `root` and `queued` are root
+    # names, so they stay on the loopback API like `current_dir`; members get
+    # `root_pos`, a position in the roots table they already opened from the
+    # sealed index, and the length of the queue.
+    root:          str  = ""
+    root_pos:      int  = -1
+    # "scan" (a root walked for the first time), "rescan" (one that came back),
+    # "reconcile" (files the watcher missed), "watch" (a burst of events), or
+    # "" when idle.
+    kind:          str  = ""
+    files_done:    int  = 0
+    files_total:   int  = 0
+    # Roots waiting for the scan lock, in the order they will be walked.
+    queued:        list[str] = dataclass_field(default_factory=list)
 
 
 def _virtual_dir(root: Root, file_path: Path) -> str:
@@ -323,6 +337,8 @@ class DirectoryIndexer:
         self._job_running = False
         self._burst_scanned = 0
         self._burst_total = 0
+        self._burst_files_done = 0
+        self._burst_files_total = 0
         self._burst_dir = ""
         # Ids whose entry this indexer threw away and rebuilt from disk, since
         # the last time a consumer drained this. A rebuilt entry carries only
@@ -373,20 +389,27 @@ class DirectoryIndexer:
     async def _initial_scan(self) -> None:
         self.roots.refresh_availability()
         total = 0
-        for root in self.roots:
-            if not root.available:
-                log.warning("Root %r is not readable at startup (%s) — its files "
-                            "are not indexed yet and will appear when it returns",
-                            root.name, root.path)
-                continue
-            total += await self._scan_root(root)
+        waiting = [r.name for r in self.roots if r.available]
+        self._queue(waiting)
+        try:
+            for root in self.roots:
+                if not root.available:
+                    log.warning("Root %r is not readable at startup (%s) — its files "
+                                "are not indexed yet and will appear when it returns",
+                                root.name, root.path)
+                    continue
+                waiting.remove(root.name)
+                self._unqueue([root.name])
+                total += await self._scan_root(root)
+        finally:
+            self._unqueue(waiting)
         self._index.version = int(time.time())
         self._index.roots = self.roots.describe()
         self._report_collisions()
         log.info("Initial scan complete: %d files across %d root(s)",
                  total, len(self.roots))
 
-    async def _scan_root(self, root: Root) -> int:
+    async def _scan_root(self, root: Root, *, kind: str = "scan") -> int:
         log.info("Scanning %s (root %r) ...", root.path, root.name)
         count = 0
         loop = asyncio.get_event_loop()
@@ -402,7 +425,7 @@ class DirectoryIndexer:
         # had not yet turned on, even though the node was already several
         # seconds into walking and sizing a large root — total_bytes is
         # unknown at this point, so it starts at 0 and is corrected below.
-        self._begin_job(root.name)
+        self._begin_job(root.name, root=root, kind=kind)
         try:
             try:
                 files = await loop.run_in_executor(self._executor, _walk_root, root)
@@ -418,11 +441,13 @@ class DirectoryIndexer:
             sized = await loop.run_in_executor(self._executor, _size_files, files)
 
             self.progress.total_bytes = sum(size for _, size in sized)
+            self.progress.files_total = len(sized)
             self.progress.current_dir = ""
             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
+                self.progress.files_done += 1
                 if entry:
                     self._index.add_entry(entry)
                     count += 1
@@ -433,18 +458,30 @@ class DirectoryIndexer:
             self._end_job()
         return count
 
-    def _begin_job(self, current_dir: str, total_bytes: int = 0) -> None:
+    def _begin_job(self, current_dir: str, total_bytes: int = 0, *,
+                   root: Root, kind: str, files_total: int = 0) -> None:
         """`progress` now describes a whole-root walk, whatever a burst is doing."""
         self._job_running = True
-        self.progress.scanning = True
-        self.progress.scanned_bytes = 0
-        self.progress.total_bytes = total_bytes
-        self.progress.current_dir = current_dir
+        p = self.progress
+        p.scanning = True
+        p.scanned_bytes = 0
+        p.total_bytes = total_bytes
+        p.files_done = 0
+        p.files_total = files_total
+        p.current_dir = current_dir
+        p.kind = kind
+        p.root = root.name
+        p.root_pos = next((i for i, r in enumerate(self.roots)
+                           if r.folded == root.folded), -1)
 
     def _end_job(self) -> None:
         self._job_running = False
-        self.progress.scanning = False
-        self.progress.current_dir = ""
+        p = self.progress
+        p.scanning = False
+        p.current_dir = ""
+        p.kind = ""
+        p.root = ""
+        p.root_pos = -1
         if self._burst_inflight > 0:
             # A burst that started during the walk is still hashing.
             self._show_burst()
@@ -456,11 +493,25 @@ class DirectoryIndexer:
         if self._burst_inflight > 0:
             p.scanning = True
             p.current_dir = self._burst_dir
+            p.kind = "watch"
+            p.root = ""
+            p.root_pos = -1
         elif not self._pending_timers:
             p.scanning = False
             p.current_dir = ""
+            p.kind = ""
         p.scanned_bytes = self._burst_scanned
         p.total_bytes = self._burst_total
+        p.files_done = self._burst_files_done
+        p.files_total = self._burst_files_total
+
+    def _queue(self, names: list[str]) -> None:
+        self.progress.queued.extend(names)
+
+    def _unqueue(self, names: list[str]) -> None:
+        for name in names:
+            if name in self.progress.queued:
+                self.progress.queued.remove(name)
 
     async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None:
         """
@@ -633,6 +684,7 @@ class DirectoryIndexer:
         """
         old_names = {r.folded for r in self.roots}
         new_names = {r.folded for r in roots}
+        self.progress.queued[:] = [n for n in self.progress.queued if fold(n) in new_names]
 
         for root in self.roots:
             if root.folded not in new_names:
@@ -655,6 +707,7 @@ class DirectoryIndexer:
                 await self.on_change(self)
             return
 
+        self._queue([r.name for r in added])
         task = asyncio.create_task(self._scan_added_roots(added))
         self._scan_tasks.add(task)
         task.add_done_callback(self._scan_tasks.discard)
@@ -668,9 +721,12 @@ class DirectoryIndexer:
         return any(r.folded == root.folded and r.path == root.path for r in self.roots)
 
     async def _scan_added_roots(self, added: list[Root]) -> None:
+        waiting = [r.name for r in added]
         try:
             async with self._scan_lock:
                 for root in added:
+                    waiting.remove(root.name)
+                    self._unqueue([root.name])
                     # A later retarget may have removed it while this waited.
                     if not self._holds(root):
                         continue
@@ -689,6 +745,8 @@ class DirectoryIndexer:
             raise
         except Exception:
             log.exception("Scanning the added root(s) failed")
+        finally:
+            self._unqueue(waiting)
 
     # ── Reconciliation ────────────────────────────────────────────────────────
 
@@ -818,12 +876,14 @@ class DirectoryIndexer:
                 except OSError:
                     added_sized.append((p, 0))
 
-            self._begin_job("", sum(size for _, size in added_sized))
+            self._begin_job("", sum(size for _, size in added_sized), root=root,
+                            kind="reconcile", files_total=len(added_sized))
             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
+                    self.progress.files_done += 1
                     if not entry:
                         continue
                     # The index is keyed by **content hash**, so two identical
@@ -883,7 +943,7 @@ class DirectoryIndexer:
         """
         carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)}
         self._drop_root_entries(root)
-        count = await self._scan_root(root)
+        count = await self._scan_root(root, kind="rescan")
         for entry in self._entries_under(root):
             old = carried.get((entry.id, entry.name, entry.path))
             if old is None:
@@ -1002,12 +1062,20 @@ class DirectoryIndexer:
             # A whole-root walk like any other, so it waits its turn: run beside
             # an added root's scan, the two reset each other's progress and read
             # the same drive in alternation.
-            async with self._scan_lock:
-                # Ejected or removed again while it waited. `_rescan_root` drops
-                # the entries before it walks, so going ahead would empty a root
-                # that is not there to be read.
-                if self._holds(root) and not root.ejected and root.is_live():
-                    await self._rescan_root(root)
+            self._queue([root.name])
+            waiting = True
+            try:
+                async with self._scan_lock:
+                    self._unqueue([root.name])
+                    waiting = False
+                    # Ejected or removed again while it waited. `_rescan_root`
+                    # drops the entries before it walks, so going ahead would
+                    # empty a root that is not there to be read.
+                    if self._holds(root) and not root.ejected and root.is_live():
+                        await self._rescan_root(root)
+            finally:
+                if waiting:
+                    self._unqueue([root.name])
         self._restart_observer()
         self._index.roots = self.roots.describe()
         self._index.version = int(time.time())
@@ -1044,7 +1112,10 @@ class DirectoryIndexer:
             if self._burst_inflight <= 0:
                 self._burst_scanned = 0
                 self._burst_total = 0
+                self._burst_files_done = 0
+                self._burst_files_total = 0
             self._burst_total += size
+            self._burst_files_total += 1
             self._burst_dir = file_path.parent.name
             self._burst_sizes[key] = size
             self._burst_inflight += 1
@@ -1118,6 +1189,7 @@ class DirectoryIndexer:
             size = self._burst_sizes.pop(str(file_path), None)
             if size is not None:
                 self._burst_scanned += size
+                self._burst_files_done += 1
                 self._burst_inflight -= 1
                 self._show_burst()
 
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 eff40f1..79a97cc 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -3268,19 +3268,25 @@ class WebRTCPeerSession:
 
     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.
+        The counters of `_push_index_progress` (daemon.py), for the handshake
+        ack — never a path, a filename or a root name, which stay 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": False, "scanned_bytes": 0, "total_bytes": 0,
+                    "files_done": 0, "files_total": 0, "kind": "",
+                    "root_pos": -1, "queued": 0}
         return {
             "scanning": progress.scanning,
             "scanned_bytes": progress.scanned_bytes,
             "total_bytes": progress.total_bytes,
+            "files_done": progress.files_done,
+            "files_total": progress.files_total,
+            "kind": progress.kind,
+            "root_pos": progress.root_pos,
+            "queued": len(progress.queued),
         }
 
     # ── Transfer slots ───────────────────────────────────────────────────────
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index de2542e..22b5afb 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -437,6 +437,34 @@ def create_ui_app(state: dict) -> FastAPI:
             "current_dir": progress.current_dir,
         }
 
+    @app.get("/api/index-status")
+    async def index_status_all():
+        """
+        Every group's indexing at once, for the client's progress band — which
+        is on screen whatever page the operator is on, so it cannot ask per
+        group. Names roots, like `current_dir` above: loopback only, the
+        operator's own screen. Reads state["indexers"] for the same reason.
+        """
+        config = state.get("config")
+        names = {g.id: g.name for g in config.groups} if config else {}
+        groups = []
+        for gid, indexer in list(state.get("indexers", {}).items()):
+            p = indexer.progress
+            groups.append({
+                "group_id": gid,
+                "group_name": names.get(gid, gid[:8]),
+                "scanning": p.scanning,
+                "kind": p.kind,
+                "root": p.root,
+                "current_dir": p.current_dir,
+                "scanned_bytes": p.scanned_bytes,
+                "total_bytes": p.total_bytes,
+                "files_done": p.files_done,
+                "files_total": p.files_total,
+                "queued": list(p.queued),
+            })
+        return {"groups": groups}
+
     # ── Enabled apps (operator only, localhost) ────────────────────────────
     #
     # Same loopback shape as member-upload: the Create Group wizard sets this