diff --git a/arborist/cli.py b/arborist/cli.py
index ecc7d09..94bf49e 100644
--- a/arborist/cli.py
+++ b/arborist/cli.py
@@ -2868,7 +2868,21 @@ class _LocalOnlyBackend:
 
 
 def _cmd_cold_unpack(args: argparse.Namespace) -> int:
-    from arborist.evict import pull_pack
+    """Hydrate a shard from a metadata pack hash. SPV-style consumer:
+
+    - default mode "just-enough": pull only the metadata pack. Schema
+      restored, every chunks row has content=NULL. Queryable immediately
+      for metadata; chunk-body queries return null until a future JIT-
+      fetch path fills them on cache miss.
+    - "--full": after the metadata pack lands, iterate its
+      `_chunk_pack_hashes` and pull every chunk pack. Final state: full
+      corpus offline-queryable.
+
+    Either way, `args.pack_hash` must be a metadata pack hash. Chunk
+    packs are anonymous from the consumer's perspective — accessed via
+    the metadata pack's reference list, not by name.
+    """
+    from arborist.evict import hydrate_from_metadata_pack
 
     backend = _make_cold_backend()
     conn = (
@@ -2877,7 +2891,9 @@ def _cmd_cold_unpack(args: argparse.Namespace) -> int:
         else connect(args.db)
     )
     try:
-        result = pull_pack(conn, backend, args.pack_hash)
+        result = hydrate_from_metadata_pack(
+            conn, backend, args.pack_hash, mode=args.mode,
+        )
     finally:
         conn.close()
     print(json.dumps(result, indent=2, ensure_ascii=False))
@@ -2906,32 +2922,47 @@ def _cmd_cold_stats(args: argparse.Namespace) -> int:
 
 
 def _cmd_cold_list(args: argparse.Namespace) -> int:
-    """Enumerate packs in the bucket for new-peer hydration.
+    """Enumerate packs in the bucket for new-peer hydration. SPV-aware.
 
-    For each `<pack>.tar.zst`, HEAD the object to get its compressed size,
-    and (unless --no-manifest) fetch the small manifest sidecar to count
-    chunks. Output is a JSON array — pipe through jq or feed to a
-    hydration script that calls `cold unpack` for each entry.
+    For each `<hash>.{metadata,chunks}.tar.zst`, HEAD for size and
+    (unless --no-manifest) fetch the small manifest sidecar to surface
+    kind + chunk_count / table_count + chunk_pack_hashes. Output
+    enables a consumer to:
+      - find the metadata pack(s) — the entry points
+      - know what chunk packs each metadata pack references (full sync)
     """
     from arborist.cold_object import PACK_PREFIX, pack_key, parse_manifest
 
     backend = _make_cold_backend()
     packs = []
     for key in backend.list_keys(PACK_PREFIX):
-        if not key.endswith(".tar.zst"):
+        # v3 keys: packs/<hash>.metadata.tar.zst  or  packs/<hash>.chunks.tar.zst
+        if key.endswith(".metadata.tar.zst"):
+            kind = "metadata"
+            pack_hash = key[len(PACK_PREFIX):-len(".metadata.tar.zst")]
+        elif key.endswith(".chunks.tar.zst"):
+            kind = "chunks"
+            pack_hash = key[len(PACK_PREFIX):-len(".chunks.tar.zst")]
+        else:
             continue
-        # key = "packs/<pack_hash>.tar.zst"
-        pack_hash = key[len(PACK_PREFIX):-len(".tar.zst")]
         entry = {
             "pack_hash": pack_hash,
+            "kind": kind,
             "compressed_bytes": backend.object_size(key),
         }
         if args.fetch_manifest:
             try:
-                manifest_bytes = backend.get(pack_key(pack_hash, manifest=True))
-                entry["chunk_count"] = len(parse_manifest(manifest_bytes).chunks)
+                manifest_bytes = backend.get(
+                    pack_key(pack_hash, kind=kind, manifest=True)
+                )
+                parsed = parse_manifest(manifest_bytes)
+                if kind == "metadata":
+                    entry["table_count"] = len(parsed.tables)
+                    entry["snapshot_root"] = parsed.snapshot_root
+                    entry["chunk_pack_hashes"] = list(parsed.chunk_pack_hashes)
+                else:
+                    entry["chunk_count"] = len(parsed.chunks)
             except Exception as e:  # noqa: BLE001 — surface but don't fail
-                entry["chunk_count"] = None
                 entry["manifest_error"] = repr(e)
         packs.append(entry)
     out = {
@@ -5707,9 +5738,31 @@ def build_parser() -> argparse.ArgumentParser:
 
     cold_unpack = cold_sub.add_parser(
         "unpack",
-        help="pull one pack from bucket and restore chunks locally (hash-verified)",
+        help=(
+            "hydrate a shard from a metadata pack hash. default is "
+            "--just-enough (metadata only, chunks NULL); --full also "
+            "pulls every referenced chunk pack."
+        ),
+    )
+    cold_unpack.add_argument(
+        "pack_hash",
+        help="metadata pack hash (NOT a chunk pack hash). Find via `cold list`.",
+    )
+    cold_unpack.add_argument(
+        "--full",
+        dest="mode",
+        action="store_const",
+        const="full",
+        default="just-enough",
+        help="after metadata, pull every chunk pack the metadata references",
+    )
+    cold_unpack.add_argument(
+        "--just-enough",
+        dest="mode",
+        action="store_const",
+        const="just-enough",
+        help="(default) metadata only — chunks remain NULL for JIT fetch",
     )
-    cold_unpack.add_argument("pack_hash", help="pack_hash = sha256(manifest)")
     cold_unpack.set_defaults(func=_cmd_cold_unpack)
 
     cold_stats = cold_sub.add_parser(
diff --git a/arborist/cold_object.py b/arborist/cold_object.py
index 65a7ca9..0ad363a 100644
--- a/arborist/cold_object.py
+++ b/arborist/cold_object.py
@@ -77,11 +77,32 @@ def hash_file_leaf(path: Path, *, chunk_size: int = 1024 * 1024) -> str:
     return h.hexdigest()
 
 
-def pack_key(pack_hash: str, *, manifest: bool = False) -> str:
-    """Bucket key for a pack body (default) or its manifest sidecar."""
+def pack_key(
+    pack_hash: str,
+    *,
+    kind: str = "chunks",
+    manifest: bool = False,
+) -> str:
+    """Bucket key for a pack body (default) or its manifest sidecar.
+
+    `kind` distinguishes the two artifact shapes in the v3 SPV layout:
+      - "chunks"   — `packs/<hash>.chunks.tar.zst` (chunk bodies, N per shard)
+      - "metadata" — `packs/<hash>.metadata.tar.zst` (one per shard, references chunk_pack_hashes)
+
+    Both can be pulled independently; a "just-enough" sync only needs the
+    metadata pack, a "full" sync also pulls every chunk pack in
+    `_chunk_pack_hashes`. Kind goes into the bucket key so a fresh peer
+    can list-by-prefix and prioritize metadata downloads.
+    """
+    if kind not in ("metadata", "chunks"):
+        raise ValueError(f"unknown pack kind: {kind!r}")
     if manifest:
-        return f"{PACK_PREFIX}{pack_hash}.manifest.ndjson"
-    return f"{PACK_PREFIX}{pack_hash}.tar.zst"
+        return f"{PACK_PREFIX}{pack_hash}.{kind}.manifest.ndjson"
+    return f"{PACK_PREFIX}{pack_hash}.{kind}.tar.zst"
+
+
+PACK_KIND_METADATA = "metadata"
+PACK_KIND_CHUNKS = "chunks"
 
 
 @dataclass(frozen=True)
@@ -157,26 +178,37 @@ class ObjectStoreBackend(abc.ABC):
         ...
 
     # Convenience wrappers — keep call sites short and consistent.
-    def put_pack(self, pack_hash: str, body: bytes) -> None:
-        self.put(pack_key(pack_hash), body, content_type="application/zstd")
+    # `kind` distinguishes metadata packs from chunk packs in the v3 SPV
+    # layout. Default kind="chunks" preserves the call-site shape for
+    # chunk-pack producers; metadata-pack producers pass kind="metadata".
+    def put_pack(self, pack_hash: str, body: bytes, *, kind: str = "chunks") -> None:
+        self.put(pack_key(pack_hash, kind=kind), body, content_type="application/zstd")
 
-    def put_pack_file(self, pack_hash: str, path: Path) -> None:
+    def put_pack_file(self, pack_hash: str, path: Path, *, kind: str = "chunks") -> None:
         """File-backed pack upload — `stream_packs` produces these.
 
         boto3 `upload_file` reads parts from disk for the multipart upload
         instead of loading the whole pack into memory. Required for any
         parallelism above ~2 workers at the 4.4 GB cap.
         """
-        self.put_file(pack_key(pack_hash), path, content_type="application/zstd")
+        self.put_file(
+            pack_key(pack_hash, kind=kind), path, content_type="application/zstd"
+        )
 
-    def get_pack(self, pack_hash: str) -> bytes:
-        return self.get(pack_key(pack_hash))
+    def get_pack(self, pack_hash: str, *, kind: str = "chunks") -> bytes:
+        return self.get(pack_key(pack_hash, kind=kind))
 
-    def put_pack_manifest(self, pack_hash: str, body: bytes) -> None:
-        self.put(pack_key(pack_hash, manifest=True), body, content_type="application/x-ndjson")
+    def put_pack_manifest(
+        self, pack_hash: str, body: bytes, *, kind: str = "chunks"
+    ) -> None:
+        self.put(
+            pack_key(pack_hash, kind=kind, manifest=True),
+            body,
+            content_type="application/x-ndjson",
+        )
 
-    def get_pack_manifest(self, pack_hash: str) -> bytes:
-        return self.get(pack_key(pack_hash, manifest=True))
+    def get_pack_manifest(self, pack_hash: str, *, kind: str = "chunks") -> bytes:
+        return self.get(pack_key(pack_hash, kind=kind, manifest=True))
 
 
 class S3CompatibleBackend(ObjectStoreBackend):
@@ -444,7 +476,11 @@ def build_pack(chunks: list[tuple[str, bytes]]) -> Pack:
             raise ValueError(f"bad leaf_hash length: {leaf_hash}")
         by_hash[leaf_hash] = body
 
-    manifest_lines = []
+    # v3 chunks-only manifest (matches stream_packs output shape).
+    manifest_lines = [
+        json.dumps({"_format_version": 3}, sort_keys=True),
+        json.dumps({"_kind": PACK_KIND_CHUNKS}, sort_keys=True),
+    ]
     entries = []
     for leaf_hash in sorted(by_hash):
         body = by_hash[leaf_hash]
@@ -486,7 +522,6 @@ def stream_packs(
     max_compressed_bytes: int,
     level: int = 3,
     work_dir: Path | str | None = None,
-    extra_members: list[tuple[str, Path, str]] | None = None,
 ) -> Iterator[FilePack]:
     """Stream (leaf_hash, raw_utf8_body) chunks into FILE-BACKED packs that
     each fit in `max_compressed_bytes` compressed bytes.
@@ -521,22 +556,16 @@ def stream_packs(
     Single chunks larger than the cap still get their own pack (we don't
     drop data — the alternative is silently losing it).
 
-    `extra_members` is a list of (tar_member_name, source_file_path,
-    content_hash) for files to include AT THE START OF THE FIRST PACK.
-    Used by pack format v2 to ship shard-metadata table dumps
-    (`tables/<name>.jsonl`) ahead of the chunk bodies. `content_hash`
-    must be `hash_leaf(source_file_contents).hex()` — it goes into the
-    pack manifest so `pack_hash` content-addresses the *full* pack
-    (tables + chunks), not just chunks. Subsequent packs (if the chunk
-    set overflows the cap) do not repeat the extra members — they're
-    conceptually shard-level state, not chunk-level. If the v1 chunks-
-    only pack shape is needed, pass `extra_members=None` (default).
+    Pack v3 (SPV split): this emits CHUNK-ONLY packs. Metadata lives in
+    a separate artifact built by `build_metadata_pack` so consumers
+    pulling metadata-only can skip the chunk bodies entirely. The
+    metadata pack carries the list of every chunk_pack_hash for its
+    shard so a "full" consumer knows what to fetch.
     """
     if max_compressed_bytes <= 0:
         raise ValueError("max_compressed_bytes must be positive")
     work_path = Path(work_dir) if work_dir is not None else Path(tempfile.gettempdir())
     work_path.mkdir(parents=True, exist_ok=True)
-    pending_extras = list(extra_members) if extra_members else []
 
     def _start_pack() -> dict:
         # Tempfile in work_dir, opened for binary write. We hand the raw
@@ -558,36 +587,20 @@ def stream_packs(
             "writer": writer,
             "tar": tar,
             "entries_by_hash": {},
-            "table_refs": [],  # (member_name, content_hash, size) — populated for first pack only
         }
 
     def _finalize_pack(state: dict) -> FilePack:
         entries_sorted = sorted(state["entries_by_hash"].keys())
-        manifest_lines = []
-
-        # Pack format v2: tag explicitly. Consumers branch on this to
-        # know they should expect + verify `table_file` entries before
-        # the chunk entries. Absence of the header line = legacy v1
-        # (chunks-only) — accepted by `parse_manifest` for read but no
-        # longer produced by `stream_packs`.
-        has_tables = bool(state["table_refs"])
-        if has_tables:
-            manifest_lines.append(
-                json.dumps({"_format_version": 2}, sort_keys=True)
-            )
-            # Table refs first, sorted by tar member name for determinism.
-            # content_hash binds pack_hash to the file bytes; two writers
-            # with same tables produce same hashes.
-            for member_name, content_hash, table_size in sorted(state["table_refs"]):
-                manifest_lines.append(json.dumps(
-                    {
-                        "table_file": member_name,
-                        "hash": content_hash,
-                        "size": table_size,
-                    },
-                    sort_keys=True,
-                ))
-
+        # Pack v3 chunk-pack manifest:
+        #   header line  : {"_format_version": 3}
+        #   kind line    : {"_kind": "chunks"}
+        #   chunk lines  : {"leaf_hash": "...", "size": N}
+        # The chunk pack is anonymous w.r.t. its metadata pack — the
+        # backlink lives in the metadata pack's _chunk_pack_hashes list.
+        manifest_lines = [
+            json.dumps({"_format_version": 3}, sort_keys=True),
+            json.dumps({"_kind": PACK_KIND_CHUNKS}, sort_keys=True),
+        ]
         entries = []
         for leaf_hash in entries_sorted:
             size = state["entries_by_hash"][leaf_hash]
@@ -614,32 +627,6 @@ def stream_packs(
 
     state = _start_pack()
     try:
-        # Pack format v2: ship the metadata table dumps as the FIRST tar
-        # members of the first pack. Stream from disk — tarfile.addfile
-        # reads the file handle in chunks, never loading the whole thing
-        # into RAM. A 4 GB edges.jsonl thus uses ~tens of KB peak RAM
-        # instead of 4 GB per worker. TarInfo built by hand (NOT via
-        # gettarinfo) so mtime/uid/gid are zeroed — two writers' packs
-        # converge byte-for-byte regardless of filesystem timestamps.
-        #
-        # Each extra_member carries (tar_name, source_path, content_hash).
-        # The content_hash goes into the manifest so pack_hash binds the
-        # full pack (tables + chunks). Caller computed the hash via
-        # hash_file_leaf() at dump time.
-        for member_name, source_path, content_hash in pending_extras:
-            info = tarfile.TarInfo(name=member_name)
-            file_size = source_path.stat().st_size
-            info.size = file_size
-            # mtime, uid, gid, mode default to 0 / "" — deterministic.
-            with open(source_path, "rb") as f:
-                state["tar"].addfile(info, f)
-            state["table_refs"].append((member_name, content_hash, file_size))
-        # Force the compressor to flush so size accounting reflects the
-        # metadata write before we start streaming chunks.
-        if pending_extras:
-            state["writer"].flush(zstandard.FLUSH_BLOCK)
-            state["file_obj"].flush()
-
         for leaf_hash, body in chunks:
             if len(leaf_hash) != 64:
                 raise ValueError(f"bad leaf_hash length: {leaf_hash}")
@@ -693,6 +680,112 @@ def stream_packs(
         raise
 
 
+def build_metadata_pack(
+    table_files: list[tuple[str, Path, str]],
+    *,
+    snapshot_root: str,
+    chunk_pack_hashes: list[str],
+    work_dir: Path | str | None = None,
+    level: int = 3,
+) -> FilePack:
+    """Build ONE metadata pack from the shard's table dumps.
+
+    `table_files` is a list of (tar_member_name, source_file_path,
+    content_hash) — the same shape as the old extra_members. Caller has
+    already dumped each shipped table to a JSONL file on disk.
+
+    The metadata pack's manifest carries:
+      {"_format_version": 3}
+      {"_kind": "metadata"}
+      {"_snapshot_root": "..."}            # corpus state at pack time
+      {"_chunk_pack_hashes": [...]}        # every chunk pack covering this shard
+      {"table_file": ..., "hash": ..., "size": ...} per table
+
+    `pack_hash = hash_leaf(manifest_bytes).hex()` — content-addressed,
+    so two writers producing the same shard state + same chunk packs
+    converge on the same metadata_pack_hash.
+
+    Streaming write to disk (tempfile in work_dir): worker RAM stays
+    bounded by zstd internal buffers regardless of metadata size.
+    """
+    work_path = Path(work_dir) if work_dir is not None else Path(tempfile.gettempdir())
+    work_path.mkdir(parents=True, exist_ok=True)
+
+    fd, path_str = tempfile.mkstemp(
+        dir=work_path,
+        prefix="arborist-meta-pack-",
+        suffix=".tar.zst.tmp",
+    )
+    file_obj = os.fdopen(fd, "wb")
+    cctx = zstandard.ZstdCompressor(level=level)
+    writer = cctx.stream_writer(file_obj, closefd=False)
+    tar = tarfile.open(fileobj=writer, mode="w|")
+
+    try:
+        # Tar entries: tables/<name>.jsonl per shipped table.
+        # Stream from disk (handle iteration), zeroed TarInfo metadata for
+        # determinism.
+        table_refs_sorted: list[tuple[str, str, int]] = []
+        for member_name, source_path, content_hash in sorted(table_files):
+            info = tarfile.TarInfo(name=member_name)
+            file_size = source_path.stat().st_size
+            info.size = file_size
+            with open(source_path, "rb") as f:
+                tar.addfile(info, f)
+            table_refs_sorted.append((member_name, content_hash, file_size))
+
+        # Build manifest. Sort chunk_pack_hashes for determinism.
+        chunk_pack_hashes_sorted = sorted(set(chunk_pack_hashes))
+        manifest_lines = [
+            json.dumps({"_format_version": 3}, sort_keys=True),
+            json.dumps({"_kind": PACK_KIND_METADATA}, sort_keys=True),
+            json.dumps({"_snapshot_root": snapshot_root}, sort_keys=True),
+            json.dumps(
+                {"_chunk_pack_hashes": chunk_pack_hashes_sorted},
+                sort_keys=True,
+            ),
+        ]
+        for member_name, content_hash, table_size in table_refs_sorted:
+            manifest_lines.append(json.dumps(
+                {
+                    "table_file": member_name,
+                    "hash": content_hash,
+                    "size": table_size,
+                },
+                sort_keys=True,
+            ))
+        manifest_bytes = ("\n".join(manifest_lines) + "\n").encode("utf-8")
+
+        # Manifest goes LAST in the tar — open_pack iterates by name and
+        # ignores order, and writing-last lets us include chunk_pack_hashes
+        # (which weren't known when tar started).
+        info = tarfile.TarInfo(name="manifest.ndjson")
+        info.size = len(manifest_bytes)
+        tar.addfile(info, io.BytesIO(manifest_bytes))
+        tar.close()
+        writer.close()
+        file_obj.close()
+        pack_hash = hash_leaf(manifest_bytes).hex()
+        body_size = Path(path_str).stat().st_size
+        return FilePack(
+            pack_hash=pack_hash,
+            manifest_bytes=manifest_bytes,
+            body_path=Path(path_str),
+            body_size=body_size,
+            entries=tuple(),  # metadata pack has no PackEntry-style chunks
+        )
+    except BaseException:
+        try: tar.close()
+        except Exception: pass
+        try: writer.close()
+        except Exception: pass
+        try: file_obj.close()
+        except Exception: pass
+        try: Path(path_str).unlink()
+        except FileNotFoundError: pass
+        raise
+
+
 def open_pack(body_bytes: bytes) -> Iterator[tuple[str, bytes]]:
     """Stream (leaf_hash, body) pairs from a pack body.
 
@@ -748,21 +841,39 @@ class TableRef:
 
 @dataclass(frozen=True)
 class ParsedManifest:
-    """Result of parsing a pack manifest. Distinguishes v1 (chunks-only)
-    from v2 (tables + chunks) so consumers can branch on which
-    invariants to verify."""
-    format_version: int           # 1 (legacy) or 2 (table-bound)
-    tables: tuple[TableRef, ...]   # empty for v1
+    """Result of parsing a pack manifest.
+
+    - format_version: 1 (legacy chunks-only), 2 (legacy tables + chunks),
+      or 3 (SPV-split: pure metadata or pure chunks).
+    - kind: "metadata" | "chunks" | None (for v1/v2 mixed).
+    - snapshot_root: corpus state the pack covers (v3 only).
+    - chunk_pack_hashes: every chunk pack a v3 metadata pack references.
+    - tables: TableRef list (empty for chunks-only packs).
+    - chunks: PackEntry list (empty for metadata-only packs).
+    """
+    format_version: int
+    kind: str | None
+    snapshot_root: str | None
+    chunk_pack_hashes: tuple[str, ...]
+    tables: tuple[TableRef, ...]
     chunks: tuple[PackEntry, ...]
 
 
 def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
-    """Parse a manifest.ndjson stream into table refs + chunk entries.
+    """Parse a manifest.ndjson stream.
 
-    Format detection: a line `{"_format_version": N}` at the top means
-    v2. Absent means legacy v1 (chunks-only). Lenient on blank lines.
+    Format detection:
+      - line `{"_format_version": N}` at top → that version.
+      - absence → legacy v1.
+      - v3 packs additionally carry `_kind` (metadata/chunks),
+        `_snapshot_root`, and (metadata only) `_chunk_pack_hashes`.
+
+    Lenient on blank lines and unknown record shapes (forward-compat).
     """
     fmt = 1
+    kind: str | None = None
+    snapshot_root: str | None = None
+    chunk_pack_hashes: list[str] = []
     tables: list[TableRef] = []
     chunks: list[PackEntry] = []
     for line in manifest_bytes.decode("utf-8").splitlines():
@@ -772,6 +883,12 @@ def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
         rec = json.loads(line)
         if "_format_version" in rec:
             fmt = int(rec["_format_version"])
+        elif "_kind" in rec:
+            kind = str(rec["_kind"])
+        elif "_snapshot_root" in rec:
+            snapshot_root = str(rec["_snapshot_root"])
+        elif "_chunk_pack_hashes" in rec:
+            chunk_pack_hashes = [str(h) for h in rec["_chunk_pack_hashes"]]
         elif "table_file" in rec:
             tables.append(TableRef(
                 member_name=rec["table_file"],
@@ -785,6 +902,9 @@ def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
         # silently ignore unknown record shapes (forward compat)
     return ParsedManifest(
         format_version=fmt,
+        kind=kind,
+        snapshot_root=snapshot_root,
+        chunk_pack_hashes=tuple(chunk_pack_hashes),
         tables=tuple(tables),
         chunks=tuple(chunks),
     )
diff --git a/arborist/evict.py b/arborist/evict.py
index 7b1cec0..5a1c69d 100644
--- a/arborist/evict.py
+++ b/arborist/evict.py
@@ -314,7 +314,11 @@ def push_pack(
     import tempfile
     from pathlib import Path
 
-    from arborist.cold_object import hash_file_leaf, stream_packs
+    from arborist.cold_object import (
+        build_metadata_pack,
+        hash_file_leaf,
+        stream_packs,
+    )
     from arborist.cold_pack_metadata import dump_shard_metadata
     from arborist.snapshot import compute_snapshot_root
 
@@ -402,75 +406,114 @@ def push_pack(
     # upload path is also memory-bounded.
     import shutil
 
-    # Pack format v2: dump every shipped table to JSONL in a temp dir, then
-    # include those files as `tables/<name>.jsonl` tar members at the front
-    # of the first pack. Edges goes through fan-in restructure inside
-    # `dump_shard_metadata` for ~5-10x compression vs the live-schema row
-    # layout. Lives only long enough to be tarred + uploaded, then deleted.
+    # SPV-split (pack format v3): every shard produces TWO artifact kinds.
     #
-    # Each table file's content_hash is computed via hash_file_leaf (stream
-    # the file through sha256 in 1 MB chunks) and passed into stream_packs
-    # so the pack manifest binds pack_hash to (tables + chunks), not just
-    # chunks. Two writers with the same shard state → same pack_hash.
+    #   1. N chunk packs (each ≤ cap, content-addressed by chunk-only
+    #      manifest) — heavy, optional. A "just-enough" consumer skips.
+    #   2. ONE metadata pack (small, content-addressed by table+chunk-
+    #      pack-refs manifest) — light, always pulled by any consumer.
+    #
+    # The metadata pack's manifest carries `_chunk_pack_hashes`, so a
+    # "full" consumer iterates that list and pulls the chunk packs.
+    # Producer always emits both; consumer chooses sync mode.
+
+    # Phase A: dump every shipped table to JSONL on disk.
     metadata_dir = Path(tempfile.mkdtemp(prefix="arborist-meta-"))
     metadata_summary = dump_shard_metadata(conn, metadata_dir)
-    extra_members = [
+    table_files = [
         (
             f"tables/{name}.jsonl",
             info["path"],
             hash_file_leaf(info["path"]),
         )
         for name, info in metadata_summary.items()
+        if info["rows"] > 0  # skip tables that exist but have no data rows;
+                              # avoids shipping bare-header jsonl files in a
+                              # metadata pack for a freshly-initialized DB
     ]
 
+    # Phase B: stream chunk packs. Each chunk pack is anonymous (no
+    # backlink to the metadata pack — the metadata pack is the entry
+    # point). Collect each chunk pack's hash for the metadata manifest.
+    chunk_pack_hashes_in_order: list[str] = []
     for pack in stream_packs(
         _chunk_source(),
         max_compressed_bytes=max_pack_bytes,
-        extra_members=extra_members,
     ):
         pack_uncompressed = uncompressed_running[0] - last_finalized_uncompressed[0]
         last_finalized_uncompressed[0] = uncompressed_running[0]
         try:
             if push_to_bucket:
-                backend.put_pack_file(pack.pack_hash, pack.body_path)
-                backend.put_pack_manifest(pack.pack_hash, pack.manifest_bytes)
+                backend.put_pack_file(pack.pack_hash, pack.body_path, kind="chunks")
+                backend.put_pack_manifest(pack.pack_hash, pack.manifest_bytes, kind="chunks")
             if out_dir is not None:
                 short = pack.pack_hash[:16]
-                final_path = out_dir / f"arborist-pack-{short}.tar.zst"
-                # Move (cheap if on same filesystem) instead of read+write.
+                final_path = out_dir / f"arborist-pack-{short}.chunks.tar.zst"
                 shutil.move(str(pack.body_path), str(final_path))
-                (out_dir / f"arborist-pack-{short}.manifest.ndjson").write_bytes(
+                (out_dir / f"arborist-pack-{short}.chunks.manifest.ndjson").write_bytes(
                     pack.manifest_bytes
                 )
-            # Deliberately NO audit_append for the push itself. The pack is
-            # its own receipt (content-addressed file + snapshot_root binding
-            # inside the pack's metadata). An audit row here would leak into
-            # the next push's audit_events.jsonl dump and break the
-            # "two writers at the same corpus state produce the same pack_hash"
-            # determinism property. The bucket / disc IS the audit trail for
-            # the push.
+            chunk_pack_hashes_in_order.append(pack.pack_hash)
             pack_results.append({
                 "pack_hash": pack.pack_hash,
+                "kind": "chunks",
                 "chunk_count": len(pack.entries),
                 "uncompressed_bytes": pack_uncompressed,
                 "compressed_bytes": pack.body_size,
                 "snapshot_root": snapshot_root,
             })
         finally:
-            # Clean up the temp file when neither local_dir nor an exception
-            # claimed it. `shutil.move` already removed it on the local_dir
-            # path; this handles the push-only path and the audit-failure
-            # case.
             if pack.body_path.exists():
                 try:
                     pack.body_path.unlink()
                 except OSError:
                     pass
 
+    # Phase C: build + push the metadata pack now that we know every
+    # chunk pack's hash. The metadata pack's manifest includes
+    # _chunk_pack_hashes so the consumer can iterate them for full-sync.
+    metadata_pack_hash: str | None = None
+    metadata_pack_bytes: int = 0
+    if table_files:
+        meta_pack = build_metadata_pack(
+            table_files,
+            snapshot_root=snapshot_root,
+            chunk_pack_hashes=chunk_pack_hashes_in_order,
+        )
+        try:
+            if push_to_bucket:
+                backend.put_pack_file(
+                    meta_pack.pack_hash, meta_pack.body_path, kind="metadata"
+                )
+                backend.put_pack_manifest(
+                    meta_pack.pack_hash, meta_pack.manifest_bytes, kind="metadata"
+                )
+            if out_dir is not None:
+                short = meta_pack.pack_hash[:16]
+                final_path = out_dir / f"arborist-pack-{short}.metadata.tar.zst"
+                shutil.move(str(meta_pack.body_path), str(final_path))
+                (out_dir / f"arborist-pack-{short}.metadata.manifest.ndjson").write_bytes(
+                    meta_pack.manifest_bytes
+                )
+            metadata_pack_hash = meta_pack.pack_hash
+            metadata_pack_bytes = meta_pack.body_size
+            pack_results.append({
+                "pack_hash": meta_pack.pack_hash,
+                "kind": "metadata",
+                "table_count": len(table_files),
+                "chunk_pack_hashes": list(chunk_pack_hashes_in_order),
+                "compressed_bytes": meta_pack.body_size,
+                "snapshot_root": snapshot_root,
+            })
+        finally:
+            if meta_pack.body_path.exists():
+                try:
+                    meta_pack.body_path.unlink()
+                except OSError:
+                    pass
+
     skipped_hash_mismatch = nonlocal_skipped[0]
 
-    # Clean up the temp dir that held the table JSONL dumps. The tar
-    # already inlined their bytes; the on-disk copies are no longer needed.
     try:
         shutil.rmtree(metadata_dir, ignore_errors=True)
     except Exception:
@@ -483,9 +526,16 @@ def push_pack(
         "status": "pushed" if push_to_bucket else "written",
         "packs": pack_results,
         "pack_count": len(pack_results),
-        "total_chunks": sum(p["chunk_count"] for p in pack_results),
-        "total_uncompressed_bytes": sum(p["uncompressed_bytes"] for p in pack_results),
+        "metadata_pack_hash": metadata_pack_hash,
+        "chunk_pack_hashes": chunk_pack_hashes_in_order,
+        "total_chunks": sum(
+            p.get("chunk_count", 0) for p in pack_results if p["kind"] == "chunks"
+        ),
+        "total_uncompressed_bytes": sum(
+            p.get("uncompressed_bytes", 0) for p in pack_results if p["kind"] == "chunks"
+        ),
         "total_compressed_bytes": sum(p["compressed_bytes"] for p in pack_results),
+        "metadata_pack_bytes": metadata_pack_bytes,
         "snapshot_root": snapshot_root,
         "snapshot_doc_count": doc_count_at_pack,
         "snapshot_ts": pack_ts,
@@ -495,21 +545,94 @@ def push_pack(
     }
 
 
-def pull_pack(
+def pull_metadata_pack(
     conn: sqlite3.Connection,
     backend: "ObjectStoreBackend",
     pack_hash: str,
 ) -> dict:
-    """Pull one pack from the bucket and restore everything it contains:
-    metadata tables (pack v2) + chunk bodies.
+    """Pull a metadata pack and restore every shipped table into the
+    local schema. Chunks rows land with `content=NULL` and `tier='cold'`
+    — the metadata view is fully populated but chunk bodies are empty.
 
-    Order matters: tables go first (so `chunks` rows exist before we try
-    to fill their content), then chunk bodies fill `chunks.content` and
-    populate `chunks_fts`. INSERT OR IGNORE on table restore means
-    re-pulling a pack against a populated DB is idempotent.
+    This is the "just-enough" hydration path: schema is queryable
+    immediately for documents / audit chain / edges / Merkle structure;
+    chunk-body queries fall through to JIT fetch (future) or to a
+    follow-up `pull_chunk_pack` against each `_chunk_pack_hashes` entry.
 
-    v1 packs (chunks-only, no `tables/` entries) still work — the metadata
-    restore step becomes a no-op when no table files are found in the tar.
+    Returns the parsed manifest's chunk_pack_hashes so a "full" consumer
+    can iterate them.
+    """
+    return _pull_pack_inner(conn, backend, pack_hash, kind="metadata")
+
+
+def pull_chunk_pack(
+    conn: sqlite3.Connection,
+    backend: "ObjectStoreBackend",
+    pack_hash: str,
+) -> dict:
+    """Pull a chunk pack and fill `chunks.content` + `chunks_fts` for
+    every leaf_hash whose row already exists locally. Idempotent against
+    a populated DB (skips rows already content-filled).
+
+    Assumes `pull_metadata_pack` has already run (or schema otherwise
+    populated) — chunks without a matching row are reported as
+    `skipped_unknown` and skipped.
+    """
+    return _pull_pack_inner(conn, backend, pack_hash, kind="chunks")
+
+
+def hydrate_from_metadata_pack(
+    conn: sqlite3.Connection,
+    backend: "ObjectStoreBackend",
+    metadata_pack_hash: str,
+    *,
+    mode: str = "just-enough",
+) -> dict:
+    """Higher-level consumer entry point. Modes:
+
+    - `"just-enough"`: pull only the metadata pack. Schema fully
+      restored, every chunks row has `content=NULL`. Node is
+      immediately queryable for the metadata graph (documents, edges,
+      audit chain, etc.). Chunk-body queries return null until a future
+      JIT-fetch path (cache miss → CDN/mesh) fills them.
+
+    - `"full"`: pull metadata pack, then iterate
+      `_chunk_pack_hashes` from its manifest and pull every chunk pack.
+      Final state: full corpus offline-queryable.
+    """
+    if mode not in ("just-enough", "full"):
+        raise ValueError(f"unknown sync mode: {mode!r}")
+
+    meta_result = pull_metadata_pack(conn, backend, metadata_pack_hash)
+    chunk_results = []
+    if mode == "full":
+        for ch_hash in meta_result.get("chunk_pack_hashes_referenced", []):
+            chunk_results.append(pull_chunk_pack(conn, backend, ch_hash))
+
+    return {
+        "status": "hydrated",
+        "mode": mode,
+        "metadata_pack_hash": metadata_pack_hash,
+        "metadata_pull": meta_result,
+        "chunk_pulls": chunk_results,
+        "chunk_packs_pulled": len(chunk_results),
+    }
+
+
+def _pull_pack_inner(
+    conn: sqlite3.Connection,
+    backend: "ObjectStoreBackend",
+    pack_hash: str,
+    *,
+    kind: str,
+) -> dict:
+    """Shared pull-and-restore body for metadata + chunks packs.
+
+    `kind` ∈ {"metadata", "chunks"} drives both the bucket key suffix
+    and which restore phase runs. Manifest's `_kind` field is checked
+    on extraction; a kind mismatch raises ValueError (defends against
+    a bucket where a chunks-key has been overwritten with metadata
+    contents).
     """
     import io
     import shutil
@@ -524,7 +647,7 @@ def pull_pack(
     from arborist.merkle import hash_leaf
 
     backend_id = backend.identity.to_audit_body()
-    body = backend.get_pack(pack_hash)
+    body = backend.get_pack(pack_hash, kind=kind)
 
     # Streaming decompression — packs built by stream_packs lack the
     # content-size frame header so the one-shot ZstdDecompressor.decompress
@@ -657,17 +780,23 @@ def pull_pack(
                 )
             restored += 1
 
+    # Parse manifest a second time for chunk_pack_hashes surface (for the
+    # metadata-pack-pull caller — it needs the list to drive full-sync).
+    chunk_pack_hashes_referenced: tuple[str, ...] = ()
+    if manifest_bytes is not None:
+        chunk_pack_hashes_referenced = parse_manifest(manifest_bytes).chunk_pack_hashes
+
     with transaction(conn):
         append_audit(
             conn,
             event_type="cold_pack_pulled",
             body={
                 "pack_hash": pack_hash,
+                "pack_kind": kind,
                 "chunks_restored": restored,
                 "chunks_skipped_already_hot": skipped_already_hot,
                 "chunks_skipped_unknown_locally": skipped_unknown,
                 "tables_restored": tables_restored,
-                "pack_format": "v2" if table_files_seen else "v1",
                 "backend": backend_id,
             },
         )
@@ -675,7 +804,8 @@ def pull_pack(
     return {
         "status": "pulled",
         "pack_hash": pack_hash,
-        "pack_format": "v2" if table_files_seen else "v1",
+        "pack_kind": kind,
+        "chunk_pack_hashes_referenced": list(chunk_pack_hashes_referenced),
         "tables_restored": tables_restored,
         "chunks_restored": restored,
         "chunks_skipped_already_hot": skipped_already_hot,
diff --git a/tests/test_cold_object.py b/tests/test_cold_object.py
index 434b198..e4e60f9 100644
--- a/tests/test_cold_object.py
+++ b/tests/test_cold_object.py
@@ -23,7 +23,9 @@ from arborist.cold_object import (
 from arborist.compress import unpack_chunk
 from arborist.document import Document
 from arborist.evict import (
-    pull_pack,
+    hydrate_from_metadata_pack,
+    pull_chunk_pack,
+    pull_metadata_pack,
     push_pack,
 )
 from arborist.ingest import ingest_source
@@ -60,10 +62,13 @@ LONG = (
 
 def test_pack_key_layout():
     h = "b" * 64
-    assert pack_key(h).endswith(".tar.zst")
-    assert pack_key(h, manifest=True).endswith(".manifest.ndjson")
-    # Different scheme so a `list_keys("packs/")` can distinguish.
+    # Default kind="chunks". v3 SPV-split keys carry the kind in the path.
+    assert pack_key(h).endswith(".chunks.tar.zst")
+    assert pack_key(h, manifest=True).endswith(".chunks.manifest.ndjson")
+    assert pack_key(h, kind="metadata").endswith(".metadata.tar.zst")
+    assert pack_key(h, kind="metadata", manifest=True).endswith(".metadata.manifest.ndjson")
     assert pack_key(h) != pack_key(h, manifest=True)
+    assert pack_key(h, kind="chunks") != pack_key(h, kind="metadata")
 
 
 def test_build_pack_is_content_addressed():
@@ -101,10 +106,9 @@ def test_parse_manifest_round_trip():
     leaf = hash_leaf(body).hex()
     pack = build_pack([(leaf, body)])
     parsed = parse_manifest(pack.manifest_bytes)
-    # build_pack produces a chunks-only manifest (no tables) — that's the
-    # v1 shape because build_pack doesn't accept extra_members. Tests of
-    # the v2 shape live in test_push_pack_v2_hydrates_fresh_empty_db.
-    assert parsed.format_version == 1
+    # build_pack produces a chunks-only v3 manifest.
+    assert parsed.format_version == 3
+    assert parsed.kind == "chunks"
     assert parsed.tables == ()
     assert len(parsed.chunks) == 1
     assert parsed.chunks[0].leaf_hash == leaf
@@ -151,7 +155,9 @@ def test_memory_backend_identity_carries_no_credentials():
 # ---------------------------------------------------------------------------
 
 
-def test_push_pack_then_pull_pack_restores_chunks(tmp_path):
+def test_push_pack_then_pull_full_restores_chunks(tmp_path):
+    """Push produces 1 metadata pack + N chunk packs. Full hydration
+    (mode="full") pulls metadata then iterates chunk_pack_hashes."""
     db = tmp_path / "pack.db"
     conn = connect(db)
     backend = MemoryBackend()
@@ -161,41 +167,85 @@ def test_push_pack_then_pull_pack_restores_chunks(tmp_path):
             "SELECT document_root FROM documents WHERE document_uri='html://p'"
         ).fetchone()["document_root"]
 
-        # Pack while content is still local. Tiny corpus → one pack.
         push_result = push_pack(conn, backend, document_root=root)
         assert push_result["status"] == "pushed"
-        assert push_result["pack_count"] == 1
-        # Pack is bound to a snapshot_root — packs are delayed point-in-time
-        # snapshots; the audit row pins which corpus state this pack covers
-        # so falsifications between repacks produce a different pack_hash.
-        assert len(push_result["snapshot_root"]) == 64
-        first = push_result["packs"][0]
-        assert first["chunk_count"] > 0
-        assert first["compressed_bytes"] <= first["uncompressed_bytes"] + 200
-        assert first["snapshot_root"] == push_result["snapshot_root"]
-        pack_hash = first["pack_hash"]
-        assert backend.head(pack_key(pack_hash))
-        assert backend.head(pack_key(pack_hash, manifest=True))
+        # SPV split: at least 1 metadata pack + at least 1 chunks pack.
+        assert push_result["metadata_pack_hash"] is not None
+        assert len(push_result["chunk_pack_hashes"]) >= 1
 
-        # Now NULL local content so pull_pack has something to do.
-        conn.execute("UPDATE chunks SET content = NULL, tier='cold'")
-        conn.execute("DELETE FROM chunks_fts")
-        conn.commit()
+        metadata_hash = push_result["metadata_pack_hash"]
+        # Both pack kinds are in the bucket at their kind-specific keys.
+        assert backend.head(pack_key(metadata_hash, kind="metadata"))
+        assert backend.head(pack_key(metadata_hash, kind="metadata", manifest=True))
+        for ch in push_result["chunk_pack_hashes"]:
+            assert backend.head(pack_key(ch, kind="chunks"))
 
-        pull_result = pull_pack(conn, backend, pack_hash)
-        assert pull_result["status"] == "pulled"
-        assert pull_result["chunks_restored"] == first["chunk_count"]
+        # NULL local content + delete documents so a full pull has to
+        # restore the whole schema, not just chunks.
+        with conn:
+            conn.execute("DELETE FROM chunks_fts")
+            conn.execute("DELETE FROM chunks")
+            conn.execute("DELETE FROM documents")
 
+        result = hydrate_from_metadata_pack(
+            conn, backend, metadata_hash, mode="full",
+        )
+        assert result["status"] == "hydrated"
+        assert result["mode"] == "full"
+        assert result["chunk_packs_pulled"] == len(push_result["chunk_pack_hashes"])
+
+        # Schema restored.
+        assert conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0] == 1
+        # All chunks restored to hot tier with content.
         cold = conn.execute(
-            "SELECT COUNT(*) FROM chunks WHERE tier='cold'"
+            "SELECT COUNT(*) FROM chunks WHERE tier='cold' OR content IS NULL"
         ).fetchone()[0]
         assert cold == 0
     finally:
         conn.close()
 
 
+def test_push_pack_then_just_enough_hydrate_leaves_chunks_null(tmp_path):
+    """just-enough mode: pull only metadata pack. Schema fully restored
+    but chunks rows have content=NULL — ready for JIT-fetch later."""
+    db = tmp_path / "je.db"
+    conn = connect(db)
+    backend = MemoryBackend()
+    try:
+        ingest_source(conn, FakeSource([_doc("html://je", LONG)]))
+        push_result = push_pack(conn, backend)
+        metadata_hash = push_result["metadata_pack_hash"]
+        expected_chunks = sum(p.get("chunk_count", 0) for p in push_result["packs"]
+                              if p.get("kind") == "chunks")
+    finally:
+        conn.close()
+
+    rx_db = tmp_path / "rx.db"
+    conn_rx = connect(rx_db)
+    try:
+        result = hydrate_from_metadata_pack(
+            conn_rx, backend, metadata_hash, mode="just-enough",
+        )
+        assert result["status"] == "hydrated"
+        assert result["mode"] == "just-enough"
+        assert result["chunk_packs_pulled"] == 0
+        # Documents row exists.
+        assert conn_rx.execute("SELECT COUNT(*) FROM documents").fetchone()[0] == 1
+        # Chunks rows exist (metadata restored) but content is NULL.
+        n_chunks = conn_rx.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
+        n_null = conn_rx.execute(
+            "SELECT COUNT(*) FROM chunks WHERE content IS NULL"
+        ).fetchone()[0]
+        assert n_chunks == expected_chunks
+        assert n_null == n_chunks
+    finally:
+        conn_rx.close()
+
+
 def test_push_pack_idempotent(tmp_path):
-    """Same chunk set → same pack_hash(es) → bucket overwrite is a no-op."""
+    """Same shard state on two runs produces the same pack_hashes:
+    metadata_pack_hash + every chunk_pack_hash. Bucket overwrite is a
+    no-op."""
     db = tmp_path / "pack-id.db"
     conn = connect(db)
     backend = MemoryBackend()
@@ -203,9 +253,8 @@ def test_push_pack_idempotent(tmp_path):
         ingest_source(conn, FakeSource([_doc("html://p", LONG)]))
         a = push_pack(conn, backend)
         b = push_pack(conn, backend)
-        assert [p["pack_hash"] for p in a["packs"]] == [
-            p["pack_hash"] for p in b["packs"]
-        ]
+        assert a["metadata_pack_hash"] == b["metadata_pack_hash"]
+        assert a["chunk_pack_hashes"] == b["chunk_pack_hashes"]
     finally:
         conn.close()
 
@@ -258,31 +307,38 @@ def test_push_pack_splits_to_fit_dvdr(tmp_path):
 
         # Tar trailer (~1 KB padding) + zstd frame footer get emitted after
         # the last in-loop size check, so overshoot is bounded by trailer
-        # size + one chunk's worth of compressed bytes.
+        # size + one chunk's worth of compressed bytes. Only the chunks
+        # packs are subject to the cap; the metadata pack is its own
+        # artifact and can be any size.
         OVERSHOOT_SLACK = 4_096
-        for p in result["packs"]:
+        chunk_packs = [p for p in result["packs"] if p["kind"] == "chunks"]
+        assert len(chunk_packs) > 1, "small cap should force multi-chunks-pack split"
+        for p in chunk_packs:
             assert p["compressed_bytes"] <= CAP + OVERSHOOT_SLACK, (
-                f"pack {p['pack_hash'][:8]} compressed={p['compressed_bytes']} "
-                f"exceeds cap {CAP}+slack {OVERSHOOT_SLACK}"
+                f"chunks pack {p['pack_hash'][:8]} compressed="
+                f"{p['compressed_bytes']} exceeds cap {CAP}+slack"
             )
             assert p["chunk_count"] >= 1
 
-        # Round-trip: NULL local content, pull every pack back.
+        # Round-trip: NULL local content, hydrate-full from metadata pack.
         conn.execute("UPDATE chunks SET content = NULL, tier='cold'")
         conn.execute("DELETE FROM chunks_fts")
         conn.commit()
-
-        total_restored = 0
-        for p in result["packs"]:
-            r = pull_pack(conn, backend, p["pack_hash"])
-            total_restored += r["chunks_restored"]
+        hyd = hydrate_from_metadata_pack(
+            conn, backend, result["metadata_pack_hash"], mode="full",
+        )
+        assert hyd["status"] == "hydrated"
+        # Total chunks restored == total chunks across all chunk packs.
+        total_restored = sum(r["chunks_restored"] for r in hyd["chunk_pulls"])
         assert total_restored == result["total_chunks"]
     finally:
         conn.close()
 
 
 def test_push_pack_local_dir_writes_files(tmp_path):
-    """--local-dir writes pack + manifest files for burning to physical media."""
+    """--local-dir writes pack + manifest files for burning to physical media.
+    v3: both metadata pack and chunk packs land in local-dir, with kind
+    suffix in the filename."""
     db = tmp_path / "burn.db"
     conn = connect(db)
     backend = MemoryBackend()
@@ -293,13 +349,17 @@ def test_push_pack_local_dir_writes_files(tmp_path):
         assert result["status"] == "pushed"
         assert result["local_dir"] == str(burn_dir)
         assert burn_dir.exists()
-        pack_files = sorted(burn_dir.glob("arborist-pack-*.tar.zst"))
-        manifest_files = sorted(burn_dir.glob("arborist-pack-*.manifest.ndjson"))
-        assert len(pack_files) == result["pack_count"]
-        assert len(manifest_files) == result["pack_count"]
-        # Local files are byte-identical to what we'd burn.
-        for pack_file, summary in zip(pack_files, result["packs"]):
-            assert pack_file.stat().st_size == summary["compressed_bytes"]
+        # v3 filenames: arborist-pack-<short>.{metadata,chunks}.tar.zst
+        metadata_files = sorted(burn_dir.glob("arborist-pack-*.metadata.tar.zst"))
+        chunks_files = sorted(burn_dir.glob("arborist-pack-*.chunks.tar.zst"))
+        assert len(metadata_files) == 1
+        assert len(chunks_files) == len(result["chunk_pack_hashes"])
+        # Manifest sidecars present too.
+        assert len(sorted(burn_dir.glob("arborist-pack-*.metadata.manifest.ndjson"))) == 1
+        assert (
+            len(sorted(burn_dir.glob("arborist-pack-*.chunks.manifest.ndjson")))
+            == len(result["chunk_pack_hashes"])
+        )
     finally:
         conn.close()
 
@@ -373,44 +433,44 @@ def test_edges_fan_in_batches_huge_destinations(tmp_path):
         conn.close()
 
 
-def test_v2_pack_hash_binds_table_contents(tmp_path):
-    """Two packs with the same chunk set but different table content
-    must NOT collide on pack_hash.
-
-    The v1 bug: pack_hash was hash_leaf(chunk-only-manifest), so
-    same-chunks + different-tables produced same pack_hash. Bucket
-    overwrites or mesh-peer disagreement → silent table swap. v2
-    fixes this by including content_hash for each tables/<name>.jsonl
-    in the manifest before chunk entries.
-
-    This test forces the issue by ingesting two different document
-    sets that happen to share a chunk — wait, that's impossible by
-    construction. Instead we check the manifest format directly: a
-    v2 pack's manifest has a `_format_version: 2` header and
-    `table_file` entries before `leaf_hash` entries.
-    """
-    db = tmp_path / "v2bind.db"
+def test_v3_metadata_pack_manifest_shape(tmp_path):
+    """v3 metadata pack manifest carries: _format_version=3, _kind="metadata",
+    _snapshot_root, _chunk_pack_hashes list, and one table_file entry per
+    non-empty shipped table. Chunks pack manifest separately carries
+    _kind="chunks" and leaf_hash entries."""
+    db = tmp_path / "v3shape.db"
     conn = connect(db)
     backend = MemoryBackend()
     try:
         ingest_source(conn, FakeSource([_doc("html://a", LONG)]))
         result = push_pack(conn, backend)
-        manifest = backend.get_pack_manifest(result["packs"][0]["pack_hash"])
-        parsed = parse_manifest(manifest)
+        metadata_hash = result["metadata_pack_hash"]
+        chunk_hashes = result["chunk_pack_hashes"]
+        assert metadata_hash is not None
+        assert len(chunk_hashes) >= 1
 
-        assert parsed.format_version == 2
-        # All shipped tables that had rows should appear in the manifest.
-        # documents + chunks + merkle_nodes + audit_events minimum.
-        table_names = {t.member_name for t in parsed.tables}
+        meta_manifest = backend.get_pack_manifest(metadata_hash, kind="metadata")
+        meta = parse_manifest(meta_manifest)
+        assert meta.format_version == 3
+        assert meta.kind == "metadata"
+        assert meta.snapshot_root and len(meta.snapshot_root) == 64
+        assert list(meta.chunk_pack_hashes) == chunk_hashes
+        assert meta.chunks == ()
+        table_names = {t.member_name for t in meta.tables}
         assert "tables/documents.jsonl" in table_names
         assert "tables/chunks.jsonl" in table_names
         assert "tables/audit_events.jsonl" in table_names
-        # Every table_file ref carries a hash and size.
-        for t in parsed.tables:
-            assert len(t.content_hash) == 64  # sha256 hex
+        for t in meta.tables:
+            assert len(t.content_hash) == 64
             assert t.size > 0
-        # Chunks are also there.
-        assert len(parsed.chunks) > 0
+
+        # First chunks pack: chunks only, no tables.
+        first_chunks_manifest = backend.get_pack_manifest(chunk_hashes[0], kind="chunks")
+        ch = parse_manifest(first_chunks_manifest)
+        assert ch.format_version == 3
+        assert ch.kind == "chunks"
+        assert ch.tables == ()
+        assert len(ch.chunks) > 0
     finally:
         conn.close()
 
@@ -433,14 +493,14 @@ def test_pull_pack_rejects_tampered_table(tmp_path):
     try:
         ingest_source(conn, FakeSource([_doc("html://t", LONG)]))
         result = push_pack(conn, backend)
-        pack_hash = result["packs"][0]["pack_hash"]
+        metadata_hash = result["metadata_pack_hash"]
     finally:
         conn.close()
 
-    # Decompress the pack, rewrite documents.jsonl with corrupted bytes,
-    # repack, replace the pack in the bucket. Manifest stays unchanged
-    # so the recorded table content_hash no longer matches.
-    body = backend.get_pack(pack_hash)
+    # Decompress the metadata pack, rewrite documents.jsonl with corrupted
+    # bytes, repack, replace in the bucket. Manifest stays unchanged so
+    # the recorded table content_hash no longer matches.
+    body = backend.get_pack(metadata_hash, kind="metadata")
     raw_tar = _zstd.ZstdDecompressor().stream_reader(_io.BytesIO(body)).read()
     out_buf = _io.BytesIO()
     cctx = _zstd.ZstdCompressor(level=3)
@@ -454,20 +514,20 @@ def test_pull_pack_rejects_tampered_table(tmp_path):
                     data = f.read()
                     if m.name == "tables/documents.jsonl":
                         data = data + b'\n{"injected": "tamper"}\n'
-                        info = _tarfile.TarInfo(name=m.name)
-                        info.size = len(data)
-                    else:
-                        info = _tarfile.TarInfo(name=m.name)
-                        info.size = len(data)
+                    info = _tarfile.TarInfo(name=m.name)
+                    info.size = len(data)
                     out_tar.addfile(info, _io.BytesIO(data))
-    backend._tamper(f"packs/{pack_hash}.tar.zst", out_buf.getvalue())
+    backend._tamper(
+        f"packs/{metadata_hash}.metadata.tar.zst", out_buf.getvalue()
+    )
 
-    # Fresh DB; expect ValueError on pull.
+    # Fresh DB; expect ValueError on hydrate (metadata pack tamper fires
+    # the hash-check during pull_metadata_pack).
     rx_db = tmp_path / "rx.db"
     conn_rx = connect(rx_db)
     try:
         with pytest.raises(ValueError, match="table-hash mismatch"):
-            pull_pack(conn_rx, backend, pack_hash)
+            pull_metadata_pack(conn_rx, backend, metadata_hash)
     finally:
         conn_rx.close()
 
@@ -507,7 +567,8 @@ def test_push_pack_v2_hydrates_fresh_empty_db(tmp_path):
 
         result = push_pack(conn_src, backend)
         assert result["status"] == "pushed"
-        pack_hash = result["packs"][0]["pack_hash"]
+        metadata_hash = result["metadata_pack_hash"]
+        assert metadata_hash is not None
     finally:
         conn_src.close()
 
@@ -520,20 +581,19 @@ def test_push_pack_v2_hydrates_fresh_empty_db(tmp_path):
             "SELECT COUNT(*) FROM documents"
         ).fetchone()[0] == 0
 
-        pull_result = pull_pack(conn_rx, backend, pack_hash)
-        assert pull_result["status"] == "pulled"
-        assert pull_result["pack_format"] == "v2"
-        # Every table that was non-empty in src should have rows now in rx.
-        assert pull_result["tables_restored"]["documents"] == src_docs
-        assert pull_result["tables_restored"]["chunks"] == src_chunks
-        assert pull_result["tables_restored"]["merkle_nodes"] == src_merkle
-        assert pull_result["tables_restored"]["audit_events"] == src_audit
-        # Chunk bodies restored.
+        hyd = hydrate_from_metadata_pack(
+            conn_rx, backend, metadata_hash, mode="full",
+        )
+        assert hyd["status"] == "hydrated"
+        assert hyd["mode"] == "full"
+        assert conn_rx.execute(
+            "SELECT COUNT(*) FROM documents"
+        ).fetchone()[0] == src_docs
+        # Chunks rows + content restored.
         rx_hot = conn_rx.execute(
             "SELECT COUNT(*) FROM chunks WHERE tier='hot' AND content IS NOT NULL"
         ).fetchone()[0]
         assert rx_hot == src_chunks
-        # FTS5 rows rebuilt.
         rx_fts = conn_rx.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0]
         assert rx_fts == src_chunks
         # All three documents reachable by URI.