#000061: cold list + total bytes in cold stats

New peers landing on a public bucket need to enumerate pack_hashes
before they can `arborist cold unpack <hash>`. `cold stats` only gives
a count; `cold list` returns the per-pack details:

  - pack_hash (taken from the key)
  - compressed_bytes (one HEAD round-trip via new object_size method)
  - chunk_count (parsed from the small manifest sidecar; skippable
                 via --no-manifest for huge-bucket fast listing)

`cold stats` also now reports total bucket footprint (sums HEAD sizes).

Backend ABC gains `object_size(key) -> int | None` so both subcommands
get sizes without paying egress for the body. S3 impl uses HEAD;
MemoryBackend reads from the dict.

Verified live against DO Spaces NYC3: list-empty → push tiny pack →
list-with-manifest (pack_hash, size=5311 B, chunk_count=5) → list
--no-manifest → cold stats → cleanup.

23 passed in tests/test_cold_object.py + tests/test_evict.py.

Makefile

@@ -41,7 +41,7 @@ SEARCH_Q     ?= computer
         textbooks-summary textbooks-urls fetch-textbooks textbooks-stats textbooks-verify \
         crawl-textbooks crawl-textbooks-stats textbook textbook-list bench-jaggedness \
         monitor-poll monitor-graph monitor-access \
-        bootstrap-object-store cold-pack cold-pack-dvd cold-unpack cold-stats
+        bootstrap-object-store cold-pack cold-pack-dvd cold-unpack cold-stats cold-list
 
 all: bootstrap fetch-cur ingest-cur verify stats ## bootstrap → fetch cur → ingest cur → verify → stats
 
@@ -880,9 +880,12 @@ cold-unpack: bootstrap ## pull pack PACK=<hash> and restore chunks locally
 	@if [ -z "$(PACK)" ]; then echo "PACK=<pack_hash> required"; exit 2; fi
 	$(ARBORIST) --db $(DB) cold unpack $(PACK)
 
-cold-stats: bootstrap ## bucket summary: chunk count, pack count, endpoint
+cold-stats: bootstrap ## bucket summary: pack count, total bytes, endpoint
 	$(ARBORIST) --db $(DB) cold stats
 
+cold-list: bootstrap ## enumerate packs (pack_hash, size, chunk_count) for new-peer hydration
+	$(ARBORIST) --db $(DB) cold list
+
 test: bootstrap ## run pytest suite (excludes opt-in crawler tests)
 	$(VENV)/bin/pytest -q --ignore=tests/crawler -n auto
 

arborist/cli.py

@@ -2889,12 +2889,55 @@ def _cmd_cold_stats(args: argparse.Namespace) -> int:
 
     backend = _make_cold_backend()
     pack_count = 0
+    total_bytes = 0
     for key in backend.list_keys(PACK_PREFIX):
         if key.endswith(".tar.zst"):
             pack_count += 1
+            size = backend.object_size(key)
+            if size is not None:
+                total_bytes += size
     out = {
         "backend": backend.identity.to_audit_body(),
         "packs": pack_count,
+        "packs_compressed_bytes": total_bytes,
+    }
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+    return 0
+
+
+def _cmd_cold_list(args: argparse.Namespace) -> int:
+    """Enumerate packs in the bucket for new-peer hydration.
+
+    For each `<pack>.tar.zst`, HEAD the object to get its compressed size,
+    and (unless --no-manifest) fetch the small manifest sidecar to count
+    chunks. Output is a JSON array — pipe through jq or feed to a
+    hydration script that calls `cold unpack` for each entry.
+    """
+    from arborist.cold_object import PACK_PREFIX, pack_key, parse_manifest
+
+    backend = _make_cold_backend()
+    packs = []
+    for key in backend.list_keys(PACK_PREFIX):
+        if not key.endswith(".tar.zst"):
+            continue
+        # key = "packs/<pack_hash>.tar.zst"
+        pack_hash = key[len(PACK_PREFIX):-len(".tar.zst")]
+        entry = {
+            "pack_hash": pack_hash,
+            "compressed_bytes": backend.object_size(key),
+        }
+        if args.fetch_manifest:
+            try:
+                manifest_bytes = backend.get(pack_key(pack_hash, manifest=True))
+                entry["chunk_count"] = len(parse_manifest(manifest_bytes))
+            except Exception as e:  # noqa: BLE001 — surface but don't fail
+                entry["chunk_count"] = None
+                entry["manifest_error"] = repr(e)
+        packs.append(entry)
+    out = {
+        "backend": backend.identity.to_audit_body(),
+        "pack_count": len(packs),
+        "packs": packs,
     }
     print(json.dumps(out, indent=2, ensure_ascii=False))
     return 0
@@ -5671,10 +5714,23 @@ def build_parser() -> argparse.ArgumentParser:
 
     cold_stats = cold_sub.add_parser(
         "stats",
-        help="bucket summary: pack count + backend identity (no credentials)",
+        help="bucket summary: pack count, total bytes, backend identity (no credentials)",
     )
     cold_stats.set_defaults(func=_cmd_cold_stats)
 
+    cold_list = cold_sub.add_parser(
+        "list",
+        help="enumerate packs in the bucket with metadata (pack_hash, size, chunk_count) for new-peer hydration",
+    )
+    cold_list.add_argument(
+        "--no-manifest",
+        dest="fetch_manifest",
+        action="store_false",
+        help="skip per-pack manifest fetch (faster on huge buckets; "
+        "chunk_count column shows null)",
+    )
+    cold_list.set_defaults(func=_cmd_cold_list, fetch_manifest=True)
+
     activity_cmd = sub.add_parser(
         "activity",
         help="recent Q&A + freshly cached docs (agent-readable timeline)",

arborist/cold_object.py

@@ -113,6 +113,15 @@ class ObjectStoreBackend(abc.ABC):
     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]:
         ...
@@ -233,6 +242,16 @@ class S3CompatibleBackend(ObjectStoreBackend):
                 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):
@@ -277,6 +296,10 @@ class MemoryBackend(ObjectStoreBackend):
     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):

tests/test_cold_object.py

@@ -123,6 +123,15 @@ def test_memory_backend_put_get_head_list():
     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()