fix(cli): apply index updates in place instead of fetching the whole index

The interactive mode fetched and re-parsed the full index on every
index_delta, on the loop receiving the download: 67 times in 80 s while
a node enriched a library.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

4 files changed, 126 insertions, 2 deletions

diff --git a/packages/meshbay-cli/src/meshbay_cli/client/tree.py b/packages/meshbay-cli/src/meshbay_cli/client/tree.py
index 6e537ef..dd3acc8 100644
--- a/packages/meshbay-cli/src/meshbay_cli/client/tree.py
+++ b/packages/meshbay-cli/src/meshbay_cli/client/tree.py
@@ -57,6 +57,39 @@ class Tree:
         dirs |= set(roots)
         return cls(roots, dirs, files)
 
+    def apply_delta(self, delta: dict) -> bool:
+        """
+        Bring the tree up to an `index_delta`, in place. True when the delta
+        changed what the tree holds — a file added or removed, a root's state —
+        rather than only the fields of files already there.
+
+        A delta carries no `dirs`, so an empty folder made elsewhere is not in
+        it: a structural change is where the caller asks for the whole index
+        again, once. The common delta is enrichment filling in a file's fields,
+        about one a second while a library is being enriched, and asking for
+        the whole index on each of those took the download's own event loop:
+        67 full indexes in the 80 s of one download.
+        """
+        updates = {e["id"]: e for e in delta.get("updates", [])}
+        gone = set(delta.get("deletions", []))
+        for parent, entries in self.files.items():
+            if gone or any(e["id"] in updates for e in entries):
+                self.files[parent] = [updates.get(e["id"], e) for e in entries
+                                      if e["id"] not in gone]
+        known = {e["id"] for entries in self.files.values() for e in entries}
+        added = [e for e in delta.get("additions", []) if e["id"] not in known]
+        for e in added:
+            self.files.setdefault(e["path"], []).append(e)
+            parts = e["path"].split("/")
+            for i in range(1, len(parts) + 1):
+                self.dirs.add("/".join(parts[:i]))
+        roots = {r["name"]: r for r in delta.get("roots", [])}
+        moved_roots = bool(roots) and roots != self.roots
+        if moved_roots:
+            self.roots = roots
+            self.dirs |= set(roots)
+        return bool(gone or added or moved_roots)
+
     def root_available(self, path: str) -> bool:
         root = self.roots.get(path.split("/", 1)[0])
         return bool(root.get("available", True)) if root else True
diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/screens.py b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py
index 39b5df3..b6fd347 100644
--- a/packages/meshbay-cli/src/meshbay_cli/tui/screens.py
+++ b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py
@@ -41,6 +41,9 @@ from .panes import LocalPane, RemotePane
 from .widgets import Choice, TransferStrip
 
 PER_GROUP = 2
+# A burst of index updates that added or removed files is answered by one full
+# index, this long after the last of them.
+REFRESH_AFTER_DELTA_S = 2.0
 
 
 class SignInScreen(Screen):
@@ -118,6 +121,7 @@ class MainScreen(Screen):
         self._trees: dict[str, Tree] = {}
         self._gates: dict[str, asyncio.Semaphore] = {}
         self._watching: set[str] = set()
+        self._refreshes: dict[str, asyncio.TimerHandle] = {}
 
     def compose(self) -> ComposeResult:
         with Horizontal():
@@ -150,12 +154,36 @@ class MainScreen(Screen):
             self._trees[group["id"]] = Tree.from_index(await session.index())
             if group["id"] not in self._watching:
                 self._watching.add(group["id"])
-                session.ch.on_push("index_delta", lambda _msg, g=group: self._changed(g))
+                session.ch.on_push(
+                    "index_delta", lambda msg, g=group, s=session: self._delta(g, s, msg))
         return self._trees[group["id"]]
 
     def forget_tree(self, group: dict) -> None:
         self._trees.pop(group["id"], None)
 
+    def _delta(self, group: dict, session, msg: dict) -> None:
+        """A pushed index update, applied to the tree in place. Only one that
+        adds or removes something is drawn, and asks for the whole index — once
+        for a burst, not once per message."""
+        tree = self._trees.get(group["id"])
+        if tree is None:
+            return   # the next look asks for the whole index anyway
+        try:
+            structural = tree.apply_delta(session.open_delta(msg))
+        except CliError:
+            structural = True
+        if not structural:
+            return
+        pending = self._refreshes.pop(group["id"], None)
+        if pending is not None:
+            pending.cancel()
+        self._refreshes[group["id"]] = asyncio.get_running_loop().call_later(
+            REFRESH_AFTER_DELTA_S, self._refreshed, group)
+
+    def _refreshed(self, group: dict) -> None:
+        self._refreshes.pop(group["id"], None)
+        self._changed(group)
+
     def _changed(self, group: dict) -> None:
         """Something changed in the group: the index is asked for again, and the
         pane redrawn if it is showing that group."""
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
index c618ee4..47f82b3 100644
--- a/packages/meshbay-cli/tests/fake_node.py
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -79,6 +79,8 @@ class FakeNode:
         self.members: dict[str, Identity] = {}   # sender → the device it signs with
         self.unresponsive = False            # pings go unanswered, as a dead link's
         self.admin_ops: dict[str, dict] = {}
+        self.channels: list[FakeChannel] = []
+        self.index_syncs = 0                 # full indexes asked for
 
     def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict:
         fid = hashlib.blake2b(data + path.encode() + name.encode(), digest_size=32).hexdigest()
@@ -92,9 +94,26 @@ class FakeNode:
         return entry
 
     def session(self) -> GroupSession:
-        return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek,
+        ch = FakeChannel(self)
+        self.channels.append(ch)
+        return GroupSession(None, ch, GROUP_ID, "node-1", NODE_PK, self.gek,
                             dict(self.ack), self.identity, self.nonce, self.role)
 
+    def push_delta(self, *, additions=(), updates=(), deletions=()) -> None:
+        """An `index_delta` to every session, sealed as the node seals one. The
+        entries are applied here first, so a later `index_sync` agrees."""
+        for e in (*additions, *updates):
+            self.files[e["id"]] = e
+        for fid in deletions:
+            self.files.pop(fid, None)
+        wire = lambda es: [{k: v for k, v in e.items() if k != "data"} for e in es]  # noqa: E731
+        payload = {"base_version": 1, "version": 1, "additions": wire(additions),
+                   "deletions": list(deletions), "updates": wire(updates)}
+        msg = {"type": "index_delta", "group_id": GROUP_ID,
+               **seal(self.gek, PURPOSE_INDEX, "index_delta", GROUP_ID, payload)}
+        for ch in self.channels:
+            ch._emit(msg)
+
     def list_device(self, sender: str, device: Identity, *,
                     countersigned_by: Identity | None = None, proof: bool = True) -> None:
         """A roster row for `sender`'s device: a root, or countersigned by another
@@ -189,6 +208,7 @@ class FakeChannel:
 
     def _index_sync(self, msg):
         n = self.node
+        n.index_syncs += 1
         payload = {"version": 1, "roots": n.roots, "dirs": sorted(n.dirs),
                    "entries": [{k: v for k, v in e.items() if k != "data"}
                                for e in n.files.values()]}
diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py
index 36b7f47..9960533 100644
--- a/packages/meshbay-cli/tests/test_interactive.py
+++ b/packages/meshbay-cli/tests/test_interactive.py
@@ -444,3 +444,46 @@ async def test_put_sends_a_file_into_the_chat_folder_and_get_saves_one_here(
         await say("/get ../share/x")
         await _until(pilot, lambda: "name a file of the chat's folder" in _said(room))
         await pilot.press("escape")
+
+
+async def test_index_updates_are_applied_not_asked_for_again(world, monkeypatch):
+    """
+    A library being enriched pushes an `index_delta` about once a second, each
+    one filling in a file's fields. The panes used to answer every one with the
+    whole index — 67 full indexes in the 80 s of one download on a real node,
+    on the event loop the download was being received on. Updates are applied
+    where they land; only a delta that adds or removes a file asks again, once
+    for a burst.
+    """
+    from meshbay_cli.tui import screens
+    monkeypatch.setattr(screens, "REFRESH_AFTER_DELTA_S", 0.2)
+    hub, node, work = world
+    film = node.add("share", "film.mkv", b"F" * 5000)
+    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
+                                              "--password", PASSPHRASE, "login"]) == 0
+    app = _app()
+    async with app.run_test(size=(140, 40)) as pilot:
+        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
+        remote = app.screen.query_one(RemotePane)
+        await _open(pilot, remote, f"Family@{USERNAME}/")
+        await _open(pilot, remote, "share/")
+        await _until(pilot, lambda: "film.mkv" in _names(remote))
+        asked = node.index_syncs
+
+        for i in range(20):
+            node.push_delta(updates=[{**film, "duration": 100 + i}])
+        await pilot.pause(0.6)
+        assert node.index_syncs == asked, "an update must not fetch the whole index"
+        tree = app.screen._trees["g1"]
+        assert tree.file("share/film.mkv")["duration"] == 119
+
+        extra = [node.add("share", f"new{i}.txt", b"n%d" % i) for i in range(3)]
+        node.files = {k: v for k, v in node.files.items() if v not in extra}
+        for e in extra:
+            node.push_delta(additions=[e])
+        await _until(pilot, lambda: "new2.txt" in _names(remote))
+        await pilot.pause(0.6)
+        assert node.index_syncs == asked + 1, "a burst of additions asks again once"
+
+        node.push_delta(deletions=[film["id"]])
+        await _until(pilot, lambda: "film.mkv" not in _names(remote))