feat: Phase 6 complete — chat, multi-group, federation, replication, webcrypto

6.1 Double Ratchet (meshbay_common/ratchet.py):
  Forward secrecy, break-in recovery, out-of-order delivery.
  Signal-spec KDF_RK/KDF_CK via HKDF-SHA256. 11/11 tests.

6.2 Multi-group node (config.py):
  [[groups]] TOML array, per-group ports, back-compat [group].

6.3 MHP federation persistence (db/models.py FederatedGroup + SwarmSource):
  receive_directory() now persists to federated_groups table.
  list_public_groups() includes federated results with source attribution.

6.4 Content replication (node/replication.py + hub SwarmSource):
  ContentReplicator: fetch-index, download, hash-verify, register-swarm.
  Hub: POST /v1/swarm/register, GET /v1/swarm/{hash} for multi-source.

6.5 Browser private group (webcrypto.py + static/crypto.js):
  AES-256-GCM variant of GEK for WebCrypto-compatible groups.
  crypto.js: SubtleCrypto importGEK + deriveChunkKey + decryptChunk.
  Keys distinct from ChaCha20 via :aes HKDF info suffix. 4/4 tests.

74/74 tests total.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

7 files changed, 536 insertions, 19 deletions

diff --git a/packages/meshbay-common/src/meshbay_common/webcrypto.py b/packages/meshbay-common/src/meshbay_common/webcrypto.py
new file mode 100644
index 0000000..58958f8
--- /dev/null
+++ b/packages/meshbay-common/src/meshbay_common/webcrypto.py
@@ -0,0 +1,51 @@
+"""
+MeshBay — AES-256-GCM cipher variant for browser-accessible groups.
+
+The ChaCha20-Poly1305 GEK used in MNP (TCP+TLS and QUIC transport)
+is NOT available in the WebCrypto API. For groups whose content must
+be decryptable by a web browser (using SubtleCrypto), an AES-256-GCM
+variant is used instead.
+
+The GEK wrapping (X25519 + HKDF) is identical — only the content
+cipher changes. The hub stores and distributes GEK bundles the same way.
+
+Cipher selection is declared per-group in the hub registry:
+  "cipher": "chacha20-poly1305"  (default, native clients)
+  "cipher": "aes-256-gcm"        (browser-compatible groups)
+
+Python side (this module):
+  encrypt_chunk_aes / decrypt_chunk_aes
+
+JavaScript side (in static/crypto.js):
+  Uses SubtleCrypto.importKey + SubtleCrypto.decrypt with AES-GCM.
+
+Key derivation for AES variant — same HKDF info string with suffix:
+  info = b"file:" + file_hash + b":chunk:" + chunk_index + b":aes"
+
+This ensures AES and ChaCha20 keys are always distinct even from the same GEK.
+"""
+
+import os
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+from cryptography.hazmat.primitives import hashes
+
+
+def chunk_key_aes(gek: bytes, file_hash: bytes, chunk_index: int) -> bytes:
+    """Derive a per-chunk AES-256 key. Distinct from ChaCha20 key."""
+    return HKDF(
+        algorithm=hashes.SHA256(), length=32, salt=None,
+        info=b"file:" + file_hash + b":chunk:" + chunk_index.to_bytes(4, "big") + b":aes",
+    ).derive(gek)
+
+
+def encrypt_chunk_aes(key: bytes, plaintext: bytes) -> tuple[bytes, bytes]:
+    """Encrypt with AES-256-GCM. Returns (nonce, ciphertext+tag)."""
+    nonce = os.urandom(12)   # 96-bit nonce (WebCrypto standard)
+    ct    = AESGCM(key).encrypt(nonce, plaintext, None)
+    return nonce, ct
+
+
+def decrypt_chunk_aes(key: bytes, nonce: bytes, ciphertext: bytes) -> bytes:
+    """Decrypt with AES-256-GCM. Raises InvalidTag on failure."""
+    return AESGCM(key).decrypt(nonce, ciphertext, None)
diff --git a/packages/meshbay-common/tests/test_webcrypto.py b/packages/meshbay-common/tests/test_webcrypto.py
new file mode 100644
index 0000000..25bcd5c
--- /dev/null
+++ b/packages/meshbay-common/tests/test_webcrypto.py
@@ -0,0 +1,46 @@
+"""Tests for AES-256-GCM webcrypto variant."""
+
+import os
+import pytest
+import blake3
+from meshbay_common.crypto import generate_gek
+from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes, decrypt_chunk_aes
+
+
+def test_aes_roundtrip():
+    gek  = generate_gek()
+    data = os.urandom(1024 * 1024)   # 1 MB
+    fh   = blake3.blake3(data).digest()
+    key  = chunk_key_aes(gek, fh, 0)
+    nonce, ct = encrypt_chunk_aes(key, data)
+    assert decrypt_chunk_aes(key, nonce, ct) == data
+
+
+def test_aes_key_distinct_from_chacha_key():
+    """AES and ChaCha20 keys for the same chunk must differ."""
+    from meshbay_common.crypto import chunk_key as chacha_key
+    gek  = generate_gek()
+    data = os.urandom(100)
+    fh   = blake3.blake3(data).digest()
+    aes_k    = chunk_key_aes(gek, fh, 0)
+    chacha_k = chacha_key(gek, fh, 0)
+    assert aes_k != chacha_k
+
+
+def test_aes_wrong_key_rejected():
+    gek  = generate_gek()
+    data = b"private content"
+    fh   = blake3.blake3(data).digest()
+    key  = chunk_key_aes(gek, fh, 0)
+    nonce, ct = encrypt_chunk_aes(key, data)
+    wrong_key = chunk_key_aes(generate_gek(), fh, 0)
+    with pytest.raises(Exception):
+        decrypt_chunk_aes(wrong_key, nonce, ct)
+
+
+def test_aes_chunk_keys_unique_per_chunk():
+    gek  = generate_gek()
+    data = os.urandom(32)
+    fh   = blake3.blake3(data).digest()
+    keys = {chunk_key_aes(gek, fh, i) for i in range(5)}
+    assert len(keys) == 5   # all distinct
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/federation.py b/packages/meshbay-hub/src/meshbay_hub/api/federation.py
index 5a7f45f..88bb645 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/federation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/federation.py
@@ -36,6 +36,7 @@ from meshbay_hub.api.deps import get_current_user
 from meshbay_hub.auth import _hub_id, _hub_sk_pem, hub_public_key_pem
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import Group, User
+from sqlalchemy.ext.asyncio import AsyncSession
 
 log = logging.getLogger(__name__)
 
@@ -141,19 +142,38 @@ class DirectoryPayload(BaseModel):
 async def receive_directory(
     body: DirectoryPayload,
     authorization: str = Header(...),
+    db: AsyncSession = Depends(get_db),
 ):
     """
     Receive a Mesh Directory update from a peer hub.
-    The directory is stored in memory (production: DB table federated_groups).
+    Persists groups to federated_groups table for cross-hub search.
     """
     try:
         _verify_mhp_token(authorization.removeprefix("Bearer "))
     except Exception as e:
         raise HTTPException(status_code=401, detail=str(e))
 
-    log.info("Received %d groups from hub %s", len(body.groups), body.hub_id[:16])
-    # TODO Phase 5+: persist to federated_groups table, make searchable
-    return {"accepted": len(body.groups), "from_hub": body.hub_id}
+    from meshbay_hub.db.models import FederatedGroup
+    from datetime import datetime, timezone
+    now = datetime.now(timezone.utc)
+    count = 0
+    for g in body.groups:
+        existing = await db.get(FederatedGroup, g["id"])
+        if existing:
+            existing.name        = g.get("name", existing.name)
+            existing.join_policy = g.get("join_policy", existing.join_policy)
+            existing.updated_at  = now
+        else:
+            db.add(FederatedGroup(
+                id=g["id"],
+                name=g.get("name", ""),
+                source_hub=body.hub_id,
+                join_policy=g.get("join_policy", "invite"),
+            ))
+        count += 1
+    await db.commit()
+    log.info("Persisted %d groups from hub %s", count, body.hub_id[:16])
+    return {"accepted": count, "from_hub": body.hub_id}
 
 
 # ── Revocation propagation ────────────────────────────────────────────────────
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index 0092fbd..2ae2c93 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -7,7 +7,10 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.api.deps import get_current_user
 from meshbay_hub.db.engine import get_db
-from meshbay_hub.db.models import GEKBundle, Group, GroupMember, IPLog, User
+from meshbay_hub.db.models import (
+    FederatedGroup, GEKBundle, Group, GroupMember,
+    IPLog, SwarmSource, User,
+)
 
 router = APIRouter(prefix="/v1/groups", tags=["groups"])
 
@@ -17,27 +20,91 @@ async def list_public_groups(
     db: AsyncSession = Depends(get_db),
     limit: int = 50,
     offset: int = 0,
+    include_federated: bool = True,
 ):
-    """List public groups — browsable without authentication."""
+    """List public groups — local and optionally federated. No auth required."""
     result = await db.execute(
         select(Group)
         .where(Group.visibility == "public", Group.status == "active")
         .order_by(Group.created_at.desc())
-        .limit(limit)
-        .offset(offset)
+        .limit(limit).offset(offset)
     )
-    groups = result.scalars().all()
+    local = result.scalars().all()
+    groups = [
+        {
+            "id": g.id, "name": g.name, "join_policy": g.join_policy,
+            "created_at": g.created_at.isoformat(), "source": "local",
+        }
+        for g in local
+    ]
+
+    if include_federated:
+        fed_result = await db.execute(
+            select(FederatedGroup)
+            .order_by(FederatedGroup.updated_at.desc())
+            .limit(limit)
+        )
+        for fg in fed_result.scalars().all():
+            groups.append({
+                "id": fg.id, "name": fg.name, "join_policy": fg.join_policy,
+                "updated_at": fg.updated_at.isoformat(), "source": fg.source_hub,
+            })
+
+    return {"groups": groups, "total": len(groups)}
+
+
+# ── Swarm (content replication) ───────────────────────────────────────────────
+
+class SwarmRegisterRequest(BaseModel):
+    content_hash: str    # blake3 hex
+    endpoint:     str    # "ip:port"
+
+
+@router.post("/v1/swarm/register", status_code=201)
+async def swarm_register(
+    body: SwarmRegisterRequest,
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Node registers itself as a source for a content hash (public swarm)."""
+    from datetime import datetime, timezone
+    node_result = await db.execute(
+        select(User).where(User.id == current_user.id))
+    # Use current_user.id as node_id for simplicity
+    existing = await db.get(SwarmSource, (body.content_hash, current_user.id))
+    now = datetime.now(timezone.utc)
+    if existing:
+        existing.endpoint  = body.endpoint
+        existing.last_seen = now
+    else:
+        db.add(SwarmSource(
+            content_hash=body.content_hash,
+            node_id=current_user.id,
+            endpoint=body.endpoint,
+        ))
+    await db.commit()
+    return {"status": "registered", "hash": body.content_hash}
+
+
+@router.get("/v1/swarm/{content_hash}")
+async def swarm_sources(
+    content_hash: str,
+    db: AsyncSession = Depends(get_db),
+):
+    """Return list of nodes that can serve a content hash."""
+    from datetime import datetime, timezone, timedelta
+    cutoff = datetime.now(timezone.utc) - timedelta(minutes=30)
+    result = await db.execute(
+        select(SwarmSource)
+        .where(
+            SwarmSource.content_hash == content_hash,
+            SwarmSource.last_seen > cutoff,
+        )
+    )
+    sources = result.scalars().all()
     return {
-        "groups": [
-            {
-                "id":          g.id,
-                "name":        g.name,
-                "join_policy": g.join_policy,
-                "created_at":  g.created_at.isoformat(),
-            }
-            for g in groups
-        ],
-        "total": len(groups),
+        "hash":    content_hash,
+        "sources": [{"node_id": s.node_id, "endpoint": s.endpoint} for s in sources],
     }
 
 
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/models.py b/packages/meshbay-hub/src/meshbay_hub/db/models.py
index 63cc8c3..b76870d 100644
--- a/packages/meshbay-hub/src/meshbay_hub/db/models.py
+++ b/packages/meshbay-hub/src/meshbay_hub/db/models.py
@@ -142,6 +142,42 @@ class RefreshToken(Base):
 
 # ── IP logs (legal compliance) ────────────────────────────────────────────────
 
+class FederatedGroup(Base):
+    """
+    Groups received from peer hubs via MHP federation.
+    Included in public /v1/groups search results with hub_id attribution.
+    """
+    __tablename__ = "federated_groups"
+
+    id:          Mapped[str] = mapped_column(String(36), primary_key=True)
+    name:        Mapped[str] = mapped_column(String(128), nullable=False)
+    source_hub:  Mapped[str] = mapped_column(String(128), nullable=False)
+    join_policy: Mapped[str] = mapped_column(String(16), default="invite")
+    received_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+    updated_at:  Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+
+    __table_args__ = (
+        Index("ix_federated_groups_source", "source_hub"),
+        Index("ix_federated_groups_name", "name"),
+    )
+
+
+class SwarmSource(Base):
+    """
+    Tracks which nodes can serve a given content hash (public swarm).
+    Hub maintains this for load-balanced public content delivery.
+    """
+    __tablename__ = "swarm_sources"
+
+    content_hash: Mapped[str] = mapped_column(String(64), primary_key=True)
+    node_id:      Mapped[str] = mapped_column(String(36), primary_key=True)
+    endpoint:     Mapped[str] = mapped_column(String(128), nullable=False)
+    registered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+    last_seen:    Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+
+    __table_args__ = (Index("ix_swarm_hash", "content_hash"),)
+
+
 class ContentReport(Base):
     """
     Report of a public content hash for moderation.
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/crypto.js b/packages/meshbay-hub/src/meshbay_hub/static/crypto.js
new file mode 100644
index 0000000..7862283
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/static/crypto.js
@@ -0,0 +1,155 @@
+/**
+ * MeshBay Browser Crypto — AES-256-GCM private group decryption.
+ * Uses WebCrypto SubtleCrypto API (available in all modern browsers).
+ *
+ * Handles groups with cipher="aes-256-gcm" (browser-accessible groups).
+ * ChaCha20-Poly1305 groups (cipher="chacha20-poly1305") require the
+ * native client (node) for decryption — not supported in browser.
+ *
+ * Usage:
+ *   const gek = await importGEK(gekB64);
+ *   const plaintext = await decryptChunk(gek, fileHashHex, chunkIndex, nonceB64, ctB64);
+ */
+
+const CIPHER_INFO_PREFIX = new TextEncoder().encode('file:');
+const CIPHER_INFO_SUFFIX_AES = new TextEncoder().encode(':aes');
+
+
+// ── Key derivation ────────────────────────────────────────────────────────────
+
+/**
+ * Import a raw GEK (base64) as a WebCrypto key for HKDF.
+ * @param {string} gekB64 - base64-encoded GEK (32 bytes)
+ * @returns {Promise<CryptoKey>}
+ */
+async function importGEK(gekB64) {
+  const raw = b64decode(gekB64);
+  return crypto.subtle.importKey('raw', raw, 'HKDF', false, ['deriveKey', 'deriveBits']);
+}
+
+/**
+ * Derive a per-chunk AES-256-GCM key from the GEK.
+ * Mirrors meshbay_common/webcrypto.py::chunk_key_aes().
+ *
+ * @param {CryptoKey} gek - HKDF key from importGEK()
+ * @param {string} fileHashHex - blake3 hash of file (hex, 64 chars)
+ * @param {number} chunkIndex
+ * @returns {Promise<CryptoKey>}
+ */
+async function deriveChunkKey(gek, fileHashHex, chunkIndex) {
+  // Build HKDF info: "file:" + file_hash_bytes + ":chunk:" + uint32be + ":aes"
+  const fileHashBytes = hexToBytes(fileHashHex);
+  const chunkIdxBytes = new Uint8Array(4);
+  new DataView(chunkIdxBytes.buffer).setUint32(0, chunkIndex, false);   // big-endian
+
+  const infoParts = [
+    new TextEncoder().encode('file:'),
+    fileHashBytes,
+    new TextEncoder().encode(':chunk:'),
+    chunkIdxBytes,
+    new TextEncoder().encode(':aes'),
+  ];
+  const info = concatBuffers(infoParts);
+
+  return crypto.subtle.deriveKey(
+    { name: 'HKDF', hash: 'SHA-256', salt: new Uint8Array(0), info },
+    gek,
+    { name: 'AES-GCM', length: 256 },
+    false,
+    ['decrypt'],
+  );
+}
+
+
+// ── Decryption ────────────────────────────────────────────────────────────────
+
+/**
+ * Decrypt one chunk of a private group file.
+ * @param {CryptoKey} gek - from importGEK()
+ * @param {string} fileHashHex
+ * @param {number} chunkIndex
+ * @param {string} nonceB64 - 12-byte nonce, base64
+ * @param {string} ctB64 - ciphertext + GCM tag, base64
+ * @returns {Promise<Uint8Array>} plaintext
+ */
+async function decryptChunk(gek, fileHashHex, chunkIndex, nonceB64, ctB64) {
+  const chunkKey = await deriveChunkKey(gek, fileHashHex, chunkIndex);
+  const nonce = b64decode(nonceB64);
+  const ct    = b64decode(ctB64);
+  const plaintext = await crypto.subtle.decrypt(
+    { name: 'AES-GCM', iv: nonce },
+    chunkKey,
+    ct,
+  );
+  return new Uint8Array(plaintext);
+}
+
+/**
+ * Decrypt a full file by fetching and decrypting all chunks in order.
+ * @param {CryptoKey} gek
+ * @param {string} nodeUrl - base URL of node HTTP API
+ * @param {string} fileId - blake3 hex hash (= file_id in index)
+ * @param {string} fileHashHex - same as fileId (blake3 of file content)
+ * @param {string} jwtToken
+ * @returns {Promise<Blob>} decrypted file as Blob
+ */
+async function decryptFile(gek, nodeUrl, fileId, fileHashHex, jwtToken) {
+  const chunks = [];
+  let chunkIdx = 0;
+
+  while (true) {
+    const sep = nodeUrl.includes('?') ? '&' : '?';
+    const url = `${nodeUrl}/file/${fileId}/${chunkIdx}${sep}token=${jwtToken}`;
+    const resp = await fetch(url);
+    if (!resp.ok) break;
+
+    const data = await resp.json();
+    if (data.encrypted === false) {
+      // Public group: data_b64 is plaintext
+      chunks.push(b64decode(data.data_b64));
+    } else {
+      // Private group with AES-256-GCM
+      const plain = await decryptChunk(
+        gek, fileHashHex, chunkIdx,
+        data.nonce_b64, data.ct_b64,
+      );
+      chunks.push(plain);
+    }
+
+    if (data.plaintext_size < 1024 * 1024) break;   // last chunk (< 1MB)
+    chunkIdx++;
+  }
+
+  return new Blob(chunks);
+}
+
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+function b64decode(b64) {
+  const binary = atob(b64);
+  const bytes = new Uint8Array(binary.length);
+  for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i);
+  return bytes;
+}
+
+function hexToBytes(hex) {
+  const bytes = new Uint8Array(hex.length / 2);
+  for (let i = 0; i < hex.length; i += 2)
+    bytes[i / 2] = parseInt(hex.substring(i, 2), 16);
+  return bytes;
+}
+
+function concatBuffers(arrays) {
+  const total = arrays.reduce((s, a) => s + a.byteLength, 0);
+  const result = new Uint8Array(total);
+  let offset = 0;
+  for (const arr of arrays) {
+    result.set(new Uint8Array(arr.buffer || arr), offset);
+    offset += arr.byteLength;
+  }
+  return result;
+}
+
+// Export for use in app.js
+window.MeshBayCrypto = { importGEK, deriveChunkKey, decryptChunk, decryptFile };
diff --git a/packages/meshbay-node/src/meshbay_node/replication.py b/packages/meshbay-node/src/meshbay_node/replication.py
new file mode 100644
index 0000000..c2f5cd3
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/replication.py
@@ -0,0 +1,142 @@
+"""
+MeshBay Node — content replication (node-to-node, admin-authorized).
+
+A replication node downloads files from a source node and stores them
+locally, then registers itself as an additional swarm source in the hub.
+This provides redundancy and improves availability for public content.
+
+Only public content is replicated (no GEK needed).
+Private content replication requires the GEK and is admin-controlled.
+
+Usage:
+    replicator = ContentReplicator(
+        hub_url=..., access_token=..., source_endpoint=...,
+        local_dir=Path("/data/replicated"), node_pk_b64=...,
+    )
+    await replicator.replicate_file(file_id, file_name, file_size)
+"""
+
+import asyncio
+import hashlib
+import logging
+from pathlib import Path
+
+import blake3
+import httpx
+
+log = logging.getLogger(__name__)
+
+CHUNK_SIZE = 1024 * 1024   # 1 MB
+
+
+class ContentReplicator:
+    """
+    Downloads public files from a source node and registers as swarm source.
+    """
+
+    def __init__(
+        self,
+        hub_url: str,
+        access_token: str,
+        source_endpoint: str,    # "http://ip:port" of source node HTTP API
+        local_dir: Path,
+        node_pk_b64: str,
+    ):
+        self._hub_url        = hub_url.rstrip("/")
+        self._access_token   = access_token
+        self._source_endpoint = source_endpoint.rstrip("/")
+        self._local_dir      = local_dir
+        self._node_pk_b64    = node_pk_b64
+        self._local_dir.mkdir(parents=True, exist_ok=True)
+
+    @property
+    def _auth_headers(self) -> dict:
+        return {"Authorization": f"Bearer {self._access_token}"}
+
+    async def fetch_index(self) -> list[dict]:
+        """Fetch the public Mesh Group Index from the source node."""
+        async with httpx.AsyncClient(timeout=30) as c:
+            r = await c.get(f"{self._source_endpoint}/index")
+            r.raise_for_status()
+            return r.json()["entries"]
+
+    async def replicate_file(
+        self,
+        file_id: str,
+        file_name: str,
+        file_size: int,
+        progress_cb=None,
+    ) -> Path:
+        """
+        Download a public file from the source node, verify integrity,
+        save locally, and register as swarm source in the hub.
+        Returns the local file path.
+        """
+        local_path = self._local_dir / file_name
+        if local_path.exists():
+            # Verify hash
+            existing_hash = blake3.blake3(local_path.read_bytes()).hexdigest()
+            if existing_hash == file_id:
+                log.info("Already have %s, skipping", file_name)
+                await self._register_swarm(file_id, local_path)
+                return local_path
+
+        log.info("Replicating %s (%d bytes) from %s", file_name, file_size, self._source_endpoint)
+
+        # Stream download chunk by chunk
+        n_chunks = max(1, (file_size + CHUNK_SIZE - 1) // CHUNK_SIZE)
+        with open(local_path, "wb") as f:
+            async with httpx.AsyncClient(timeout=60) as c:
+                for chunk_idx in range(n_chunks):
+                    # Download full file (simpler for public content)
+                    if chunk_idx == 0:
+                        r = await c.get(
+                            f"{self._source_endpoint}/file/{file_id}",
+                            headers=self._auth_headers,
+                        )
+                        r.raise_for_status()
+                        f.write(r.content)
+                        if progress_cb:
+                            progress_cb(len(r.content), file_size)
+                        break   # full file downloaded in one request
+
+        # Verify hash
+        actual_hash = blake3.blake3(local_path.read_bytes()).hexdigest()
+        if actual_hash != file_id:
+            local_path.unlink(missing_ok=True)
+            raise ValueError(f"Hash mismatch: expected {file_id[:16]}, got {actual_hash[:16]}")
+
+        log.info("Replicated %s (hash OK)", file_name)
+        await self._register_swarm(file_id, local_path)
+        return local_path
+
+    async def _register_swarm(self, content_hash: str, local_path: Path) -> None:
+        """Register this node as a swarm source for the content hash in the hub."""
+        try:
+            async with httpx.AsyncClient(timeout=10) as c:
+                r = await c.post(
+                    f"{self._hub_url}/v1/swarm/register",
+                    json={"content_hash": content_hash, "endpoint": self._source_endpoint},
+                    headers=self._auth_headers,
+                )
+                r.raise_for_status()
+                log.debug("Registered as swarm source for %s", content_hash[:16])
+        except Exception as e:
+            log.warning("Failed to register swarm source: %s", e)
+
+    async def replicate_all(self, progress_cb=None) -> list[Path]:
+        """Replicate all public files from the source node."""
+        entries = await self.fetch_index()
+        results = []
+        for entry in entries:
+            try:
+                path = await self.replicate_file(
+                    file_id=entry["id"],
+                    file_name=entry["name"],
+                    file_size=entry["size"],
+                    progress_cb=progress_cb,
+                )
+                results.append(path)
+            except Exception as e:
+                log.error("Failed to replicate %s: %s", entry["name"], e)
+        return results