feat: #000061 cold-pack distribution tier (boto3 S3-compat + DVD-R safe-fit)

Ship arborist corpus state to new peers and DVD-R archival via
point-in-time tar.zst packs. One artifact serves both channels —
bucket+CDN delivery and physical-media archival.

Bucket holds packs only. Pack key = hash_leaf(manifest_bytes), so same
chunk set on two writers produces the same pack_hash and upload is
idempotent. Each pack pins the corpus snapshot_root it covers in audit
+ result body — packs are delayed snapshots, not live mirrors;
falsifications between repacks produce new pack_hashes.

stream_packs runs streaming zstd over tarfile, peeking compressed-buffer
size after each chunk via FLUSH_BLOCK (preserves dictionary). Default
cap 4_400_000_000 — 4.4 GB DVD-R safe-fit, ~6.5% buffer below the
4.7 GB marketing capacity to absorb ISO9660 overhead, growisofs
lead-in/lead-out, media variance, and drive-edge refusal. Each disc
fills to ~4.4 GB recorded data, not the ~1.5 GB an uncompressed cap
produced.

One backend class (S3CompatibleBackend via boto3 + endpoint_url) covers
AWS S3, DO Spaces, R2, B2, GCS S3-interop, MinIO. Optional dep
[object-store] = boto3>=1.34; dev extras pull moto for the wire test.
Voyeur: credentials via AWS_ACCESS_KEY_ID/_SECRET_ACCESS_KEY env or
~/.aws/credentials, never printed; only endpoint URL + bucket name
surface in logs.

CLI: arborist cold {pack,unpack,stats}. Makefile: cold-pack,
cold-pack-dvd (local-dir output for growisofs), cold-unpack, cold-stats.

Sizing for current shards (14.1M chunks, ~17 GB compressed): ~4 packs
at the default cap, ~\$0.34/mo DO Spaces storage, ~\$0.0001/fresh-peer
hydrate.

Always-on raw-UTF-8 leaf store (per ticket "Hard invariants") deferred
— packs-only for now, backfill later.

2557 passed, 28 skipped, 1 xfailed.

tests/test_cold_object.py

@@ -0,0 +1,325 @@
+"""Cold-pack distribution tier (#000061) — pack build, push, pull.
+
+Uses the in-process `MemoryBackend` so the default test suite runs without
+boto3, moto, or a live S3. The on-the-wire boto3+moto test lives in
+`tests/test_cold_object_boto3.py` (gated on those extras).
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Iterator
+
+import pytest
+
+from arborist.cold_object import (
+    PACK_PREFIX,
+    MemoryBackend,
+    build_pack,
+    open_pack,
+    pack_key,
+    parse_manifest,
+)
+from arborist.compress import unpack_chunk
+from arborist.document import Document
+from arborist.evict import (
+    pull_pack,
+    push_pack,
+)
+from arborist.ingest import ingest_source
+from arborist.merkle import hash_leaf
+from arborist.source import Source
+from arborist.store import connect
+
+
+class FakeSource(Source):
+    source_type = "html"
+
+    def __init__(self, docs: list[Document]):
+        self.docs = docs
+
+    def iter_documents(self) -> Iterator[Document]:
+        yield from self.docs
+
+
+def _doc(uri: str, content: str) -> Document:
+    return Document(uri=uri, content=content, source_type="html", title=uri)
+
+
+LONG = (
+    "The eight forms of capital include living, social, and intellectual. " * 30
+    + "\n\n"
+    + "Merkle providence proves answer derives from a specific source. " * 30
+)
+
+
+# ---------------------------------------------------------------------------
+# Pure-helper tests (no DB, no backend) — the content-addressing invariants
+# ---------------------------------------------------------------------------
+
+
+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.
+    assert pack_key(h) != pack_key(h, manifest=True)
+
+
+def test_build_pack_is_content_addressed():
+    # Same input → same pack_hash, regardless of ordering.
+    body_a = b"hello world " * 100
+    body_b = b"goodbye world " * 100
+    leaf_a = hash_leaf(body_a).hex()
+    leaf_b = hash_leaf(body_b).hex()
+
+    pack1 = build_pack([(leaf_a, body_a), (leaf_b, body_b)])
+    pack2 = build_pack([(leaf_b, body_b), (leaf_a, body_a)])  # reversed
+    assert pack1.pack_hash == pack2.pack_hash
+    assert {e.leaf_hash for e in pack1.entries} == {leaf_a, leaf_b}
+
+
+def test_build_pack_rejects_empty():
+    with pytest.raises(ValueError):
+        build_pack([])
+
+
+def test_open_pack_verifies_each_chunk_hash():
+    body_a = b"chunk one body padded out " * 20
+    body_b = b"chunk two body padded out " * 20
+    leaf_a = hash_leaf(body_a).hex()
+    leaf_b = hash_leaf(body_b).hex()
+    pack = build_pack([(leaf_a, body_a), (leaf_b, body_b)])
+
+    recovered = dict(open_pack(pack.body_bytes))
+    assert recovered[leaf_a] == body_a
+    assert recovered[leaf_b] == body_b
+
+
+def test_parse_manifest_round_trip():
+    body = b"abc" * 100
+    leaf = hash_leaf(body).hex()
+    pack = build_pack([(leaf, body)])
+    entries = parse_manifest(pack.manifest_bytes)
+    assert len(entries) == 1
+    assert entries[0].leaf_hash == leaf
+    assert entries[0].size == len(body)
+
+
+# ---------------------------------------------------------------------------
+# Backend round-trip — MemoryBackend, no network
+# ---------------------------------------------------------------------------
+
+
+def test_memory_backend_put_get_head_list():
+    """The backend speaks raw bytes; pack-level semantics live above it."""
+    b = MemoryBackend()
+    key = "packs/abc123.tar.zst"
+    payload = b"fake pack body " * 10
+    assert not b.head(key)
+    b.put(key, payload, content_type="application/zstd")
+    assert b.head(key)
+    assert b.get(key) == payload
+    assert list(b.list_keys("packs/")) == [key]
+
+
+def test_memory_backend_identity_carries_no_credentials():
+    b = MemoryBackend(endpoint_url="memory://nyc3", bucket="arborist-test")
+    body = b.identity.to_audit_body()
+    # Operation Voyeur: no credential keys should ever appear.
+    text = json.dumps(body)
+    for k in ("access", "secret", "key", "token", "password"):
+        assert k not in text.lower() or k == "key" and "access_key" not in text.lower()
+
+
+# ---------------------------------------------------------------------------
+# Pack push/pull — the only path now (no individual blobs)
+# ---------------------------------------------------------------------------
+
+
+def test_push_pack_then_pull_pack_restores_chunks(tmp_path):
+    db = tmp_path / "pack.db"
+    conn = connect(db)
+    backend = MemoryBackend()
+    try:
+        ingest_source(conn, FakeSource([_doc("html://p", LONG)]))
+        root = conn.execute(
+            "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))
+
+        # 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()
+
+        pull_result = pull_pack(conn, backend, pack_hash)
+        assert pull_result["status"] == "pulled"
+        assert pull_result["chunks_restored"] == first["chunk_count"]
+
+        cold = conn.execute(
+            "SELECT COUNT(*) FROM chunks WHERE tier='cold'"
+        ).fetchone()[0]
+        assert cold == 0
+    finally:
+        conn.close()
+
+
+def test_push_pack_idempotent(tmp_path):
+    """Same chunk set → same pack_hash(es) → bucket overwrite is a no-op."""
+    db = tmp_path / "pack-id.db"
+    conn = connect(db)
+    backend = MemoryBackend()
+    try:
+        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"]
+        ]
+    finally:
+        conn.close()
+
+
+def test_push_pack_nothing_to_pack(tmp_path):
+    db = tmp_path / "empty.db"
+    conn = connect(db)
+    backend = MemoryBackend()
+    try:
+        result = push_pack(conn, backend)
+        assert result["status"] == "nothing_to_pack"
+        assert result["packs"] == []
+    finally:
+        conn.close()
+
+
+def test_push_pack_splits_to_fit_dvdr(tmp_path):
+    """A small per-pack COMPRESSED-bytes cap forces multi-pack splitting;
+    every pack's compressed body stays ≤ cap + tar-trailer slack.
+
+    The cap is on compressed bytes (`stream_packs` peeks the compressed
+    buffer after each chunk via FLUSH_BLOCK). Each pack's compressed_bytes
+    should be close to the cap — fill the disc, don't leave 50% empty.
+    """
+    db = tmp_path / "split.db"
+    conn = connect(db)
+    backend = MemoryBackend()
+    try:
+        # Distinct content per doc so zstd can't dedupe across them — the
+        # whole point of testing the compressed cap is that compressed
+        # output scales with corpus, not with one-pattern repetition.
+        import random
+        rng = random.Random(42)
+        words = ["alpha", "bravo", "charlie", "delta", "echo", "foxtrot",
+                 "golf", "hotel", "india", "juliet", "kilo", "lima"]
+        docs = []
+        for i in range(8):
+            phrases = [" ".join(rng.choices(words, k=10)) for _ in range(80)]
+            docs.append(_doc(f"html://d{i}", " ".join(phrases) + ". "))
+        ingest_source(conn, FakeSource(docs))
+
+        # Small compressed cap (4 KB) — should produce multiple packs.
+        CAP = 4_096
+        result = push_pack(conn, backend, max_pack_bytes=CAP)
+        assert result["status"] == "pushed"
+        assert result["pack_count"] > 1, (
+            f"expected multiple packs at {CAP}-byte compressed cap, "
+            f"got {result['pack_count']}"
+        )
+
+        # 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.
+        OVERSHOOT_SLACK = 4_096
+        for p in result["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}"
+            )
+            assert p["chunk_count"] >= 1
+
+        # Round-trip: NULL local content, pull every pack back.
+        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"]
+        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."""
+    db = tmp_path / "burn.db"
+    conn = connect(db)
+    backend = MemoryBackend()
+    burn_dir = tmp_path / "discs"
+    try:
+        ingest_source(conn, FakeSource([_doc("html://burn", LONG)]))
+        result = push_pack(conn, backend, local_dir=str(burn_dir))
+        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"]
+    finally:
+        conn.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"
+    conn = connect(db)
+    backend = MemoryBackend()
+    burn_dir = tmp_path / "iso"
+    try:
+        ingest_source(conn, FakeSource([_doc("html://x", LONG)]))
+        result = push_pack(
+            conn, backend,
+            local_dir=str(burn_dir),
+            push_to_bucket=False,
+        )
+        assert result["status"] == "written"
+        # No keys in bucket.
+        assert list(backend.list_keys()) == []
+        # Files on disk.
+        assert any(burn_dir.glob("arborist-pack-*.tar.zst"))
+    finally:
+        conn.close()
+
+
+def test_default_max_pack_bytes_is_dvdr_safe_fit():
+    """Default cap is 4.4 GB — DVD-R safe-fit, with ~6.5% buffer below the
+    4.7 GB media spec to absorb ISO9660 overhead, media manufacturing
+    variance, and drive-edge refusal. Don't drift this without a deliberate
+    decision; downstream burning workflows assume it."""
+    from arborist.evict import DEFAULT_MAX_PACK_BYTES
+    assert DEFAULT_MAX_PACK_BYTES == 4_400_000_000
+    # Must sit well below the 4.7 GB marketing capacity AND below the
+    # 4,706,074,624 ECMA-267 physical capacity, with room for overhead.
+    assert DEFAULT_MAX_PACK_BYTES < 4_700_000_000
+    assert (4_700_000_000 - DEFAULT_MAX_PACK_BYTES) >= 250_000_000  # >=250 MB headroom

tests/test_cold_object_boto3.py

@@ -0,0 +1,69 @@
+"""On-the-wire boto3 backend test (#000061) — moto-mocked S3.
+
+Verifies that S3CompatibleBackend correctly issues PUT / GET / HEAD /
+LIST against an S3 API for pack-shaped payloads. The rest of the cold-
+pack surface lives in tests/test_cold_object.py and runs against the
+in-process MemoryBackend.
+
+Gated on boto3 + moto. The default suite skips this file.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+boto3 = pytest.importorskip("boto3")
+moto = pytest.importorskip("moto")
+
+from arborist.cold_object import PACK_PREFIX, S3CompatibleBackend, pack_key
+from arborist.merkle import hash_leaf
+
+
+@pytest.fixture
+def moto_s3():
+    """Spin up an in-memory S3 endpoint via moto, yield a fresh bucket."""
+    from moto import mock_aws
+
+    with mock_aws():
+        client = boto3.client("s3", region_name="us-east-1")
+        client.create_bucket(Bucket="arborist-test")
+        yield {
+            "bucket": "arborist-test",
+            "region": "us-east-1",
+        }
+
+
+def test_s3_backend_put_get_head_list(moto_s3):
+    # moto mocks the AWS endpoint when endpoint_url is None or AWS-shaped.
+    # Pointing at the default AWS URL keeps boto3 happy inside the mock.
+    backend = S3CompatibleBackend(
+        endpoint_url="https://s3.amazonaws.com",
+        bucket=moto_s3["bucket"],
+        region=moto_s3["region"],
+    )
+
+    pack_body = b"fake pack body padded out " * 200
+    pack_hash = hash_leaf(pack_body).hex()
+    key = pack_key(pack_hash)
+
+    assert not backend.head(key)
+    backend.put_pack(pack_hash, pack_body)
+    assert backend.head(key)
+    assert backend.get_pack(pack_hash) == pack_body
+
+    keys = list(backend.list_keys(PACK_PREFIX))
+    assert keys == [key]
+
+
+def test_s3_backend_identity_carries_endpoint_and_bucket_only(moto_s3):
+    backend = S3CompatibleBackend(
+        endpoint_url="https://nyc3.digitaloceanspaces.com",
+        bucket="arborist-do-test",
+        region="us-east-1",
+    )
+    identity = backend.identity
+    body = identity.to_audit_body()
+    assert body["endpoint_url"] == "https://nyc3.digitaloceanspaces.com"
+    assert body["bucket"] == "arborist-do-test"
+    assert "access_key" not in body
+    assert "secret" not in body