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, 88 insertions, 3 deletions

diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py
index 52cd78b..df51e85 100644
--- a/packages/meshbay-node/tests/test_index_progress.py
+++ b/packages/meshbay-node/tests/test_index_progress.py
@@ -36,19 +36,26 @@ def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" *
 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}
+        "scanning": False, "scanned_bytes": 0, "total_bytes": 0,
+        "files_done": 0, "files_total": 0, "kind": "", "root_pos": -1, "queued": 0}
 
 
 def test_indexing_status_reflects_live_progress():
     progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000,
-                             current_dir="StarWars")
+                             current_dir="Season 2", root="series", root_pos=1,
+                             kind="scan", files_done=3, files_total=9,
+                             queued=["archive", "photos"])
     session = _session_with_progress(progress)
 
     status = session._indexing_status()
 
-    assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000}
+    assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000,
+                      "files_done": 3, "files_total": 9, "kind": "scan",
+                      "root_pos": 1, "queued": 2}
     assert "current_dir" not in status, \
         "the directory name is operator-local detail, never sent to a member"
+    assert not {"series", "archive", "photos"} & {str(v) for v in status.values()}, \
+        "a root name reached a member"
 
 
 # ── /api/groups/{id}/index-status (loopback) ────────────────────────────────
@@ -81,6 +88,36 @@ def test_index_status_route_reflects_indexer_progress():
     }
 
 
+# ── /api/index-status (loopback, every group) ───────────────────────────────
+
+def test_every_group_is_described_including_one_still_being_attached():
+    """The band reads one route for the whole node. A group in its initial scan
+    is in state["indexers"] and not yet in groups_ctx, and must be there."""
+    busy, idle = MagicMock(), MagicMock()
+    busy.progress = IndexProgress(
+        scanning=True, scanned_bytes=10, total_bytes=40, current_dir="2024",
+        root="results", root_pos=1, kind="scan", files_done=1, files_total=4,
+        queued=["archive"])
+    idle.progress = IndexProgress()
+    config = MagicMock()
+    named = MagicMock()
+    named.id, named.name = "a" * 32, "outputs"
+    config.groups = [named]
+    client = _ui_client({"config": config,
+                         "indexers": {"a" * 32: busy, "b" * 32: idle}})
+
+    groups = {g["group_id"]: g for g in client.get("/api/index-status").json()["groups"]}
+
+    assert groups["a" * 32] == {
+        "group_id": "a" * 32, "group_name": "outputs", "scanning": True,
+        "kind": "scan", "root": "results", "current_dir": "2024",
+        "scanned_bytes": 10, "total_bytes": 40, "files_done": 1, "files_total": 4,
+        "queued": ["archive"],
+    }
+    assert groups["b" * 32]["group_name"] == "b" * 8
+    assert groups["b" * 32]["scanning"] is False
+
+
 # ── _push_index_progress / _progress_pusher ─────────────────────────────────
 
 def _daemon(tmp_path) -> NodeDaemon:
@@ -123,6 +160,54 @@ async def test_push_index_progress_only_reaches_same_group_peers(tmp_path):
 
 
 @pytest.mark.asyncio
+async def test_push_index_progress_carries_counters_never_root_names(tmp_path):
+    daemon = _daemon(tmp_path)
+    session = MagicMock()
+    session._group_id = "a" * 32
+    mock_webrtc = MagicMock()
+    mock_webrtc._sessions = {"p1": session}
+    daemon._webrtc = mock_webrtc
+
+    daemon._push_index_progress("a" * 32, IndexProgress(
+        scanning=True, scanned_bytes=10, total_bytes=100, current_dir="2024",
+        root="results", root_pos=2, kind="rescan", files_done=5, files_total=50,
+        queued=["archive", "photos"]))
+
+    msg = session._send.call_args[0][0]
+    assert (msg["kind"], msg["root_pos"], msg["queued"]) == ("rescan", 2, 2)
+    assert (msg["files_done"], msg["files_total"]) == (5, 50)
+    assert not {"results", "archive", "photos", "2024"} & {str(v) for v in msg.values()}
+
+
+@pytest.mark.asyncio
+async def test_progress_pusher_speaks_while_a_root_only_waits(tmp_path):
+    """Between one root's scan ending and the next one taking the lock, nothing
+    is scanning — but there is work coming, and the band must not drop it."""
+    daemon = _daemon(tmp_path)
+    session = MagicMock()
+    session._group_id = "a" * 32
+    mock_webrtc = MagicMock()
+    mock_webrtc._sessions = {"p1": session}
+    daemon._webrtc = mock_webrtc
+
+    indexer = MagicMock()
+    indexer.group_id = "a" * 32
+    indexer.progress = IndexProgress(scanning=False, queued=["archive"])
+
+    task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05))
+    try:
+        await asyncio.sleep(0.12)
+        assert session._send.call_count >= 2
+        assert session._send.call_args.args[0]["queued"] == 1
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+
+@pytest.mark.asyncio
 async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path):
     daemon = _daemon(tmp_path)