arborist/tests/test_cold_object.py

"""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 (
    hydrate_from_metadata_pack,
    pull_chunk_pack,
    pull_metadata_pack,
    push_pack as _push_pack,
)


def push_pack(*args, **kwargs):
    """Test wrapper for evict.push_pack that opts into 'unknown'
    license_class. FakeSource uses source_type='html' which classifies
    as 'unknown', and the Gap-2 license gate refuses unknown-class
    pushes to the bucket by default. Tests here exercise pack
    semantics, not the license gate — that gets its own test below.
    """
    kwargs.setdefault("allow_license_class", "unknown")
    return _push_pack(*args, **kwargs)
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
    # 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():
    # 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)])
    parsed = parse_manifest(pack.manifest_bytes)
    # 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
    assert parsed.chunks[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_object_size_returns_content_length_or_none():
    b = MemoryBackend()
    key = "packs/abc.tar.zst"
    assert b.object_size(key) is None  # missing
    b.put(key, b"x" * 1234)
    assert b.object_size(key) == 1234
    assert b.object_size("packs/nonexistent") is None


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_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()
    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"]

        push_result = push_pack(conn, backend, document_root=root)
        assert push_result["status"] == "pushed"
        # 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

        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"))

        # 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' 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 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()
    try:
        ingest_source(conn, FakeSource([_doc("html://p", LONG)]))
        a = push_pack(conn, backend)
        b = push_pack(conn, backend)
        assert a["metadata_pack_hash"] == b["metadata_pack_hash"]
        assert a["chunk_pack_hashes"] == b["chunk_pack_hashes"]
    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. 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
        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"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, hydrate-full from metadata pack.
        conn.execute("UPDATE chunks SET content = NULL, tier='cold'")
        conn.execute("DELETE FROM chunks_fts")
        conn.commit()
        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.
    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()
    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()
        # 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()


def test_edges_fan_in_batches_huge_destinations(tmp_path):
    """Regression test for the memory-bounded edges dump path.

    A destination with > _FAN_IN_BATCH inbound links should be split
    into multiple JSON rows on dump, and restored to the live schema
    with no missing or duplicated edges. Without this batching the
    groupby accumulator held the whole src_roots list in Python heap
    (~5 GB per worker observed on the v2 corpus run).
    """
    import sqlite3
    from arborist.cold_pack_metadata import (
        _dump_edges_fan_in,
        _restore_edges_fan_out,
        _FAN_IN_BATCH,
    )
    from arborist.store import connect as store_connect

    # Build a minimal schema-bearing DB and stuff edges directly.
    src_db = tmp_path / "edges.db"
    conn = store_connect(src_db)
    try:
        # Insert N_LINKS edges all pointing at the same dst_uri so the
        # groupby for this destination spans multiple batches.
        N_LINKS = _FAN_IN_BATCH + 137  # one full batch + a partial
        dst_uri = "https://example.com/popular"
        edge_type = "wikilink"
        anchor = ""
        dst_root = ""
        with conn:
            conn.executemany(
                "INSERT OR IGNORE INTO edges "
                "(src_root, dst_root, dst_uri, edge_type, anchor) "
                "VALUES (?, ?, ?, ?, ?)",
                [
                    (f"src{i:08d}" + "0" * 24, dst_root, dst_uri, edge_type, anchor)
                    for i in range(N_LINKS)
                ],
            )

        # Dump
        dump_path = tmp_path / "edges.jsonl"
        n_rows_written = _dump_edges_fan_in(conn, dump_path)
        # One destination split into ceil(N_LINKS / FAN_IN_BATCH) rows.
        expected_rows = (N_LINKS + _FAN_IN_BATCH - 1) // _FAN_IN_BATCH
        assert n_rows_written == expected_rows, (
            f"expected {expected_rows} fan-in rows for {N_LINKS} edges, "
            f"got {n_rows_written}"
        )

        # Wipe edges and restore. INSERT OR IGNORE handles any duplicate
        # PK rows from chunk boundaries cleanly.
        with conn:
            conn.execute("DELETE FROM edges")
        n_restored = _restore_edges_fan_out(conn, dump_path)
        conn.commit()
        assert n_restored == N_LINKS, (
            f"expected {N_LINKS} edges restored, got {n_restored}"
        )

        # Verify the live table has exactly N_LINKS rows for that destination.
        final_count = conn.execute(
            "SELECT COUNT(*) FROM edges WHERE dst_uri = ?",
            (dst_uri,),
        ).fetchone()[0]
        assert final_count == N_LINKS
    finally:
        conn.close()


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)
        metadata_hash = result["metadata_pack_hash"]
        chunk_hashes = result["chunk_pack_hashes"]
        assert metadata_hash is not None
        assert len(chunk_hashes) >= 1

        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
        for t in meta.tables:
            assert len(t.content_hash) == 64
            assert t.size > 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()


def test_pull_pack_rejects_tampered_table(tmp_path):
    """If the manifest says tables/X.jsonl should hash to H but the tar
    member has different bytes, pull_pack refuses to restore.

    Tampering test: build a real v2 pack, then construct a corrupted
    copy that swaps one table's bytes but keeps the original manifest.
    pull_pack must raise on hash mismatch.
    """
    import io as _io
    import tarfile as _tarfile
    import zstandard as _zstd

    db = tmp_path / "src.db"
    conn = connect(db)
    backend = MemoryBackend()
    try:
        ingest_source(conn, FakeSource([_doc("html://t", LONG)]))
        result = push_pack(conn, backend)
        metadata_hash = result["metadata_pack_hash"]
    finally:
        conn.close()

    # 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)
    with cctx.stream_writer(out_buf, closefd=False) as writer:
        with _tarfile.open(fileobj=writer, mode="w|") as out_tar:
            with _tarfile.open(fileobj=_io.BytesIO(raw_tar), mode="r") as in_tar:
                for m in in_tar:
                    f = in_tar.extractfile(m)
                    if f is None:
                        continue
                    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)
                    out_tar.addfile(info, _io.BytesIO(data))
    backend._tamper(
        f"packs/{metadata_hash}.metadata.tar.zst", out_buf.getvalue()
    )

    # 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_metadata_pack(conn_rx, backend, metadata_hash)
    finally:
        conn_rx.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"
        metadata_hash = result["metadata_pack_hash"]
        assert metadata_hash is not None
    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

        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
        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"
    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_push_pack_is_deterministic_across_runs(tmp_path):
    """Same chunk set on two runs against the same DB produces the same
    pack_hashes — `ORDER BY leaf_hash` makes the chunk-to-pack assignment
    a function of (chunk set, cap) and nothing else. This is the
    prerequisite for parallel pack workers and for two replicas at the
    same snapshot to produce byte-identical bucket state."""
    db = tmp_path / "det.db"
    conn = connect(db)
    try:
        # Use distinct content so multiple chunks survive dedupe and
        # we have enough material to potentially split.
        import random
        rng = random.Random(7)
        words = ["alpha", "bravo", "charlie", "delta", "echo", "foxtrot"]
        docs = []
        for i in range(6):
            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))

        backend_a = MemoryBackend()
        backend_b = MemoryBackend()
        result_a = push_pack(conn, backend_a, max_pack_bytes=4_000)
        result_b = push_pack(conn, backend_b, max_pack_bytes=4_000)

        hashes_a = [p["pack_hash"] for p in result_a["packs"]]
        hashes_b = [p["pack_hash"] for p in result_b["packs"]]
        assert hashes_a == hashes_b, (
            "two runs at the same snapshot produced different pack_hashes "
            f"(a={hashes_a} vs b={hashes_b}) — ORDER BY drift?"
        )
    finally:
        conn.close()


def test_gap2_license_gate_refuses_unknown_class_to_public_bucket(tmp_path):
    """Gap 2: push_pack refuses to ship a shard whose strictest source
    license_class is more restrictive than the operator's
    allow_license_class. Default allow=public_redistributable; html-source
    shards are 'unknown' by default → push must raise."""
    db = tmp_path / "license.db"
    conn = connect(db)
    backend = MemoryBackend()
    try:
        ingest_source(conn, FakeSource([_doc("html://x", LONG)]))
        # Default allow_license_class='public_redistributable' refuses.
        with pytest.raises(ValueError, match="license_class='unknown'"):
            _push_pack(conn, backend)
        # Explicit opt-in: allow_license_class='unknown' passes.
        result = _push_pack(conn, backend, allow_license_class="unknown")
        assert result["status"] == "pushed"
    finally:
        conn.close()


def test_gap2_license_class_in_metadata_manifest(tmp_path):
    """Gap 2: the metadata pack manifest carries _license_class so
    consumers + auditors can see the producer's classification."""
    db = tmp_path / "lc-manifest.db"
    conn = connect(db)
    backend = MemoryBackend()
    try:
        ingest_source(conn, FakeSource([_doc("html://x", LONG)]))
        result = push_pack(conn, backend)  # uses test wrapper, allow='unknown'
        meta_manifest = backend.get_pack_manifest(
            result["metadata_pack_hash"], kind="metadata"
        )
        parsed = parse_manifest(meta_manifest)
        assert parsed.license_class == "unknown"
    finally:
        conn.close()


def test_gap1_latest_pointer_resolves_metadata_pack_per_snapshot(tmp_path):
    """Gap 1: after push, manifest/latest.json points to the current
    metadata pack for each snapshot_root. Fresh peer can resolve the
    metadata pack hash without enumerating every object."""
    db = tmp_path / "latest.db"
    conn = connect(db)
    backend = MemoryBackend()
    try:
        ingest_source(conn, FakeSource([_doc("html://x", LONG)]))
        result = push_pack(conn, backend)
        snapshot_root = result["snapshot_root"]
        metadata_hash = result["metadata_pack_hash"]

        # Backend has the latest pointer.
        pointer = backend.get_latest_pointer()
        assert snapshot_root in pointer
        entry = pointer[snapshot_root]
        assert entry["metadata_pack_hash"] == metadata_hash
        assert entry["snapshot_doc_count"] == result["snapshot_doc_count"]
        assert entry["license_class"] == "unknown"
    finally:
        conn.close()


def test_gap3_cold_pending_clears_on_successful_upload(tmp_path):
    """Gap 3: cold_pending rows track in-flight uploads. On success,
    they're cleared. After a clean push, cold_pending is empty."""
    db = tmp_path / "pending.db"
    conn = connect(db)
    backend = MemoryBackend()
    try:
        ingest_source(conn, FakeSource([_doc("html://x", LONG)]))
        push_pack(conn, backend)
        # Successful push leaves no orphan pending rows.
        pending = conn.execute("SELECT COUNT(*) FROM cold_pending").fetchone()[0]
        assert pending == 0
    finally:
        conn.close()


def test_gap3_cold_pending_records_inflight_upload(tmp_path):
    """Gap 3: when the backend put_pack_file fails mid-flight, the
    cold_pending row stays so a recovery tool can find the orphan
    tempfile and clean it up."""
    db = tmp_path / "pending-fail.db"
    conn = connect(db)
    try:
        ingest_source(conn, FakeSource([_doc("html://x", LONG)]))

        class FailingBackend(MemoryBackend):
            def put_file(self, key, path, *, content_type="application/octet-stream"):
                raise RuntimeError("simulated upload failure")

        backend = FailingBackend()
        with pytest.raises(RuntimeError, match="simulated upload"):
            push_pack(conn, backend)
        # The row should exist tracking what was almost uploaded.
        pending_rows = conn.execute(
            "SELECT kind, pack_hash, object_key FROM cold_pending"
        ).fetchall()
        assert len(pending_rows) >= 1
        # The recorded row identifies enough to either resume or clean up.
        assert pending_rows[0]["pack_hash"]
        assert pending_rows[0]["object_key"].startswith("packs/")
    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