feat(node): index push to WebRTC peers + swarm registration (11.5, 11.9)

When watchdog detects file changes, the daemon now:
- Pushes INDEX_SYNC to all connected WebRTC peers in that group
- Registers file hashes with hub /v1/swarm/register endpoint

Also registers all file hashes on startup for initial discovery.
hub_client: add register_swarm() method for bulk hash registration.

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

3 files changed, 173 insertions, 2 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 6a8bbc9..5b6e770 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -31,6 +31,8 @@ from pathlib import Path
 
 import uvicorn
 
+from meshbay_common import MNP_VERSION
+from meshbay_common.protocol import MNP
 from meshbay_node.chat.store import ChatStore
 from meshbay_node.config import Config, load_config, write_example_config
 from meshbay_node.hub_client import HubClient, HubConfig
@@ -161,6 +163,7 @@ class NodeDaemon:
                     group_id=group_cfg.id,
                     sk_node=keys.sk_ed25519,
                     gek=gek,
+                    on_change=self._on_index_change,
                 )
                 await indexer.start()
                 self._indexers.append(indexer)
@@ -330,7 +333,14 @@ class NodeDaemon:
                      "yes" if self._webrtc else "no",
                      "yes" if self._quic_server else "no")
 
-            # 11. Wait for shutdown
+            # 11. Initial swarm registration
+            endpoint = f"webrtc:{self._config.node.quic_port}"
+            for gctx in groups_ctx.values():
+                hashes = [e.id for e in gctx["index"].entries]
+                if hashes:
+                    asyncio.ensure_future(self._register_swarm(hashes, endpoint))
+
+            # 12. Wait for shutdown
             stop_event = asyncio.Event()
             loop = asyncio.get_event_loop()
             for sig in (signal.SIGINT, signal.SIGTERM):
@@ -339,6 +349,54 @@ class NodeDaemon:
 
         await self._shutdown()
 
+    async def _on_index_change(self, indexer: DirectoryIndexer) -> None:
+        """Called when a DirectoryIndexer detects file changes."""
+        group_id = indexer.group_id
+        idx = indexer.index
+        log.info("Index changed for group %s: %d files (v%d)",
+                 group_id[:8], idx.count, idx.version)
+
+        # 11.5 — Push updated index to connected WebRTC peers in this group
+        if self._webrtc:
+            entries = [
+                {
+                    "id": e.id, "name": e.name, "path": e.path,
+                    "size": e.size, "type": e.type, "added_at": e.added_at,
+                }
+                for e in idx.entries
+            ]
+            sync_msg = {
+                "type": MNP.INDEX_SYNC,
+                "v": MNP_VERSION,
+                "group_id": idx.group_id,
+                "version": idx.version,
+                "entries": entries,
+            }
+            pushed = 0
+            for session in list(self._webrtc._sessions.values()):
+                if session._group_id == group_id:
+                    try:
+                        session._send(sync_msg)
+                        pushed += 1
+                    except Exception:
+                        pass
+            if pushed:
+                log.info("Index pushed to %d WebRTC peers", pushed)
+
+        # 11.9 — Register file hashes with hub swarm table
+        if self._hub and self._state.get("endpoint_hint"):
+            hashes = [e.id for e in idx.entries]
+            if hashes:
+                endpoint = f"webrtc:{self._config.node.quic_port}"
+                asyncio.ensure_future(self._register_swarm(hashes, endpoint))
+
+    async def _register_swarm(self, hashes: list[str], endpoint: str) -> None:
+        try:
+            n = await self._hub.register_swarm(hashes, endpoint)
+            log.info("Swarm: registered %d/%d hashes", n, len(hashes))
+        except Exception as e:
+            log.warning("Swarm registration failed: %s", e)
+
     async def _shutdown(self) -> None:
         log.info("Shutting down...")
         self._state["status"] = "stopping"
diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py
index ca0e776..ba9d3ff 100644
--- a/packages/meshbay-node/src/meshbay_node/hub_client.py
+++ b/packages/meshbay-node/src/meshbay_node/hub_client.py
@@ -329,6 +329,27 @@ class HubClient:
             finally:
                 self._ws = None
 
+    # ── Swarm registration ─────────────────────────────────────────────────
+
+    async def register_swarm(self, content_hashes: list[str], endpoint: str) -> int:
+        """Register file hashes in the hub swarm table. Returns count registered."""
+        if self._session is None:
+            raise RuntimeError("Not logged in")
+        await self.ensure_fresh_token()
+
+        registered = 0
+        for h in content_hashes:
+            try:
+                r = await self._http.post("/v1/swarm/register", json={
+                    "content_hash": h,
+                    "endpoint": endpoint,
+                }, headers=self._session.auth_headers)
+                if r.status_code in (201, 200):
+                    registered += 1
+            except Exception:
+                pass
+        return registered
+
     # ── Convenience: full startup sequence ───────────────────────────────────
 
     async def startup(self, endpoint_hint: str | None = None) -> HubSession:
diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index e74f7fc..faf12e3 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -2,7 +2,8 @@
 Integration test: Node daemon wires all components correctly.
 
 Phase 11 — verifies that NodeDaemon creates chat stores, WebRTC transport,
-and shuts down cleanly. Hub interaction is mocked.
+index push on change, swarm registration, and shuts down cleanly.
+Hub interaction is mocked.
 """
 
 import asyncio
@@ -16,6 +17,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 from meshbay_common.crypto import generate_gek
 from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
 from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer import DirectoryIndexer
 
 
 @pytest.fixture
@@ -169,3 +171,93 @@ async def test_daemon_no_groups_exits(tmp_path):
 
     assert daemon._state["status"] == "starting"
     assert len(daemon._chat_stores) == 0
+
+
+@pytest.mark.asyncio
+async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hub_pk_pem):
+    """Index change callback pushes updated index to WebRTC peers."""
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"),
+        node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000),
+        groups=[GroupConfig(
+            id="a" * 32,
+            name="test-group",
+            shared_dir=str(shared_dir),
+            visibility="private",
+            port=29000, quic_port=29010, http_port=29001,
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._hub = AsyncMock()
+    daemon._hub.register_swarm = AsyncMock(return_value=2)
+    daemon._state["endpoint_hint"] = "node123"
+
+    sk_node = Ed25519PrivateKey.generate()
+    indexer = DirectoryIndexer(
+        root=shared_dir, group_id="a" * 32,
+        sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    mock_session = MagicMock()
+    mock_session._group_id = "a" * 32
+    mock_session._send = MagicMock()
+
+    mock_webrtc = MagicMock()
+    mock_webrtc._sessions = {"peer1": mock_session}
+    daemon._webrtc = mock_webrtc
+
+    await daemon._on_index_change(indexer)
+
+    mock_session._send.assert_called_once()
+    msg = mock_session._send.call_args[0][0]
+    assert msg["type"] == "index_sync"
+    assert msg["group_id"] == "a" * 32
+    assert len(msg["entries"]) == indexer.index.count
+
+    await asyncio.sleep(0.1)
+    daemon._hub.register_swarm.assert_called_once()
+    call_args = daemon._hub.register_swarm.call_args
+    assert len(call_args[0][0]) == indexer.index.count
+
+
+@pytest.mark.asyncio
+async def test_daemon_index_change_skips_other_group_peers(
+    tmp_path, shared_dir, gek, hub_pk_pem
+):
+    """Index change only pushes to peers in the same group."""
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"),
+        node=NodeConfig(),
+        groups=[],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._hub = AsyncMock()
+    daemon._hub.register_swarm = AsyncMock(return_value=0)
+    daemon._state["endpoint_hint"] = "node123"
+
+    sk_node = Ed25519PrivateKey.generate()
+    indexer = DirectoryIndexer(
+        root=shared_dir, group_id="a" * 32,
+        sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    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
+
+    await daemon._on_index_change(indexer)
+
+    same_group._send.assert_called_once()
+    other_group._send.assert_not_called()