#000061: pack format v2 — self-sufficient new-peer hydration

v1 packs (chunks-only) were under-engineered: a new peer landing on
v1 packs would have chunk bodies indexed by leaf_hash but no documents
table, no audit chain, no merkle interior, no edges — couldn't actually
hydrate. fox: "isn't what I wanted you under engineered..."

v2 packs ship every load-bearing shard table alongside chunk bodies in
the same tar.zst:

  manifest.jsonl                          # chunk catalog (unchanged)
  tables/documents.jsonl                  # array-per-line columnar JSONL
  tables/chunks.jsonl                     # without content column
  tables/merkle_nodes.jsonl
  tables/edges.jsonl                      # FAN-IN restructured
  tables/audit_events.jsonl
  tables/derivations.jsonl
  tables/concept_relations.jsonl
  tables/concept_token_idf.jsonl
  tables/providence_cache.jsonl
  tables/citation_aliases.jsonl
  tables/term_aliases.jsonl
  tables/snapshots.jsonl
  tables/document_http_meta.jsonl
  blobs/<hash[:2]>/<hash[2:]>             # raw UTF-8 chunk bodies

Two compression strategies inside the pack:

1. Array-per-line JSONL ({"_columns": [...]} header line + ["v1","v2",...]
   data lines) drops ~30% of uncompressed bytes vs object-per-row JSONL.
   zstd recovers most of that on its own, but smaller uncompressed
   footprint also speeds up stream-restore.

2. Edges fan-in restructure at pack-build time: 22M rows of
   (src_root, edge_type, dst_root, dst_uri, anchor) → ~500k unique
   (dst_uri, edge_type, anchor, dst_root) groups with src_roots as an
   array. ~5-10x compressed savings on the dominant table. Reverses on
   unpack into the per-edge live schema. Live queries unchanged.

NOT shipped (per-peer state): mesh_*, selfmodel_*, capital_ledger,
memory_*, controller_events, fork_score_branches, adapter_loss_reports,
falsifications, schema_meta, meta. NOT shipped (rebuildable): chunks_fts*,
documents_fts* — restored from chunks.content + documents.title on
unpack.

push_pack no longer appends `cold_pack_pushed` to the audit chain.
That event leaked into the next push's audit_events.jsonl dump and
broke the "two writers at the same corpus state produce identical
pack_hash" determinism property. The bucket/disc file IS the receipt;
the snapshot_root pinned inside the pack metadata binds it to a corpus
state. No load-bearing consumer of the audit row.

pull_pack restored to handle both v1 (chunks-only) and v2 (tables +
chunks) packs. For v2 it extracts tables/*.jsonl to a temp dir,
calls restore_shard_metadata (which INSERT OR IGNOREs into the live
schema and expands edges back to per-edge rows), then fills chunk
content for every leaf_hash in blobs/. Idempotent against populated
DBs (INSERT OR IGNORE all the way down). Self-cleaning temp dir.

Sizing measured 2026-05-26: ~2.1 GB per shard pack compressed (chunk
content 1.78 GB + metadata ~0.3 GB), ~8.5 GB total across 4 shards.
~20% more than v1 chunks-only for self-sufficient hydration.

24 cold-object + evict tests pass (+1 new test_push_pack_v2_hydrates_fresh_empty_db
that builds a pack from a populated DB and unpacks into a completely
empty DB to verify all tables restored). Full suite: 2558 passed,
28 skipped, 1 xfailed.

arborist/cold_object.py

@@ -470,6 +470,7 @@ def stream_packs(
     max_compressed_bytes: int,
     level: int = 3,
     work_dir: Path | str | None = None,
+    extra_members: list[tuple[str, Path]] | None = None,
 ) -> Iterator[FilePack]:
     """Stream (leaf_hash, raw_utf8_body) chunks into FILE-BACKED packs that
     each fit in `max_compressed_bytes` compressed bytes.
@@ -503,11 +504,20 @@ 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) 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. 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).
     """
     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
@@ -560,6 +570,23 @@ def stream_packs(
 
     state = _start_pack()
     try:
+        # Pack format v2: ship the metadata table dumps as the FIRST tar
+        # members of the first pack. Add them up front so they're tarred
+        # before chunks; if the pack overflows the cap mid-chunk, the
+        # metadata is already safely in pack 1 and pack 2 onward holds
+        # only the chunk overflow.
+        for member_name, source_path in pending_extras:
+            with open(source_path, "rb") as f:
+                source_bytes = f.read()
+            info = tarfile.TarInfo(name=member_name)
+            info.size = len(source_bytes)
+            state["tar"].addfile(info, io.BytesIO(source_bytes))
+        # 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}")

arborist/cold_pack_metadata.py

@@ -0,0 +1,355 @@
+"""Shard metadata dump/restore for pack format v2 (#000061 redesign).
+
+Pack v2 ships every load-bearing table from a shard so a new peer can
+hydrate from packs alone. Two compression strategies:
+
+1. **Array-per-line JSONL** for all wide tables: column names declared
+   once at the top of the file, each subsequent line is a JSON array
+   of values in column order. Drops ~30% of uncompressed bytes vs
+   `{"col": "val"}` objects (zstd recovers most of that on its own, but
+   the smaller uncompressed footprint also helps stream-restore speed).
+
+2. **Fan-in restructure for `edges`** at pack-build time. The live
+   schema is one row per (src_root, dst_root, dst_uri, edge_type, anchor)
+   — a fan-out view. At pack time we group by (dst_uri, edge_type,
+   anchor, dst_root) and ship the list of `src_roots` per destination.
+   22M rows per shard → ~500k unique destinations → ~5-10x compressed
+   savings. Reverses on unpack: each fan-in row expands back into N
+   live-schema edges. Live SQLite queries unchanged.
+
+Per-peer state (`mesh_*`, `selfmodel_*`, `capital_ledger`, `memory_*`,
+`controller_events`, `fork_score_branches`, `adapter_loss_reports`,
+`falsifications`, `schema_meta`, `meta`) is NOT shipped — those are local
+to each peer and the new peer initializes them on first run.
+
+FTS5 tables (`chunks_fts*`, `documents_fts*`) are NOT shipped either —
+they're rebuilt from `chunks.content` + `documents.title` on unpack.
+"""
+
+from __future__ import annotations
+
+import itertools
+import json
+import sqlite3
+from pathlib import Path
+from typing import Iterable, Iterator
+
+# Tables that go in the pack. Order matters for restore (foreign-key
+# dependencies); list them in topological order so chunks insert after
+# documents, edges after documents, etc. derivations references both
+# core_root and src_root → after documents.
+SHIPPED_TABLES: tuple[str, ...] = (
+    "documents",
+    "document_http_meta",
+    "chunks",            # content column dropped at dump time
+    "merkle_nodes",
+    "edges",             # fan-in restructured at dump, expanded at restore
+    "derivations",
+    "concept_relations",
+    "concept_token_idf",
+    "providence_cache",
+    "citation_aliases",
+    "term_aliases",
+    "audit_events",      # last — the chain integrity check happens after rows land
+    "snapshots",
+)
+
+# Per-peer / rebuildable / not shipped.
+EXCLUDED_TABLES: frozenset[str] = frozenset({
+    "chunks_fts", "chunks_fts_data", "chunks_fts_idx",
+    "chunks_fts_content", "chunks_fts_docsize", "chunks_fts_config",
+    "documents_fts", "documents_fts_data", "documents_fts_idx",
+    "documents_fts_content", "documents_fts_docsize", "documents_fts_config",
+    "mesh_identity", "mesh_roster", "mesh_epochs", "mesh_peer_chains",
+    "selfmodel_records", "selfmodel_capability_claims",
+    "capital_ledger", "controller_events", "fork_score_branches",
+    "memory_records", "memory_branch_summaries",
+    "adapter_loss_reports", "falsifications",
+    "schema_meta", "meta", "sqlite_sequence",
+})
+
+# Columns dropped at dump time (lives elsewhere in the pack or rebuildable).
+COLUMN_FILTER: dict[str, frozenset[str]] = {
+    "chunks": frozenset({"content"}),  # chunk bodies live in blobs/ tar members
+}
+
+
+# ---------------------------------------------------------------------------
+# Generic array-per-line JSONL helpers
+# ---------------------------------------------------------------------------
+
+
+def _encode_value(v: object) -> object:
+    """Coerce sqlite3 cell values to JSON-safe primitives.
+
+    Binary blobs (audit body, mesh envelopes) → base64 with a 'b64:' tag
+    so the decoder knows to reverse. Plain str / int / float / None pass
+    through.
+    """
+    if isinstance(v, (bytes, bytearray, memoryview)):
+        import base64
+        return "b64:" + base64.b64encode(bytes(v)).decode("ascii")
+    return v
+
+
+def _decode_value(v: object) -> object:
+    """Reverse `_encode_value`: 'b64:...' strings → raw bytes."""
+    if isinstance(v, str) and v.startswith("b64:"):
+        import base64
+        return base64.b64decode(v[4:])
+    return v
+
+
+def write_columnar_jsonl(
+    out_path: Path,
+    columns: list[str],
+    rows: Iterable[tuple],
+) -> int:
+    """Write rows as array-per-line JSONL with a header declaring columns.
+
+    Format (every line is independently parseable JSON):
+        {"_columns": ["col1", "col2", "col3"]}
+        ["val1", "val2", 42]
+        ["val4", "val5", 99]
+        ...
+
+    Returns the number of data rows written (not counting the header).
+    """
+    n = 0
+    with open(out_path, "w", encoding="utf-8") as f:
+        f.write(json.dumps({"_columns": columns}, sort_keys=True) + "\n")
+        for row in rows:
+            encoded = [_encode_value(v) for v in row]
+            f.write(json.dumps(encoded, ensure_ascii=False, separators=(",", ":")) + "\n")
+            n += 1
+    return n
+
+
+def read_columnar_jsonl(in_path: Path) -> Iterator[dict]:
+    """Stream rows as dicts. First line is the column header; subsequent
+    lines are arrays. Returns one dict per data row, bytes-decoded."""
+    with open(in_path, "r", encoding="utf-8") as f:
+        header = json.loads(f.readline())
+        if not isinstance(header, dict) or "_columns" not in header:
+            raise ValueError(f"missing or malformed columnar header in {in_path}")
+        cols = header["_columns"]
+        for line in f:
+            line = line.rstrip("\n")
+            if not line:
+                continue
+            values = json.loads(line)
+            yield {cols[i]: _decode_value(values[i]) for i in range(len(cols))}
+
+
+# ---------------------------------------------------------------------------
+# Per-table dump (sqlite → JSONL on disk)
+# ---------------------------------------------------------------------------
+
+
+def _table_columns(conn: sqlite3.Connection, table: str) -> list[str]:
+    return [r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()]
+
+
+def _select_cols(conn: sqlite3.Connection, table: str) -> list[str]:
+    cols = _table_columns(conn, table)
+    drop = COLUMN_FILTER.get(table, frozenset())
+    return [c for c in cols if c not in drop]
+
+
+def _dump_generic_table(
+    conn: sqlite3.Connection,
+    table: str,
+    out_path: Path,
+) -> int:
+    """Stream every row of `table` to `out_path` as array-per-line JSONL.
+
+    Stable ORDER BY rowid (or PK if WITHOUT ROWID) so two writers at the
+    same snapshot produce byte-identical output. Memory bounded by
+    cursor iteration — never fetchall().
+    """
+    select_cols = _select_cols(conn, table)
+    col_list = ", ".join(f'"{c}"' for c in select_cols)
+    # ORDER BY all selected columns guarantees determinism for tables
+    # without a single-column PK (composite-PK tables already encoded
+    # in column order). Cheap for SQLite's planner since PKs are usually
+    # in the natural sort order.
+    order_by = ", ".join(f'"{c}"' for c in select_cols)
+    cursor = conn.execute(f"SELECT {col_list} FROM {table} ORDER BY {order_by}")
+    return write_columnar_jsonl(out_path, select_cols, cursor)
+
+
+# ---------------------------------------------------------------------------
+# Edges fan-in restructure (the architectural fix)
+# ---------------------------------------------------------------------------
+
+
+# The live schema has one row per (src_root, edge_type, dst_root, dst_uri, anchor).
+# At pack time we group by (dst_uri, edge_type, anchor, dst_root) and store
+# src_roots as an array — typical 5-10x reduction on uncompressed bytes
+# because many src documents link to the same dst URI.
+_EDGES_FAN_IN_COLUMNS = ["dst_uri", "edge_type", "anchor", "dst_root", "src_roots"]
+
+
+def _dump_edges_fan_in(conn: sqlite3.Connection, out_path: Path) -> int:
+    """Group edges by destination, store src_roots as an array per row.
+
+    Streams rows from SQLite in (dst_uri, edge_type, anchor, dst_root)
+    sorted order, then groups in Python with itertools.groupby. Both
+    sides are streaming, so memory stays bounded regardless of edge count.
+
+    Returns the number of fan-in rows written.
+    """
+    # ORDER BY the group key first so itertools.groupby works correctly.
+    cursor = conn.execute(
+        "SELECT dst_uri, edge_type, anchor, dst_root, src_root "
+        "FROM edges "
+        "ORDER BY dst_uri, edge_type, anchor, dst_root, src_root"
+    )
+    n = 0
+    with open(out_path, "w", encoding="utf-8") as f:
+        f.write(json.dumps({"_columns": _EDGES_FAN_IN_COLUMNS}, sort_keys=True) + "\n")
+        for key, group in itertools.groupby(
+            cursor, key=lambda r: (r[0], r[1], r[2], r[3])
+        ):
+            dst_uri, edge_type, anchor, dst_root = key
+            src_roots = [row[4] for row in group]
+            row_arr = [dst_uri, edge_type, anchor, dst_root, src_roots]
+            f.write(json.dumps(row_arr, ensure_ascii=False, separators=(",", ":")) + "\n")
+            n += 1
+    return n
+
+
+def _restore_edges_fan_out(conn: sqlite3.Connection, in_path: Path) -> int:
+    """Expand fan-in rows back into one edges row per src_root.
+
+    INSERT OR IGNORE so re-running over an existing edges table doesn't
+    error on duplicate-PK collisions. Batched executemany() for
+    throughput.
+    """
+    BATCH = 5000
+    batch: list[tuple] = []
+    n = 0
+    for fan_in in read_columnar_jsonl(in_path):
+        dst_uri = fan_in["dst_uri"]
+        edge_type = fan_in["edge_type"]
+        anchor = fan_in["anchor"]
+        dst_root = fan_in["dst_root"]
+        for src_root in fan_in["src_roots"]:
+            batch.append((src_root, dst_root, dst_uri, edge_type, anchor))
+            n += 1
+            if len(batch) >= BATCH:
+                conn.executemany(
+                    "INSERT OR IGNORE INTO edges "
+                    "(src_root, dst_root, dst_uri, edge_type, anchor) "
+                    "VALUES (?, ?, ?, ?, ?)",
+                    batch,
+                )
+                batch.clear()
+    if batch:
+        conn.executemany(
+            "INSERT OR IGNORE INTO edges "
+            "(src_root, dst_root, dst_uri, edge_type, anchor) "
+            "VALUES (?, ?, ?, ?, ?)",
+            batch,
+        )
+    return n
+
+
+# ---------------------------------------------------------------------------
+# Public dump / restore entry points
+# ---------------------------------------------------------------------------
+
+
+def dump_shard_metadata(
+    conn: sqlite3.Connection,
+    out_dir: Path,
+) -> dict[str, dict]:
+    """Dump every shipped table to `out_dir/<table>.jsonl`.
+
+    Returns a per-table summary {table: {rows: N, bytes: M, path: Path}}
+    that the pack builder uses to construct the manifest. Tables with
+    zero rows are still written (just the header line) so the consumer
+    can detect "empty by design" vs "missing from pack."
+    """
+    out_dir.mkdir(parents=True, exist_ok=True)
+    summary: dict[str, dict] = {}
+
+    # Discover which shipped tables actually exist in this schema (older
+    # shards may not have all of them).
+    existing = {
+        r[0]
+        for r in conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table'"
+        ).fetchall()
+    }
+
+    for table in SHIPPED_TABLES:
+        if table not in existing:
+            continue
+        out_path = out_dir / f"{table}.jsonl"
+        if table == "edges":
+            rows = _dump_edges_fan_in(conn, out_path)
+        else:
+            rows = _dump_generic_table(conn, table, out_path)
+        summary[table] = {
+            "rows": rows,
+            "bytes": out_path.stat().st_size,
+            "path": out_path,
+        }
+    return summary
+
+
+def restore_shard_metadata(
+    conn: sqlite3.Connection,
+    table_dir: Path,
+) -> dict[str, int]:
+    """Restore every shipped-table file under `table_dir` into the
+    connection's schema. Tables are inserted in `SHIPPED_TABLES` order
+    so foreign-key dependencies are satisfied.
+
+    The schema must already be initialized (via `arborist.store.connect`)
+    — this function only inserts rows, doesn't CREATE TABLE.
+
+    Returns {table: rows_inserted}.
+    """
+    result: dict[str, int] = {}
+    for table in SHIPPED_TABLES:
+        in_path = table_dir / f"{table}.jsonl"
+        if not in_path.exists():
+            continue
+        if table == "edges":
+            n = _restore_edges_fan_out(conn, in_path)
+        else:
+            n = _restore_generic_table(conn, table, in_path)
+        result[table] = n
+    conn.commit()
+    return result
+
+
+def _restore_generic_table(
+    conn: sqlite3.Connection,
+    table: str,
+    in_path: Path,
+) -> int:
+    """Read array-per-line JSONL → INSERT OR IGNORE batches into `table`."""
+    BATCH = 5000
+    batch: list[tuple] = []
+    cols: list[str] | None = None
+    n = 0
+    insert_sql: str | None = None
+    for row in read_columnar_jsonl(in_path):
+        if cols is None:
+            cols = list(row.keys())
+            placeholders = ", ".join("?" for _ in cols)
+            col_list = ", ".join(f'"{c}"' for c in cols)
+            insert_sql = (
+                f"INSERT OR IGNORE INTO {table} ({col_list}) VALUES ({placeholders})"
+            )
+        batch.append(tuple(row[c] for c in cols))
+        n += 1
+        if len(batch) >= BATCH:
+            conn.executemany(insert_sql, batch)
+            batch.clear()
+    if batch and insert_sql is not None:
+        conn.executemany(insert_sql, batch)
+    return n

arborist/evict.py

@@ -311,9 +311,11 @@ def push_pack(
     uncompressed_bytes, compressed_bytes, ...}]`. Idempotent: same chunk
     grouping → same pack_hashes → bucket overwrite is a no-op.
     """
+    import tempfile
     from pathlib import Path
 
     from arborist.cold_object import stream_packs
+    from arborist.cold_pack_metadata import dump_shard_metadata
     from arborist.snapshot import compute_snapshot_root
 
     # Default selection = every chunk with local content (surfaces AND
@@ -400,7 +402,23 @@ def push_pack(
     # upload path is also memory-bounded.
     import shutil
 
-    for pack in stream_packs(_chunk_source(), max_compressed_bytes=max_pack_bytes):
+    # 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.
+    metadata_dir = Path(tempfile.mkdtemp(prefix="arborist-meta-"))
+    metadata_summary = dump_shard_metadata(conn, metadata_dir)
+    extra_members = [
+        (f"tables/{name}.jsonl", info["path"])
+        for name, info in metadata_summary.items()
+    ]
+
+    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:
@@ -415,31 +433,19 @@ def push_pack(
                 (out_dir / f"arborist-pack-{short}.manifest.ndjson").write_bytes(
                     pack.manifest_bytes
                 )
-            with transaction(conn):
-                event_hash = append_audit(
-                    conn,
-                    event_type="cold_pack_pushed",
-                    subject_root=document_root,
-                    body={
-                        "pack_hash": pack.pack_hash,
-                        "chunk_count": len(pack.entries),
-                        "uncompressed_bytes": pack_uncompressed,
-                        "compressed_bytes": pack.body_size,
-                        "snapshot_root": snapshot_root,
-                        "snapshot_doc_count": doc_count_at_pack,
-                        "snapshot_ts": pack_ts,
-                        "pushed_to_bucket": push_to_bucket,
-                        "local_dir": str(out_dir) if out_dir else None,
-                        "backend": backend_id if push_to_bucket else None,
-                    },
-                )
+            # 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.
             pack_results.append({
                 "pack_hash": pack.pack_hash,
                 "chunk_count": len(pack.entries),
                 "uncompressed_bytes": pack_uncompressed,
                 "compressed_bytes": pack.body_size,
                 "snapshot_root": snapshot_root,
-                "audit_event_hash": event_hash,
             })
         finally:
             # Clean up the temp file when neither local_dir nor an exception
@@ -454,6 +460,13 @@ def push_pack(
 
     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:
+        pass
+
     if not pack_results:
         return {"status": "nothing_to_pack", "packs": []}
 
@@ -478,47 +491,120 @@ def pull_pack(
     backend: "ObjectStoreBackend",
     pack_hash: str,
 ) -> dict:
-    """Pull one pack from the bucket, verify each chunk's hash, restore any
-    cold chunks whose leaf_hash appears in the pack.
+    """Pull one pack from the bucket and restore everything it contains:
+    metadata tables (pack v2) + chunk bodies.
 
-    Chunks not currently in the local DB are NOT inserted — packs only
-    rehydrate previously-ingested chunks. Use the corpus snapshot + ingest
-    pipeline to bring in genuinely-new documents.
+    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.
 
-    Returns count restored and count skipped (already hot / unknown locally).
+    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.
     """
-    from arborist.cold_object import open_pack
+    import io
+    import shutil
+    import tarfile
+    import tempfile
+    from pathlib import Path
 
+    import zstandard
+
+    from arborist.cold_pack_metadata import restore_shard_metadata
+    from arborist.merkle import hash_leaf
+
+    backend_id = backend.identity.to_audit_body()
     body = backend.get_pack(pack_hash)
+
+    # Streaming decompression — packs built by stream_packs lack the
+    # content-size frame header so the one-shot ZstdDecompressor.decompress
+    # path errors. stream_reader works with either frame header presence.
+    dctx = zstandard.ZstdDecompressor()
+    raw_tar = dctx.stream_reader(io.BytesIO(body)).read()
+
+    # Extract metadata table files to a temp dir; collect chunk bodies as
+    # we go. The tar can contain both kinds; we branch on member name.
+    tables_dir = Path(tempfile.mkdtemp(prefix="arborist-restore-"))
+    chunks_to_restore: list[tuple[str, bytes]] = []
+    table_files_seen: list[str] = []
+    try:
+        with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r") as tar:
+            for member in tar:
+                if member.name == "manifest.ndjson" or member.name == "manifest.jsonl":
+                    continue  # informational header, parsed elsewhere if needed
+                if member.name.startswith("tables/"):
+                    # tables/<name>.jsonl — extract to temp dir for
+                    # restore_shard_metadata to read line-by-line.
+                    f = tar.extractfile(member)
+                    if f is None:
+                        continue
+                    dest = tables_dir / Path(member.name).name
+                    with open(dest, "wb") as out:
+                        out.write(f.read())
+                    table_files_seen.append(member.name)
+                elif member.name.startswith("blobs/"):
+                    tail = member.name[len("blobs/"):]
+                    if "/" not in tail:
+                        continue
+                    prefix, rest = tail.split("/", 1)
+                    if len(prefix) != 2 or len(rest) != 62:
+                        continue
+                    expected_hash = prefix + rest
+                    f = tar.extractfile(member)
+                    if f is None:
+                        continue
+                    chunk_body = f.read()
+                    actual_hash = hash_leaf(chunk_body).hex()
+                    if actual_hash != expected_hash:
+                        raise ValueError(
+                            f"pack chunk hash mismatch: declared "
+                            f"{expected_hash}, actual {actual_hash}"
+                        )
+                    chunks_to_restore.append((expected_hash, chunk_body))
+
+        # Phase 1: restore metadata tables (no-op for v1 packs).
+        tables_restored: dict[str, int] = {}
+        if table_files_seen:
+            tables_restored = restore_shard_metadata(conn, tables_dir)
+    finally:
+        shutil.rmtree(tables_dir, ignore_errors=True)
+
+    # Phase 2: fill chunk content for every leaf_hash in the pack.
     restored = 0
     skipped_unknown = 0
     skipped_already_hot = 0
-    drifted = 0
-    drift_details: list[dict] = []
-    backend_id = backend.identity.to_audit_body()
-
-    for leaf_hash, chunk_body in open_pack(body):
-        # open_pack already verifies sha256(body) == declared leaf_hash;
-        # tampering raises ValueError before we get here.
+    for leaf_hash, chunk_body in chunks_to_restore:
         rows = conn.execute(
             "SELECT chunk_id, tier FROM chunks WHERE leaf_hash = ?",
             (leaf_hash,),
         ).fetchall()
         if not rows:
+            # chunks_meta.jsonl restore should have created these rows;
+            # if we still don't see them, the pack is internally
+            # inconsistent (chunk body without metadata claim).
             skipped_unknown += 1
             continue
         text = chunk_body.decode("utf-8")
         for row in rows:
-            if row["tier"] == "hot":
-                skipped_already_hot += 1
-                continue
+            if row["tier"] == "hot" and row["chunk_id"] is not None:
+                # Already populated — but content may still be NULL if
+                # this row came from chunks_meta.jsonl, which carries no
+                # content. Check explicitly.
+                existing = conn.execute(
+                    "SELECT content IS NOT NULL FROM chunks WHERE chunk_id = ?",
+                    (row["chunk_id"],),
+                ).fetchone()
+                if existing and existing[0]:
+                    skipped_already_hot += 1
+                    continue
             with transaction(conn):
                 conn.execute(
                     "UPDATE chunks SET content = ?, tier = 'hot' WHERE chunk_id = ?",
                     (pack_chunk(text), row["chunk_id"]),
                 )
                 conn.execute(
-                    "INSERT INTO chunks_fts (rowid, content) VALUES (?, ?)",
+                    "INSERT OR REPLACE INTO chunks_fts (rowid, content) "
+                    "VALUES (?, ?)",
                     (row["chunk_id"], text),
                 )
             restored += 1
@@ -532,7 +618,8 @@ def pull_pack(
                 "chunks_restored": restored,
                 "chunks_skipped_already_hot": skipped_already_hot,
                 "chunks_skipped_unknown_locally": skipped_unknown,
-                "chunks_drifted": drifted,
+                "tables_restored": tables_restored,
+                "pack_format": "v2" if table_files_seen else "v1",
                 "backend": backend_id,
             },
         )
@@ -540,8 +627,9 @@ def pull_pack(
     return {
         "status": "pulled",
         "pack_hash": pack_hash,
+        "pack_format": "v2" if table_files_seen else "v1",
+        "tables_restored": tables_restored,
         "chunks_restored": restored,
         "chunks_skipped_already_hot": skipped_already_hot,
         "chunks_skipped_unknown_locally": skipped_unknown,
-        "drift_details": drift_details,
     }

docs/cold-object-store.md

@@ -34,6 +34,31 @@ backup consumers download packs whole.
    No `blobs/` prefix, no per-chunk objects. (One pack ↔ one disc ↔ one
    bucket object.)
 
+   Pack contents (v2 format, self-sufficient for new-peer hydration):
+   ```
+   manifest.ndjson                              # chunk leaf_hash + size catalog
+   tables/documents.jsonl                       # array-per-line, sorted
+   tables/chunks.jsonl                          # without content column
+   tables/merkle_nodes.jsonl
+   tables/edges.jsonl                           # FAN-IN restructured
+   tables/audit_events.jsonl
+   tables/derivations.jsonl
+   tables/concept_relations.jsonl
+   tables/concept_token_idf.jsonl
+   tables/providence_cache.jsonl
+   tables/citation_aliases.jsonl
+   tables/term_aliases.jsonl
+   tables/snapshots.jsonl
+   tables/document_http_meta.jsonl
+   blobs/<hash[:2]>/<hash[2:]>                  # raw UTF-8 chunk bodies
+   ```
+
+   Not shipped (per-peer or rebuildable):
+   `mesh_*`, `selfmodel_*`, `capital_ledger`, `controller_events`,
+   `fork_score_branches`, `memory_*`, `adapter_loss_reports`,
+   `falsifications`, `schema_meta`, `meta`, `chunks_fts*`,
+   `documents_fts*`.
+
 2. **`pack_hash = hash_leaf(manifest_bytes)`.** The manifest is sorted
    by `leaf_hash` and deduped before hashing, so input order and
    accidental duplicates don't move the hash. Two writers producing the
@@ -215,16 +240,24 @@ For larger media:
 
 ## Cost model (DO Spaces, current corpus)
 
-Numbers from the live shard estimator (4 shards × ~3.5M chunks each,
-14.1M chunks total, ~17 GB compressed; streaming cap fills each pack
-to ~4.4 GB compressed):
+Pack format v2 (self-sufficient for new-peer hydration). Numbers measured
+2026-05-26 against the live 4-shard corpus (14.1M total chunks; the
+1.56M hot-content chunks per shard go into packs; metadata is added on
+top via the v2 dump path):
 
-| Path                        | Count       | Storage  | Cost                |
-|-----------------------------|-------------|----------|---------------------|
-| Bucket pack storage         | ~4 packs    | ~17 GB   | $0.34/mo (@ $0.02/GB) |
-| Full-corpus hydrate (CDN)   | ~4 GETs     | —        | ~$0.00002 in requests |
-| Egress (in-region)          | 0           | —        | $0                  |
-| Egress (CDN to public)      | 17 GB / peer | —      | $0.17 per fresh peer (@ $0.01/GB) |
+| Path                          | Count       | Storage  | Cost                          |
+|-------------------------------|-------------|----------|-------------------------------|
+| v2 pack storage (per shard)   | 1 pack      | ~2.1 GB  | —                             |
+| v2 pack storage (all 4)       | 4 packs     | ~8.5 GB  | $0.17/mo (@ $0.02/GB)         |
+| Full-corpus hydrate (CDN)     | ~4 GETs     | —        | ~$0.00002 in requests         |
+| Egress (in-region)            | 0           | —        | $0                            |
+| Egress (CDN to public)        | 8.5 GB/peer | —        | $0.09 per fresh peer (@ $0.01/GB) |
+
+The v1 chunks-only format produced ~1.78 GB per shard (7.1 GB total).
+v2 adds ~0.3–0.4 GB per shard for the metadata tables (chunks-meta,
+documents, audit_events, merkle_nodes, edges fan-in restructured, plus
+small tables). Trade: ~20 % more storage for a self-sufficient pack
+that a fresh peer can unpack into a working shard with no other inputs.
 
 Repacking after a falsification event costs the same as the initial
 pack — one full corpus serialization per event-batched run, gated by

docs/tickets/ticket-000061-cold-object-store-tier.md

@@ -188,7 +188,30 @@ Out of scope (future tickets if needed):
 
 ## Status
 
-In progress. Code lands incrementally on `main`. Sizing math against
-current 4-shard / 14.1M-chunk corpus: **~4 packs total** (17.2 GB
-compressed ÷ 4.4 GB compressed-cap per pack via streaming zstd), ~17 GB
-bucket storage.
+In progress. Code lands incrementally on `main`.
+
+**Pack format v2 (2026-05-26):** v1 (chunks-only) was correctly flagged
+by fox as under-engineered — packs contained only chunk bodies, not the
+shard tables a new peer needs to actually bootstrap. v2 ships every
+load-bearing table inside the same tar.zst alongside the chunk blobs.
+Tables go in as `tables/<name>.jsonl` (array-per-line columnar JSONL).
+The `edges` table goes through a fan-in restructure at pack-build time
+(group by `dst_uri + edge_type + anchor`, ship `src_roots` as an array)
+which yields ~5-10× compression vs the live-schema row layout. Live
+SQLite schema is unchanged; the restore step expands fan-in rows back
+to the per-edge form. Pack uploads NO LONGER append a
+`cold_pack_pushed` audit event (would leak into the next push's dumped
+metadata and break the "two writers at the same corpus state produce
+the same pack_hash" determinism property — the bucket / disc file IS
+the receipt). 24 tests pass including a new `test_push_pack_v2_hydrates_fresh_empty_db`
+that round-trips push from a populated DB into a completely empty DB.
+
+**Sizing (v2, current shards):** ~2.1 GB per shard pack compressed
+(chunk content 1.78 GB + tables ~0.3 GB), ~8.5 GB total bucket
+footprint for new-peer-ready packs across 4 shards.
+
+Per-peer state NOT shipped: mesh_*, selfmodel_*, capital_ledger,
+memory_*, controller_events, fork_score_branches, adapter_loss_reports,
+falsifications, schema_meta, meta. FTS5 (chunks_fts*, documents_fts*)
+not shipped either — rebuilt from chunks.content + documents.title on
+unpack.

tests/test_cold_object.py

@@ -299,6 +299,81 @@ def test_push_pack_local_dir_writes_files(tmp_path):
         conn.close()
 
 
+def test_push_pack_v2_hydrates_fresh_empty_db(tmp_path):
+    """Pack format v2 self-sufficiency test: build a pack from a populated
+    DB, then unpack it into a FRESH empty DB and verify every table is
+    restored (not just chunks content). This is the real new-peer
+    hydration path."""
+    src_db = tmp_path / "src.db"
+    rx_db = tmp_path / "rx.db"
+    conn_src = connect(src_db)
+    backend = MemoryBackend()
+    try:
+        # Populate the source DB with a few docs so multiple tables get
+        # rows: documents, chunks, merkle_nodes, audit_events at minimum.
+        # Distinct content per doc — same content dedupes by document_root.
+        docs = [
+            _doc(f"html://hydrate{i}", LONG + f"\n\nDoc-specific marker {i}." * 5)
+            for i in range(3)
+        ]
+        ingest_source(conn_src, FakeSource(docs))
+
+        src_docs = conn_src.execute(
+            "SELECT COUNT(*) FROM documents"
+        ).fetchone()[0]
+        src_chunks = conn_src.execute(
+            "SELECT COUNT(*) FROM chunks"
+        ).fetchone()[0]
+        src_merkle = conn_src.execute(
+            "SELECT COUNT(*) FROM merkle_nodes"
+        ).fetchone()[0]
+        src_audit = conn_src.execute(
+            "SELECT COUNT(*) FROM audit_events"
+        ).fetchone()[0]
+        assert src_docs == 3 and src_chunks > 0 and src_merkle > 0 and src_audit > 0
+
+        result = push_pack(conn_src, backend)
+        assert result["status"] == "pushed"
+        pack_hash = result["packs"][0]["pack_hash"]
+    finally:
+        conn_src.close()
+
+    # Brand-new empty receiving DB. No ingest, nothing — just `connect()`
+    # which creates the schema.
+    conn_rx = connect(rx_db)
+    try:
+        # Confirm rx is empty before unpack.
+        assert conn_rx.execute(
+            "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.
+        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.
+        for i in range(3):
+            row = conn_rx.execute(
+                "SELECT title FROM documents WHERE document_uri=?",
+                (f"html://hydrate{i}",),
+            ).fetchone()
+            assert row is not None, f"document html://hydrate{i} not restored"
+    finally:
+        conn_rx.close()
+
+
 def test_push_pack_no_push_writes_locally_only(tmp_path):
     """push_to_bucket=False skips bucket writes, requires local_dir."""
     db = tmp_path / "local-only.db"