arborist/arborist/cold_object.py

"""Object-store backend for the cold-pack distribution tier (#000061).

The bucket stores **tar.zst packs** — corpus tarballs that bring up a new
peer from empty. Each pack ≤ 4.4 GB (DVD-R safe-fit) so the same artifact
also burns to physical media. Pack key = `hash_leaf(manifest_bytes)`;
content-addressed at the pack layer, so two writers producing the same
chunk set produce the same key (idempotent upload).

Why packs and not individual blobs:

- For new-peer hydration the right granularity is one HTTPS GET per disc-
  worth of corpus, not one per chunk. 14M individual GETs vs 12 pack GETs
  is a million-x cost difference on request fees alone.
- Packs are the backup unit. The same .tar.zst that sits in the bucket
  burns directly to DVD — bit-for-bit identical, content-addressed
  filename. No re-encoding for physical media.
- Storage stays compressed. Bucket bytes ≈ on-disk compressed bytes,
  not the ~2.2× larger raw UTF-8.

Multi-cloud unification. One backend class works for AWS S3, DigitalOcean
Spaces, Google Cloud Storage (S3 interop), Cloudflare R2, Backblaze B2,
and MinIO — boto3 with a per-provider `endpoint_url` covers all of them.

CDN public-read. Bucket ACL = public-read lets any client fetch packs
from the CDN edge without credentials. Verification is local: every
chunk inside the pack has its `leaf_hash` in its tar member name, and
`open_pack` refuses to restore on hash mismatch — the bucket can be
hostile, content addressing binds the bytes.

Optional dependency. Installing arborist core does NOT pull boto3 — the
backend lives behind the `[object-store]` extras. Tests skip via
`pytest.importorskip("boto3")` when absent. Same gating pattern as
`[html]`, `[nli]`, `[mt]`.

Operation Voyeur (CLAUDE.md): credentials enter via standard boto3
discovery (env vars, ~/.aws/credentials, IAM role). They never appear
in subprocess argv, never print to stdout, never enter conversation
logs. Backend identity is endpoint URL + bucket name only.
"""

from __future__ import annotations

import abc
import io
import json
import os
import tarfile
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator

import zstandard

from arborist.merkle import hash_leaf


# Bucket layout — every reader of this module must agree on these constants.
PACK_PREFIX = "packs/"
MANIFEST_PREFIX = "manifest/"
MANIFEST_LATEST_KEY = MANIFEST_PREFIX + "latest.ndjson"


def pack_key(pack_hash: str, *, manifest: bool = False) -> str:
    """Bucket key for a pack body (default) or its manifest sidecar."""
    if manifest:
        return f"{PACK_PREFIX}{pack_hash}.manifest.ndjson"
    return f"{PACK_PREFIX}{pack_hash}.tar.zst"


@dataclass(frozen=True)
class BackendIdentity:
    """Auditable identity for a backend instance.

    Goes into audit_events body — endpoint + bucket are public, credentials
    NEVER appear here (Operation Voyeur).
    """

    endpoint_url: str
    bucket: str
    region: str | None

    def to_audit_body(self) -> dict:
        out = {"endpoint_url": self.endpoint_url, "bucket": self.bucket}
        if self.region:
            out["region"] = self.region
        return out


class ObjectStoreBackend(abc.ABC):
    """ABC for cold-object-store backends.

    All keys here are bucket keys (full path within the bucket). All bodies
    are raw bytes — callers handle text encoding before reaching the backend.

    Hash verification is the *caller's* responsibility. The backend does not
    decide whether a byte sequence is valid; it only moves bytes. This keeps
    the integrity check at exactly one layer (eviction / rehydration code
    in `arborist/evict.py`) where the audit chain lives.
    """

    @property
    @abc.abstractmethod
    def identity(self) -> BackendIdentity:
        ...

    @abc.abstractmethod
    def put(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> None:
        ...

    @abc.abstractmethod
    def put_file(self, key: str, path: Path, *, content_type: str = "application/octet-stream") -> None:
        """Upload a file from disk WITHOUT loading it into memory first.

        Used for large pack bodies — `stream_packs` writes packs to temp
        files on disk so worker memory stays bounded by zstd internal
        buffers (~few MB), not by pack size. The S3 backend hands the
        file path to boto3 `upload_file` which streams parts from disk
        for multipart uploads.
        """

    @abc.abstractmethod
    def get(self, key: str) -> bytes:
        ...

    @abc.abstractmethod
    def head(self, key: str) -> bool:
        """True if the object exists; False otherwise. Never raises on missing."""

    @abc.abstractmethod
    def object_size(self, key: str) -> int | None:
        """Return Content-Length of the object, or None if it doesn't exist.

        One HEAD round-trip; no body download. Used by `cold list` and
        `cold stats` to report bucket footprint without paying egress
        for the actual bytes.
        """

    @abc.abstractmethod
    def list_keys(self, prefix: str = "") -> Iterator[str]:
        ...

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

    def put_pack_file(self, pack_hash: str, path: Path) -> 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")

    def get_pack(self, pack_hash: str) -> bytes:
        return self.get(pack_key(pack_hash))

    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 get_pack_manifest(self, pack_hash: str) -> bytes:
        return self.get(pack_key(pack_hash, manifest=True))


class S3CompatibleBackend(ObjectStoreBackend):
    """boto3-based S3-compatible backend.

    Works across AWS S3, DigitalOcean Spaces, Google Cloud Storage (S3
    interop endpoint), Cloudflare R2, Backblaze B2, MinIO — every
    provider that speaks the S3 API. The only per-provider knob is
    `endpoint_url`.

    Credentials come from boto3's standard discovery chain (env vars,
    ~/.aws/credentials, IAM role). The class never accepts credentials as
    plaintext constructor args — that's a Voyeur-protocol guardrail (so
    callers can't accidentally leak a key by stringifying the backend).
    """

    # Multipart upload tuning. boto3's `upload_fileobj` uses these to decide
    # when to switch from a single PUT to a parallel multipart upload, the
    # part size, and how many parts to upload concurrently. Values below
    # come from a 100k-chunk DO-Spaces NYC3 bench (2026-05-26): single-PUT
    # baseline was 5.5 MB/s on a 121 MB pack; multipart at these settings
    # is ~5-10x faster. Also required for packs > 5 GB (DO Spaces single-
    # PUT limit; AWS S3 has the same limit at 5 GB).
    _MULTIPART_THRESHOLD_BYTES = 8 * 1024 * 1024   # 8 MB
    _MULTIPART_CHUNKSIZE_BYTES = 8 * 1024 * 1024   # 8 MB part size
    _MULTIPART_MAX_CONCURRENCY = 10                # parallel parts in flight

    def __init__(
        self,
        *,
        endpoint_url: str,
        bucket: str,
        region: str | None = None,
    ):
        try:
            import boto3
            from boto3.s3.transfer import TransferConfig
            from botocore.client import Config
            from botocore.exceptions import ClientError
        except ImportError as e:
            raise RuntimeError(
                "S3CompatibleBackend requires boto3. "
                "Install with: pip install 'arborist[object-store]'"
            ) from e

        # Some providers (DO Spaces) want any region string; pick a default
        # so signing works. AWS callers always pass their real region.
        effective_region = region or "us-east-1"

        self._endpoint_url = endpoint_url
        self._bucket = bucket
        self._region = region
        self._client_error = ClientError
        self._client = boto3.client(
            "s3",
            endpoint_url=endpoint_url,
            region_name=effective_region,
            config=Config(
                signature_version="s3v4",
                retries={"max_attempts": 3, "mode": "standard"},
            ),
        )
        self._transfer_config = TransferConfig(
            multipart_threshold=self._MULTIPART_THRESHOLD_BYTES,
            multipart_chunksize=self._MULTIPART_CHUNKSIZE_BYTES,
            max_concurrency=self._MULTIPART_MAX_CONCURRENCY,
            use_threads=True,
        )

    @property
    def identity(self) -> BackendIdentity:
        return BackendIdentity(
            endpoint_url=self._endpoint_url,
            bucket=self._bucket,
            region=self._region,
        )

    def put(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> None:
        # upload_fileobj uses TransferConfig to switch to parallel multipart
        # upload above the 8 MB threshold. Single PUT below threshold (no
        # multipart overhead for small objects like manifests). Required
        # for packs > 5 GB (DO Spaces / S3 single-PUT limit).
        self._client.upload_fileobj(
            Fileobj=io.BytesIO(body),
            Bucket=self._bucket,
            Key=key,
            ExtraArgs={"ContentType": content_type},
            Config=self._transfer_config,
        )

    def put_file(self, key: str, path: Path, *, content_type: str = "application/octet-stream") -> None:
        # boto3 upload_file reads parts from disk on demand for multipart
        # uploads — memory stays bounded by max_concurrency × part_size
        # (10 × 8 MB = 80 MB) instead of growing with pack size. The on-
        # disk source is left untouched; caller decides delete vs keep.
        self._client.upload_file(
            Filename=str(path),
            Bucket=self._bucket,
            Key=key,
            ExtraArgs={"ContentType": content_type},
            Config=self._transfer_config,
        )

    def get(self, key: str) -> bytes:
        resp = self._client.get_object(Bucket=self._bucket, Key=key)
        return resp["Body"].read()

    def head(self, key: str) -> bool:
        try:
            self._client.head_object(Bucket=self._bucket, Key=key)
            return True
        except self._client_error as e:
            code = e.response.get("Error", {}).get("Code", "")
            if code in ("404", "NoSuchKey", "NotFound"):
                return False
            raise

    def object_size(self, key: str) -> int | None:
        try:
            resp = self._client.head_object(Bucket=self._bucket, Key=key)
            return int(resp["ContentLength"])
        except self._client_error as e:
            code = e.response.get("Error", {}).get("Code", "")
            if code in ("404", "NoSuchKey", "NotFound"):
                return None
            raise

    def list_keys(self, prefix: str = "") -> Iterator[str]:
        paginator = self._client.get_paginator("list_objects_v2")
        for page in paginator.paginate(Bucket=self._bucket, Prefix=prefix):
            for obj in page.get("Contents", []) or []:
                yield obj["Key"]


class MemoryBackend(ObjectStoreBackend):
    """In-process backend for tests and local dry-runs.

    Stores bytes in a dict. Same contract as `S3CompatibleBackend` —
    same key conventions, same content-addressing invariants. Used by
    `tests/test_cold_object.py` so the default test suite runs without
    boto3 / moto / a live S3.

    Not a substitute for production — no network, no durability, no ACL
    semantics. The on-the-wire test is `tests/test_cold_object_boto3.py`
    which uses moto when available.
    """

    def __init__(self, *, endpoint_url: str = "memory://", bucket: str = "test"):
        self._endpoint_url = endpoint_url
        self._bucket = bucket
        self._store: dict[str, bytes] = {}

    @property
    def identity(self) -> BackendIdentity:
        return BackendIdentity(
            endpoint_url=self._endpoint_url,
            bucket=self._bucket,
            region=None,
        )

    def put(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> None:
        self._store[key] = bytes(body)

    def put_file(self, key: str, path: Path, *, content_type: str = "application/octet-stream") -> None:
        with open(path, "rb") as f:
            self._store[key] = f.read()

    def get(self, key: str) -> bytes:
        if key not in self._store:
            raise KeyError(key)
        return self._store[key]

    def head(self, key: str) -> bool:
        return key in self._store

    def object_size(self, key: str) -> int | None:
        body = self._store.get(key)
        return len(body) if body is not None else None

    def list_keys(self, prefix: str = "") -> Iterator[str]:
        # Sort for stable iteration (S3 list returns lex order too).
        for key in sorted(self._store):
            if key.startswith(prefix):
                yield key

    def _tamper(self, key: str, new_body: bytes) -> None:
        """Test-only helper: overwrite a key with arbitrary bytes so the
        rehydration hash-mismatch path can be exercised."""
        self._store[key] = new_body


# ---------------------------------------------------------------------------
# Pack format. A pack is a zstd-compressed tar containing:
#   - blobs/<hash[:2]>/<hash[2:]>     one entry per chunk, body = raw UTF-8
#   - manifest.ndjson                  contents sidecar (written last in the
#                                      streaming builder; consumers iterate
#                                      and skip non-blob entries, so order
#                                      doesn't matter)
# Pack is keyed by `hash_leaf(manifest_bytes)` so identical chunk sets from
# two writers collide idempotently.
#
# build_pack:   in-memory builder for small/test cases (whole tar in RAM).
# stream_packs: streaming builder that caps each pack's COMPRESSED size at
#               max_compressed_bytes, so packs fill physical media (4.4 GB
#               DVD-R safe-fit) instead of leaving ~50% of every disc empty.
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class PackEntry:
    leaf_hash: str
    size: int


@dataclass(frozen=True)
class Pack:
    """In-memory pack — used by `build_pack` for small/test cases.

    `body_bytes` is the whole zstd-compressed tar held in RAM. Fine for
    unit tests and for the few-MB packs built from a single document.
    For full-corpus 4.4 GB packs use `stream_packs` + `FilePack` instead.
    """
    pack_hash: str
    manifest_bytes: bytes
    body_bytes: bytes
    entries: tuple[PackEntry, ...]


@dataclass(frozen=True)
class FilePack:
    """File-backed pack — `stream_packs` produces these.

    `body_path` is a tempfile on disk; the caller owns its lifecycle
    (`push_pack` uploads it via `put_pack_file`, then either moves to
    `local_dir` or deletes). Worker memory during build stays bounded
    by zstd internal buffers (~few MB) regardless of pack size.
    """
    pack_hash: str
    manifest_bytes: bytes
    body_path: Path
    body_size: int
    entries: tuple[PackEntry, ...]


def build_pack(chunks: list[tuple[str, bytes]]) -> Pack:
    """Bundle a list of (leaf_hash, raw_utf8_body) pairs into one pack.

    Every input must already be hash-verified by the caller —
    `sha256(body) == leaf_hash`. The builder does NOT re-verify; that
    contract belongs to the eviction code so verification stays at one
    layer.

    Returns a `Pack` with `pack_hash = hash_leaf(manifest_bytes).hex()` —
    content-addressed so two writers producing the same chunk set produce
    the same pack key (idempotent overwrite on push). The manifest is
    canonicalized via leaf-hash sort so input order doesn't affect the key.
    """
    if not chunks:
        raise ValueError("build_pack requires at least one chunk")

    # Canonicalize: dedupe by leaf_hash, sort, so input order is irrelevant.
    by_hash: dict[str, bytes] = {}
    for leaf_hash, body in chunks:
        if len(leaf_hash) != 64:
            raise ValueError(f"bad leaf_hash length: {leaf_hash}")
        by_hash[leaf_hash] = body

    manifest_lines = []
    entries = []
    for leaf_hash in sorted(by_hash):
        body = by_hash[leaf_hash]
        entries.append(PackEntry(leaf_hash=leaf_hash, size=len(body)))
        manifest_lines.append(
            json.dumps({"leaf_hash": leaf_hash, "size": len(body)}, sort_keys=True)
        )
    manifest_bytes = ("\n".join(manifest_lines) + "\n").encode("utf-8")
    pack_hash = hash_leaf(manifest_bytes).hex()

    # Build the tar in memory. Sort entries by leaf_hash so the byte
    # stream is deterministic for a given input set.
    tar_buf = io.BytesIO()
    with tarfile.open(fileobj=tar_buf, mode="w") as tar:
        info = tarfile.TarInfo(name="manifest.ndjson")
        info.size = len(manifest_bytes)
        tar.addfile(info, io.BytesIO(manifest_bytes))
        for leaf_hash in sorted(by_hash):
            body = by_hash[leaf_hash]
            info = tarfile.TarInfo(name=f"blobs/{leaf_hash[:2]}/{leaf_hash[2:]}")
            info.size = len(body)
            tar.addfile(info, io.BytesIO(body))
    raw_tar = tar_buf.getvalue()

    # zstd-compress the tarball. Level 3 matches `compress.py` discipline.
    body_bytes = zstandard.ZstdCompressor(level=3).compress(raw_tar)

    return Pack(
        pack_hash=pack_hash,
        manifest_bytes=manifest_bytes,
        body_bytes=body_bytes,
        entries=tuple(entries),
    )


def stream_packs(
    chunks: Iterator[tuple[str, bytes]],
    *,
    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.

    Memory bound: tens of MB per worker (zstd internal buffers + tar
    header staging) regardless of pack size. The full pack lives on
    disk — not in RAM — so 4-way parallel × 4.4 GB packs takes ~tens of
    MB of RAM total, not 17 GB. boto3 `upload_file` reads parts from
    disk during multipart upload so the upload path is also memory-
    bounded.

    Disk requirement: `work_dir` (default `tempfile.gettempdir()`) needs
    enough free space to hold one in-progress pack per parallel worker.
    For 4-way at 4.4 GB cap = ~18 GB. The temp file is the caller's to
    manage — `push_pack` moves it to `local_dir` or deletes after upload.

    How: a `ZstdCompressor.stream_writer` writes directly into an open
    file handle; a streaming tar writes into the compressor. After each
    chunk we call `flush(FLUSH_BLOCK)` + `os.fsync`-free `file.flush()`
    so `tell()` reflects on-disk size. When `tell()` ≥ cap, we finalize:
    write the manifest as the LAST tar member, close everything, yield
    the `FilePack`, start a new pack.

    Manifest goes last because the chunk set isn't known until the cap
    is hit — `open_pack` iterates by member name and ignores order so
    this is transparent to consumers.

    Caveat: tar trailer (~1 KB) + zstd frame footer (~10 B) are emitted
    after our last size check, so on-disk size can overshoot the cap by
    ~1 KB. Trivial for a 4.4 GB cap.

    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
        # FD to zstd's stream_writer which wraps it; the tar wraps that.
        # closefd=False on the zstd writer so we can flush + tell() the
        # underlying file ourselves; we'll close everything in finalize.
        fd, path_str = tempfile.mkstemp(
            dir=work_path,
            prefix="arborist-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|")
        return {
            "path": Path(path_str),
            "file_obj": file_obj,
            "writer": writer,
            "tar": tar,
            "entries_by_hash": {},
        }

    def _finalize_pack(state: dict) -> FilePack:
        entries_sorted = sorted(state["entries_by_hash"].keys())
        manifest_lines = []
        entries = []
        for leaf_hash in entries_sorted:
            size = state["entries_by_hash"][leaf_hash]
            entries.append(PackEntry(leaf_hash=leaf_hash, size=size))
            manifest_lines.append(
                json.dumps({"leaf_hash": leaf_hash, "size": size}, sort_keys=True)
            )
        manifest_bytes = ("\n".join(manifest_lines) + "\n").encode("utf-8")
        info = tarfile.TarInfo(name="manifest.ndjson")
        info.size = len(manifest_bytes)
        state["tar"].addfile(info, io.BytesIO(manifest_bytes))
        state["tar"].close()
        state["writer"].close()
        state["file_obj"].close()
        pack_hash = hash_leaf(manifest_bytes).hex()
        body_size = state["path"].stat().st_size
        return FilePack(
            pack_hash=pack_hash,
            manifest_bytes=manifest_bytes,
            body_path=state["path"],
            body_size=body_size,
            entries=tuple(entries),
        )

    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.
        for member_name, source_path in pending_extras:
            info = tarfile.TarInfo(name=member_name)
            info.size = source_path.stat().st_size
            # mtime, uid, gid, mode default to 0 / "" — deterministic.
            with open(source_path, "rb") as f:
                state["tar"].addfile(info, f)
        # 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}")
            # Dedupe within a pack — same hash → same body by construction.
            if leaf_hash in state["entries_by_hash"]:
                continue
            info = tarfile.TarInfo(name=f"blobs/{leaf_hash[:2]}/{leaf_hash[2:]}")
            info.size = len(body)
            state["tar"].addfile(info, io.BytesIO(body))
            state["entries_by_hash"][leaf_hash] = len(body)
            # FLUSH_BLOCK pushes pending bytes out of zstd's internal buffer
            # to the file handle, preserving dictionary state. Then flush
            # the Python file buffer so the OS sees them and tell() returns
            # the real on-disk size.
            state["writer"].flush(zstandard.FLUSH_BLOCK)
            state["file_obj"].flush()
            if state["file_obj"].tell() >= max_compressed_bytes:
                yield _finalize_pack(state)
                state = _start_pack()

        if state["entries_by_hash"]:
            yield _finalize_pack(state)
        else:
            # No chunks went into the last state — clean up the empty file.
            state["tar"].close()
            state["writer"].close()
            state["file_obj"].close()
            try:
                state["path"].unlink()
            except FileNotFoundError:
                pass
    except BaseException:
        # On any exception (cancel, OOM, disk full), clean up the in-flight
        # temp file so we don't litter /tmp.
        try:
            state["tar"].close()
        except Exception:
            pass
        try:
            state["writer"].close()
        except Exception:
            pass
        try:
            state["file_obj"].close()
        except Exception:
            pass
        try:
            state["path"].unlink()
        except FileNotFoundError:
            pass
        raise


def open_pack(body_bytes: bytes) -> Iterator[tuple[str, bytes]]:
    """Stream (leaf_hash, body) pairs from a pack body.

    Verifies each chunk's hash against the leaf_hash encoded in its
    tar member name. Raises ValueError on mismatch — bucket corruption
    OR tampering. Callers should additionally cross-check against the
    pack's manifest, which carries declared sizes.

    Uses streaming decompression because packs built by `stream_packs`
    don't carry a content-size frame header (the streaming compressor
    doesn't know the total size up front), and the one-shot
    `ZstdDecompressor.decompress(body)` requires that header. Streaming
    works on either header presence.
    """
    dctx = zstandard.ZstdDecompressor()
    raw_tar = dctx.stream_reader(io.BytesIO(body_bytes)).read()
    with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r") as tar:
        for member in tar:
            if member.name == "manifest.ndjson":
                continue
            if not member.name.startswith("blobs/"):
                continue
            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
            body = f.read()
            actual_hash = hash_leaf(body).hex()
            if actual_hash != expected_hash:
                raise ValueError(
                    f"pack chunk hash mismatch: declared {expected_hash}, "
                    f"actual {actual_hash}"
                )
            yield expected_hash, body


def parse_manifest(manifest_bytes: bytes) -> list[PackEntry]:
    """Parse manifest.ndjson into PackEntry list. Lenient on blank lines."""
    out = []
    for line in manifest_bytes.decode("utf-8").splitlines():
        if not line.strip():
            continue
        rec = json.loads(line)
        out.append(PackEntry(leaf_hash=rec["leaf_hash"], size=int(rec["size"])))
    return out