bitsStore

#!/usr/bin/env bash
# bitsStore — inspect, verify and clean up the bits S3 store.
# Invoke as `bits store …` (passthrough) or directly as `bitsStore …`.
#
#   bits store ls     [selection]        present the selection (default verb)
#   bits store verify [--arch A] [--deep] [--orphans]
#   bits store rm     [selection] [-n]   delete the objects in the selection
#   bits store ls|rm --stale-boms [--manifests-dir DIR]
#                                        build manifests the store no longer backs
#
# Selection (shared by ls and rm):
#   Manifest-aware (default): reads the store's manifests and lets you narrow by
#     --arch --group --package --version --certified-by --build --expired
#   Views / expansion:
#     --objects          resolve the selection to raw store objects (tarballs)
#     --links            with --objects, also include the named/dist symlinks
#     --orphans          store tarballs NOT referenced by any manifest
#     --broken-links     named/dist links whose backing store object is missing
#                          (half-cleaned store); `rm --broken-links` clears them
#                          so bits will re-upload instead of aborting the upload
#     --raw PATTERN...   select raw store keys by prefix or glob (bypass manifests)
#     --older-than DAYS  keep only objects older than DAYS (raw/orphans/objects)
#     --stale-boms       per-build BOMs no entry of which matches the store any
#                          more (e.g. after a store wipe); `rm` removes them and
#                          `verify --stale-boms` exits 1 while any exist. Narrow
#                          with --arch/--build only. With --manifests-dir DIR,
#                          check the BOM files of a local bits-manifests checkout
#                          instead of MANIFESTS/ — that is the copy certify reads,
#                          so rm there (then commit + MR) is what fixes certify;
#                          rm without it only tidies the store's MANIFESTS/ copies.
#                          In a store bits-orchestrator certifies (MANIFESTS/
#                          orchestrator.json), rm --manifests-dir requests a
#                          decertify action for each stale BOM instead (--console,
#                          a GitLab token), and rm never deletes the orchestrator's
#                          signed manifests (nor its marker, unless named exactly
#                          with --raw and --force).
#                          --deep hashes objects that carry no recorded sha256.
#
# NOTE (hash-only store): new builds write ONLY the content-addressed
# tarball under TARS/<arch>/store/<h2>/<hash>/… — no version-link or dist-symlink
# objects. So `ls`/`rm`/`verify`/`--orphans` operate on the content objects vs
# the signed manifests (the primary model). The `--links` and `--broken-links`
# selectors now only match LEGACY pointer objects left by older builds and
# are retained for one-time migration cleanup (`bits store rm --links …`).
#
# Credentials: $BITS_AWS_KEYS_FILE, else the site's paths.s3_keys_file.
# Store:       $BITS_S3_STORE, else the site's store.operations.
set -euo pipefail

PROG=$(basename "$0")

# bitsStore lives next to the bits_helpers package (bits/), so it imports the
# helpers from its own directory — no dependency on `bits` being on PATH.
SELFDIR="$(cd "$(dirname "$0")" && pwd)"
# bits' own runtime when this install brings one (see bits), else python3 from PATH.
# The runtime ignores PYTHONPATH (-E -s), so it gets this directory as BITSSTORE_SELF.
BITS_PY=(python3)
[[ -x "$SELFDIR/runtime/bin/python3" ]] && BITS_PY=("$SELFDIR/runtime/bin/python3" -E -s)

S3KEYS="${BITS_AWS_KEYS_FILE:-}"
if [[ -z "$S3KEYS" ]]; then
  S3KEYS=$(BITSSTORE_SELF="$SELFDIR" "${BITS_PY[@]}" -c 'import os, sys; sys.path.insert(0, os.environ["BITSSTORE_SELF"]); from bits_helpers import site; site.main(["paths.s3_keys_file"])')
  S3KEYS="${S3KEYS:-~/.bits/s3keys}"
  S3KEYS="${S3KEYS/#\~/$HOME}"
fi

# The Python side reads the creds file as bits build does ([bucket NAME] sections
# included); a missing file is not fatal (--help needs no S3).
export BITS_AWS_KEYS_FILE="$S3KEYS"

BITSSTORE_STORE="${BITS_S3_STORE:-}" BITSSTORE_SELF="$SELFDIR" \
  PYTHONPATH="${SELFDIR}${PYTHONPATH:+:$PYTHONPATH}" \
  exec "${BITS_PY[@]}" - "$@" <<'PY'
import argparse, datetime, fnmatch, hashlib, json, os, sys, tempfile
if sys.flags.ignore_environment:   # bits' own runtime: PYTHONPATH is not read
    sys.path.insert(0, os.environ["BITSSTORE_SELF"])

from bits_helpers.publish import _normalize_s3_store
from bits_helpers.sync import remote_from_url, resolve_and_export_s3_config
resolve_and_export_s3_config()      # the keys file and environment, as bits build reads them
from bits_helpers.utilities import resolve_store_path, resolve_links_path
from bits_helpers import orchestrated, stale_boms
try:
    from bits_helpers import trust
except Exception:
    trust = None

from bits_helpers import site
STORE = os.environ.get("BITSSTORE_STORE") or str(site.get("store.operations") or "")
if not STORE and not set(sys.argv[1:]) & {"-h", "--help"}:
    sys.exit("bits store: no store: set $BITS_S3_STORE, or store.operations in a site "
             "configuration file (see bits doctor)")
PROG = "bits store"
NOW = datetime.datetime.now(datetime.timezone.utc)
# Default bits work dir (S3 client + default MANIFESTS) for the gc/stats verbs;
# mirrors bits_helpers.args without importing that heavy module on every call.
_WORKDIR = (os.environ.get("BITS_WORK_DIR") or os.environ.get("ALICE_WORK_DIR")
            or site.get("paths.work_dir") or "sw")


# ── S3 helpers ────────────────────────────────────────────────────────────────
def make_s3():
    su = _normalize_s3_store(STORE)
    w = remote_from_url(su, su, "x86_64", "sw")     # arch only builds the client
    s3 = getattr(w, "s3", None)
    if s3 is None:
        sys.exit(f"{PROG}: {STORE!r} is not an S3 endpoint. Set $BITS_S3_STORE to an "
                 "https://s3.../<bucket> or b3://<bucket> URL.")
    return s3, (getattr(w, "remoteStore", None) or getattr(w, "writeStore", None))


def list_objs(s3, bucket, prefix=""):
    """Yield (key, size, last_modified) under prefix."""
    for pg in s3.get_paginator("list_objects_v2").paginate(Bucket=bucket, Prefix=prefix):
        for o in pg.get("Contents", []):
            yield o["Key"], o.get("Size", 0), o.get("LastModified")


def get_bytes(s3, bucket, key):
    return s3.get_object(Bucket=bucket, Key=key)["Body"].read()


def human(n):
    if n is None:
        return ""
    n = float(n)
    for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
        if n < 1024 or unit == "TiB":
            return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}"
        n /= 1024.0


def csv(vals):
    out = []
    for v in vals or []:
        out.extend(x.strip() for x in str(v).split(",") if x.strip())
    return out


def norm_sha(v):
    s = str(v or "").strip().lower()
    return s.split(":", 1)[1] if ":" in s else s


def parse_iso(s):
    if not s or s == "-":
        return None
    try:
        return datetime.datetime.fromisoformat(str(s).replace("Z", "+00:00"))
    except Exception:
        return None


# ── manifest model ────────────────────────────────────────────────────────────
class Store:
    def __init__(self, s3, bucket):
        self.s3, self.bucket = s3, bucket
        self.signed = []
        self.boms = []
        self._loaded = False

    def load(self):
        if self._loaded:
            return
        self._loaded = True
        for key, _size, _lm in list_objs(self.s3, self.bucket, "MANIFESTS/"):
            if not key.endswith(".json") or key.endswith(".sig"):
                continue
            try:
                doc = json.loads(get_bytes(self.s3, self.bucket, key))
            except Exception as exc:
                print(f"{PROG}: warning: cannot read {key} ({exc})", file=sys.stderr)
                continue
            if not isinstance(doc, dict):
                continue
            pkgs = [p for p in (doc.get("packages") or []) if isinstance(p, dict)]
            if doc.get("kind") == "common-manifest":
                verified, key_id = self._verify(key)
                exp = doc.get("expires")
                exp_dt = parse_iso(exp)
                self.signed.append({
                    "key": key, "arch": doc.get("architecture") or arch_of(pkgs),
                    "key_id": key_id, "verified": verified,
                    "certified_by": ", ".join(doc.get("certified_by") or []) or "-",
                    "expires": exp or "-",
                    "expired": bool(exp_dt and exp_dt < NOW),
                    "packages": pkgs,
                })
            else:
                self.boms.append({"key": key,
                                  "build_id": doc.get("build_id") or basename(key),
                                  "packages": pkgs})

    def _verify(self, manifest_key):
        if trust is None:
            return (None, "?")
        try:
            with tempfile.TemporaryDirectory() as d:
                mp = os.path.join(d, "m.json")
                with open(mp, "wb") as fh:
                    fh.write(get_bytes(self.s3, self.bucket, manifest_key))
                try:
                    with open(mp + ".sig", "wb") as fh:
                        fh.write(get_bytes(self.s3, self.bucket, manifest_key + ".sig"))
                except Exception:
                    return (False, "unsigned")
                kid = trust.verify_manifest(mp, mp + ".sig")
                return (bool(kid), kid or "untrusted")
        except Exception:
            return (None, "?")

    def manifests(self, f):
        """Manifests to consider, honouring --only-signed/--only-boms."""
        self.load()
        out = []
        if not f.only_boms:
            out += self.signed
        if not f.only_signed:
            out += self.boms
        return out

    def referenced_keys(self, verified_only=False, only_signed=False, only_boms=False):
        """Store object keys named by manifests. The pool is, in precedence:
        only_boms (per-build BOMs only) · only_signed (signed common manifests
        only, regardless of signature state) · verified_only (signed AND
        signature-verified) · default (signed + BOMs, i.e. referenced anywhere).

        only_signed is what "has a PROPER manifest" means: a signed common
        manifest, not a throwaway per-build BOM. It deliberately does NOT require
        a good signature, so a runner without the trust key (verified is None)
        cannot turn every object into an orphan and trigger a mass delete."""
        self.load()
        if only_boms:
            pool = self.boms
        elif only_signed:
            pool = self.signed
        elif verified_only:
            pool = [m for m in self.signed if m.get("verified")]
        else:
            pool = self.signed + self.boms
        out = set()
        for m in pool:
            for p in m["packages"]:
                k = pkg_key(p)
                if k:
                    out.add(k)
        return out


def basename(key):
    return key.rsplit("/", 1)[-1]


def arch_of(pkgs):
    for p in pkgs:
        if p.get("effective_architecture"):
            return p["effective_architecture"]
    return "?"


def pkg_key(pkg):
    arch, h, tar = (pkg.get("effective_architecture"), pkg.get("hash"), pkg.get("tarball"))
    if not (arch and h and tar):
        return None
    return resolve_store_path(arch, h) + "/" + tar


def match_pkg(pkg, f):
    if f.arch and pkg.get("effective_architecture") not in f.arch:
        return False
    if f.group and (pkg.get("group") or "") not in f.group:
        return False
    if f.package and pkg.get("package") not in f.package:
        return False
    if f.version and pkg.get("version") not in f.version:
        return False
    # Provenance filters (from the fields certify now carries into each entry).
    if getattr(f, "bits_version", None) and pkg.get("bits_version") not in f.bits_version:
        return False
    if getattr(f, "bits_dist", None):
        bd = pkg.get("bits_dist_hash") or ""
        if not (bd and any(bd.startswith(x) for x in f.bits_dist)):
            return False
    if getattr(f, "built_before_dt", None):
        # Fail-safe: an object whose completed_at is absent/unparseable is NOT
        # selected by a build-date filter, so a delete never removes an object we
        # cannot date. (Combine with --before to also catch undated old objects.)
        ca = parse_iso(pkg.get("completed_at"))
        if ca is None or not (ca <= f.built_before_dt):
            return False
    return True


def manifest_matches(m, f):
    if f.build and f.build not in (m.get("build_id") or m.get("key", "")):
        return False
    if f.certified_by and f.certified_by not in m.get("certified_by", ""):
        return False
    if f.expired and not m.get("expired"):     # only meaningful for signed manifests
        return False
    return True


# ── selection ────────────────────────────────────────────────────────────────
def select_objects(store, s3, bucket, f):
    keys = {}
    linkspecs = []
    for m in store.manifests(f):
        if not manifest_matches(m, f):
            continue
        for p in m["packages"]:
            if not match_pkg(p, f):
                continue
            k = pkg_key(p)
            if k:
                keys[k] = None
                linkspecs.append((p.get("effective_architecture"), p.get("package"),
                                  p.get("version"), p.get("revision")))
    if f.links:
        add_link_objects(s3, bucket, keys, linkspecs)
    apply_age(s3, bucket, keys, f)
    return keys


def add_link_objects(s3, bucket, keys, linkspecs):
    """Include named/dist symlink objects for each selected package version."""
    seen_dirs = {}
    for arch, package, version, revision in linkspecs:
        if not (arch and package):
            continue
        d = resolve_links_path(arch, package) + "/"
        if d not in seen_dirs:
            seen_dirs[d] = list(list_objs(s3, bucket, d))
        # Match a symlink for THIS version: "<version>-<revision>" (or the
        # version alone) to avoid pulling in sibling versions of the package.
        tag = f"{version}-{revision}" if (version and revision) else (version or "")
        for key, size, _lm in seen_dirs[d]:
            if tag and tag in key:
                keys.setdefault(key, size)


def select_orphans(store, s3, bucket, f):
    # An orphan is a content tarball no manifest names. By default any manifest
    # counts (signed common manifests + per-build BOMs). --only-signed narrows the
    # reference set to signed common manifests, so a tarball that only ever
    # appeared in a throwaway BOM (or in nothing) counts as an orphan too — i.e.
    # "has no PROPER manifest". --only-boms is the mirror.
    referenced = store.referenced_keys(only_signed=f.only_signed,
                                       only_boms=f.only_boms)
    orphans = {}
    prefixes = ["TARS/%s/store/" % a for a in f.arch] if f.arch else ["TARS/"]
    for prefix in prefixes:
        for key, size, lm in list_objs(s3, bucket, prefix):
            if "/store/" not in key or not key.endswith(".tar.gz"):
                continue
            if key not in referenced and age_ok(lm, f):
                orphans[key] = size
    return orphans


def orphan_markers(s3, bucket, orphan_keys):
    """rev-index markers (MANIFESTS/rev-index/<arch>/<pkg>/<ver-rev>, body = hash)
    whose (pkg, ver-rev, arch) matches an orphan tarball. Included with --orphans
    so `rm --orphans` clears the dangling metadata pointer alongside the
    unreferenced content object — otherwise a stale marker makes bits think the
    package still exists at that hash. Matched by the marker key (which separates
    pkg from ver-rev unambiguously) against orphan tarball basenames."""
    basenames = {k.rsplit("/", 1)[-1] for k in orphan_keys}
    markers = {}
    for key, size, _lm in list_objs(s3, bucket, "MANIFESTS/rev-index/"):
        parts = key.split("/")          # MANIFESTS / rev-index / <arch> / <pkg> / <ver-rev>
        if len(parts) < 5:
            continue
        arch, pkg, verrev = parts[2], parts[3], parts[4]
        if f"{pkg}-{verrev}.{arch}.tar.gz" in basenames:
            markers[key] = size
    return markers


def select_broken_links(store, s3, bucket, f):
    """Named/dist link objects whose backing content object is gone.

    A link (``TARS/<arch>/<pkg>/<file>.tar.gz`` — i.e. NOT under ``/store/``) is
    a small S3 object whose body points at ``TARS/<arch>/store/<h2>/<hash>/…``.
    When the target content object has been removed (a half-cleaned store), the
    link dangles and bits' uploader refuses to re-upload the package. This finds
    those dangling links so ``rm --broken-links`` can clear them.

    Content objects are addressed by hash and are indexed here in one listing
    pass, so each link is resolved by set-membership rather than a per-link HEAD.
    """
    prefixes = ["TARS/%s/" % a for a in f.arch] if f.arch else ["TARS/"]
    broken = {}
    for prefix in prefixes:
        store_keys, links = set(), []
        for key, size, lm in list_objs(s3, bucket, prefix):
            if not key.endswith(".tar.gz"):
                continue
            if "/store/" in key:
                store_keys.add(key)
            else:
                links.append((key, size, lm))
        for key, size, lm in links:
            try:
                body = get_bytes(s3, bucket, key).decode("utf-8", "replace").strip()
            except Exception:
                body = ""
            parts = key.split("/")
            arch = parts[1] if len(parts) > 1 else ""
            idx = body.find("store/")
            target = "TARS/%s/%s" % (arch, body[idx:]) if (arch and idx >= 0) else ""
            if (not target or target not in store_keys) and age_ok(lm, f):
                broken[key] = size
    return broken


def select_raw(s3, bucket, f):
    out = {}
    for pat in f.raw:
        if any(c in pat for c in "*?["):
            base = pat.split("*", 1)[0].split("?", 1)[0].split("[", 1)[0]
            for key, size, lm in list_objs(s3, bucket, base):
                if fnmatch.fnmatchcase(key, pat) and age_ok(lm, f):   # case-sensitive
                    out[key] = size
        else:                                          # plain prefix
            for key, size, lm in list_objs(s3, bucket, pat):
                if age_ok(lm, f):
                    out[key] = size
    return out


def resolve_before(s3, bucket, value):
    """Turn the --before VALUE into an absolute UTC cutoff datetime.

    Accepts, in order: an ISO-8601 timestamp; a compact build-id / manifest-name
    stamp ``YYYYMMDDTHHMMSSZ``; or an S3 key (or unique substring of one, e.g. a
    manifest/BOM name or build_id) whose own upload time is used. The last form is
    the point of this: to purge "this publication and earlier", pass the manifest
    that was published and let its upload time bound the deletion.

    Returns a tz-aware datetime, or None if VALUE cannot be resolved (the caller
    treats that as a hard error rather than silently widening the selection).
    """
    if not value:
        return None
    dt = parse_iso(value)
    if dt is None:
        try:
            dt = datetime.datetime.strptime(value, "%Y%m%dT%H%M%SZ")
        except ValueError:
            dt = None
    if dt is not None:
        return dt if dt.tzinfo else dt.replace(tzinfo=datetime.timezone.utc)
    # Not a timestamp — treat as an S3 object (exact key first, then substring).
    try:
        lm = s3.head_object(Bucket=bucket, Key=value).get("LastModified")
        if lm:
            return lm
    except Exception:
        pass
    try:
        cands = [lm for k, _s, lm in list_objs(s3, bucket, "MANIFESTS/")
                 if value in k and lm]
    except Exception:
        return None
    return max(cands) if cands else None


def age_ok(last_modified, f):
    """True if the object passes the age filters. --older-than is a relative
    window (older than N days); --before is an absolute cutoff (uploaded at or
    before a fixed instant, e.g. a publication's timestamp). Both, when set, must
    hold. An object with no known LastModified fails any active filter."""
    older = getattr(f, "older_than", None)
    before = getattr(f, "before_dt", None)
    if not older and not before:
        return True
    if last_modified is None:
        return False
    if older and not (last_modified < (NOW - datetime.timedelta(days=older))):
        return False
    if before and not (last_modified <= before):
        return False
    return True


def apply_age(s3, bucket, keys, f):
    """Drop manifest-derived keys outside the age filters (HEAD each). No-op when
    neither --older-than nor --before is set (the common case)."""
    if not getattr(f, "older_than", None) and not getattr(f, "before_dt", None):
        return
    for k in list(keys):
        try:
            lm = s3.head_object(Bucket=bucket, Key=k).get("LastModified")
        except Exception:
            keys.pop(k, None)                          # gone → not a delete target
            continue
        if not age_ok(lm, f):
            keys.pop(k, None)


def stale_bom_report(store, s3, bucket, f):
    """Classify the BOMs (MANIFESTS/ or --manifests-dir) against the store, print
    the stale ones, and return ``{key_or_path: None}`` for the fully stale ones."""
    if f.manifests_dir:
        boms = stale_boms.load_local_boms(f.manifests_dir)
        where = f.manifests_dir
    else:
        store.load()
        boms = store.boms
        where = "MANIFESTS/"
    lookup = stale_boms.store_lookup(s3, bucket, deep=f.deep)
    full, partial, checked, confirmed = [], [], 0, 0
    for m in boms:
        if not manifest_matches(m, f):
            continue
        if f.arch and arch_of(m["packages"]) not in f.arch:
            continue
        c = stale_boms.classify(m, lookup)
        if not c["judged"]:
            continue
        checked += 1
        confirmed += c[stale_boms.OK]
        if c["fully_stale"]:
            full.append((m, c))
        elif c[stale_boms.STALE] or c[stale_boms.MISSING]:
            partial.append((m, c))
    for title, rows in (("STALE BUILD MANIFESTS (no entry matches the store - prune candidates)", full),
                        ("KEPT (some entries still match the store, cannot be checked, "
                         "or are merely absent)", partial)):
        if not rows:
            continue
        print(title)
        print("  %-40s %-22s %5s %5s %7s %7s  %s" %
              ("build_id", "arch", "ok", "stale", "missing", "unknown", "manifest"))
        for m, c in sorted(rows, key=lambda r: r[0]["key"]):
            print("  %-40s %-22s %5d %5d %7d %7d  %s" %
                  (m["build_id"][:40], arch_of(m["packages"])[:22], c[stale_boms.OK],
                   c[stale_boms.STALE], c[stale_boms.MISSING], c[stale_boms.UNKNOWN], m["key"]))
        print()
    print(f"# {len(full)} stale and {len(partial)} partly stale of {checked} "
          f"build manifest(s) checked under {where} against {STORE}")
    if full and not confirmed and not f.force:
        # Not one entry anywhere matches: far more likely the wrong store
        # ($BITS_S3_STORE) than every BOM being stale. After a real wipe and
        # rebuild, the new BOMs always match.
        print(f"{PROG}: the store confirms no entry of any BOM checked — wrong store? "
              "Set $BITS_S3_STORE, or pass --force (rm) to select them anyway.",
              file=sys.stderr)
        return {}
    return {m["key"]: (m if f.manifests_dir else None) for m, _c in full}


def delete_local_files(paths, f):
    """Remove stale BOM files from a local bits-manifests checkout (--manifests-dir).
    Same dry-run / confirmation rules as the S3 delete."""
    paths = sorted(paths)
    if not paths:
        print("Nothing matched; nothing to delete.")
        return 0
    if f.dry_run:
        print(f"[dry-run] would remove {len(paths)} file(s) (nothing removed)")
        return 0
    if not f.yes:
        try:
            with open("/dev/tty") as tty:
                sys.stderr.write(f"Remove {len(paths)} file(s) from {f.manifests_dir} "
                                 f"(checked against {STORE})? [y/N] ")
                sys.stderr.flush()
                if tty.readline().strip().lower() not in ("y", "yes"):
                    print("Aborted."); return 1
        except OSError:
            print("Refusing to delete without confirmation (no tty). Use --yes.", file=sys.stderr)
            return 1
    failed = 0
    for path in paths:
        try:
            os.remove(path)
        except OSError as exc:
            failed += 1
            print(f"ERROR removing {path}: {exc}", file=sys.stderr)
    print(f"Done: removed {len(paths) - failed}/{len(paths)} file(s). Commit the removal "
          "in the manifests repo and open an MR.")
    return 1 if failed else 0


def decertify_local(boms, f, marker):
    """In a store bits-orchestrator certifies, the BOM files of its bits-manifests
    mirror (--manifests-dir) are not removed: each stale one is asked out of the
    certified set by a decertify action, whose group is the file's directory
    (manifests/<group>/). Same dry-run / confirmation rules as a delete."""
    from bits_helpers import forge
    wanted, bad = stale_boms.decertify_targets(boms)
    for path in bad:
        print(f"{PROG}: {path} is not a BOM under manifests/<group>/; skipped", file=sys.stderr)
    if not wanted:
        print("Nothing matched; nothing to decertify.")
        return 1 if bad else 0
    print(f"{orchestrated.describe(marker)} owns the certified set: the stale BOMs are "
          "decertified, not removed")
    for (group, bid, arch), paths in sorted(wanted.items()):
        print(f"    {group:<12} {bid[:40]:<40} {arch:<22} {len(paths)} BOM(s)")
    if f.dry_run:
        print(f"[dry-run] would request {len(wanted)} decertify action(s) (nothing requested)")
        return 0
    url = orchestrated.console_url(f.console)
    token = forge.resolve_gitlab_token(None)
    if not (url and token):
        sys.exit(f"{PROG}: decertifying needs the bits-console URL (--console, "
                 "$BITS_CONSOLE_URL) and a GitLab token (~/.bits/gitlab-token)")
    if not f.yes:
        try:
            with open("/dev/tty") as tty:
                sys.stderr.write(f"Ask {url} to decertify {len(wanted)} build(s)? [y/N] ")
                sys.stderr.flush()
                if tty.readline().strip().lower() not in ("y", "yes"):
                    print("Aborted."); return 1
        except OSError:
            print("Refusing to decertify without confirmation (no tty). Use --yes.",
                  file=sys.stderr)
            return 1
    console = orchestrated.Console(url, token, cafile=f.console_cafile,
                                   insecure=f.console_insecure)
    failed = len(bad)
    for (group, bid, arch) in sorted(wanted):
        try:
            action = console.settle(console.decertify(
                group, bid, arch, "stale BOM: the store no longer backs it"), timeout=30)
        except orchestrated.RequestError as exc:
            print(f"ERROR decertify {bid} ({arch}) in {group}: {exc}", file=sys.stderr)
            failed += 1
            continue
        ok, words = console.outcome(action)
        print(f"decertify {bid} ({arch}) in {group}: action {action.get('id')} {words}",
              file=sys.stdout if ok else sys.stderr)
        failed += 0 if ok else 1
    return 1 if failed else 0


# ── presentation ─────────────────────────────────────────────────────────────
def present_manifests(store, f):
    store.load()
    signed = [m for m in store.signed
              if (not f.arch or m["arch"] in f.arch) and manifest_matches(m, f)]

    if signed:
        print("SIGNED COMMON MANIFESTS")
        print("  %-20s %-8s %-8s %-12s %-5s  %s" %
              ("arch", "verified", "state", "certified_by", "pkgs", "manifest"))
        for m in sorted(signed, key=lambda x: (x["arch"], x["key"])):
            pkgs = [p for p in m["packages"] if match_pkg(p, f)]
            ver = {True: "yes", False: "NO", None: "?"}[m["verified"]]
            state = "EXPIRED" if m["expired"] else "live"
            b = basename(m["key"])
            tag = "  (legacy monolith)" if b == "common-manifest.json" else ""
            print("  %-20s %-8s %-8s %-12s %-5d  %s%s" %
                  (m["arch"], ver, state, m["certified_by"], len(pkgs), b, tag))
            if f.long:
                for p in sorted(pkgs, key=lambda x: x.get("package", "")):
                    print("        %-24s %-14s %s" % (p.get("package", "?"),
                          p.get("version", ""), (p.get("hash", "") or "")[:12]))

    rows = []
    for b in ([] if f.only_signed else store.boms):
        if not manifest_matches(b, f):
            continue
        pkgs = [p for p in b["packages"] if match_pkg(p, f)]
        if not pkgs and (f.arch or f.group or f.package or f.version):
            continue
        arches = sorted({p.get("effective_architecture", "?") for p in pkgs}) or ["?"]
        groups = sorted({p.get("group") or "-" for p in pkgs}) or ["-"]
        rows.append((b["build_id"], ",".join(arches), ",".join(groups), len(pkgs), b["key"]))
    if rows:
        print("\nBUILD MANIFESTS (per-build BOMs)")
        print("  %-30s %-18s %-8s %-5s %s" % ("build_id", "arch", "group", "pkgs", "manifest key"))
        for bid, arch, grp, n, key in sorted(rows):
            print("  %-30s %-18s %-8s %-5d %s" % (bid[:30], arch[:18], grp[:8], n, key))

    if not signed and not rows:
        print("No manifests matched under MANIFESTS/.")

    store_overview(store.s3, store.bucket)
    print("\n(--objects: backing store objects · --orphans: unreferenced tarballs · "
          "--raw PATTERN: raw keys · verify: check signatures/checksums)")


def store_overview(s3, bucket):
    """Summarise TARS/<arch>/ trees, split into content tarballs vs symlinks/other.
    A tree with 0 content tarballs but many small objects is just symlink
    scaffolding (e.g. from a build that failed before uploading real tarballs)."""
    stats = {}   # arch -> [tar_n, tar_b, other_n, other_b]
    for key, size, _lm in list_objs(s3, bucket, "TARS/"):
        parts = key.split("/")
        if len(parts) < 2 or not parts[1]:
            continue
        row = stats.setdefault(parts[1], [0, 0, 0, 0])
        if "/store/" in key and key.endswith(".tar.gz"):
            row[0] += 1; row[1] += size or 0
        else:
            row[2] += 1; row[3] += size or 0
    if stats:
        print("\nSTORE (TARS/) BY ARCHITECTURE      content tarballs | symlinks/other")
        for a in sorted(stats):
            tn, tb, on, ob = stats[a]
            print("  %-22s  %5d  %9s  |  %5d  %9s" % (a, tn, human(tb), on, human(ob)))


def present_objects(keys, title):
    total = 0
    have_size = False
    for k in sorted(keys):
        sz = keys[k]
        if isinstance(sz, int):
            total += sz; have_size = True
            print(f"{human(sz):>10}  {k}")
        else:
            print(f"    {k}")
    print(f"# {title}: {len(keys)} object(s)" + (f", {human(total)}" if have_size else ""))


# ── verify ────────────────────────────────────────────────────────────────────
def sha256_of(s3, bucket, key):
    try:
        h = hashlib.sha256()
        body = s3.get_object(Bucket=bucket, Key=key)["Body"]
        for chunk in iter(lambda: body.read(1 << 20), b""):
            h.update(chunk)
        return h.hexdigest()
    except Exception:
        return None


def do_verify(store, s3, bucket, f):
    store.load()
    signed = [m for m in store.signed if (not f.arch or m["arch"] in f.arch)]
    if not signed:
        print("No signed common manifests found.")
    rc = 0
    for m in sorted(signed, key=lambda x: x["arch"]):
        ver = {True: "yes", False: "NO", None: "?"}[m["verified"]]
        state = "EXPIRED" if m["expired"] else "live"
        b = basename(m["key"])
        legacy = "  (legacy monolith — superseded by per-arch, safe to delete)" \
                 if b == "common-manifest.json" else ""
        print(f"{m['arch']}: signature={ver} state={state} signer={m['key_id'][:16]} "
              f"certified_by={m['certified_by']} expires={str(m['expires'])[:10]} "
              f"packages={len(m['packages'])} manifest={b}{legacy}")
        if m["verified"] is not True or m["expired"]:
            rc = 1
        missing, mismatch = [], []
        for p in m["packages"]:
            k = pkg_key(p)
            if not k:
                continue
            try:
                s3.head_object(Bucket=bucket, Key=k)
            except Exception:
                missing.append(k); continue
            if f.deep:
                claimed = norm_sha(p.get("tarball_sha256"))
                actual = sha256_of(s3, bucket, k)
                if claimed and actual and claimed != actual:
                    mismatch.append(k)
        for label, lst in (("MISSING from store", missing), ("SHA256 MISMATCH", mismatch)):
            if lst:
                rc = 1
                print(f"  {label}: {len(lst)}")
                for x in lst[:10]:
                    print("    ", x)
                if len(lst) > 10:
                    print(f"     ... and {len(lst)-10} more")
    if f.orphans:
        orph = select_orphans(store, s3, bucket, f)
        print(f"\norphans (tarballs referenced by no manifest): {len(orph)}"
              + (f", {human(sum(orph.values()))}" if orph else ""))
    if f.stale_boms:
        print()
        if stale_bom_report(store, s3, bucket, f):
            rc = 1
    return rc


# ── delete ───────────────────────────────────────────────────────────────────
def do_cat(s3, bucket, key):
    """Dump a store object's bytes to stdout — handy to inspect a rev-index
    marker (its hash), a signed manifest, or any small object without a browser.
    Written verbatim so binary content (e.g. a tarball) can be piped onward."""
    try:
        data = get_bytes(s3, bucket, key)
    except Exception as exc:
        if "NoSuchKey" in str(exc) or "404" in str(exc):
            sys.exit(f"{PROG}: no such object: {key}")
        sys.exit(f"{PROG}: cannot read {key!r}: {exc}")
    try:
        sys.stdout.buffer.write(data)
        sys.stdout.buffer.flush()
    except BrokenPipeError:
        pass
    return 0


def do_delete(s3, bucket, keys, store, f):
    if not keys:
        print("Nothing matched; nothing to delete.")
        return 0
    protected = set() if f.force else store.referenced_keys(verified_only=True)
    hits = sorted(keys)
    marker = getattr(f, "marker", None)
    if marker:
        # The orchestrator's own manifests, always; its marker unless named exactly
        # with --force, as going back to the manifests CI does:
        # `rm --raw MANIFESTS/orchestrator.json --force`.
        named = f.force and orchestrated.MARKER in (f.raw or [])
        owned = [k for k in hits if k.startswith(marker["manifests"])
                 or (k == orchestrated.MARKER and not named)]
        if owned:
            print(f"{PROG}: {len(owned)} object(s) are {orchestrated.describe(marker)}'s "
                  "signed manifests or its marker — not deleted:", file=sys.stderr)
            for k in owned[:10]:
                print("   ", k, file=sys.stderr)
            hits = [k for k in hits if k not in owned]
    blocked = [k for k in hits if k in protected]
    deletable = [k for k in hits if k not in protected]
    if blocked:
        print(f"{PROG}: {len(blocked)} object(s) are referenced by a VERIFIED signed "
              f"manifest — protected (use --force to override):", file=sys.stderr)
        for k in blocked[:10]:
            print("   ", k, file=sys.stderr)
        if len(blocked) > 10:
            print(f"    ... and {len(blocked)-10} more", file=sys.stderr)
    total = sum(v for v in (keys[k] for k in deletable) if isinstance(v, int))
    print(f"Selected {len(deletable)} object(s)"
          + (f", {human(total)}" if total else "") + " for deletion.")
    for k in deletable[:20]:
        print("   ", k)
    if len(deletable) > 20:
        print(f"    ... and {len(deletable)-20} more")
    if not deletable:
        return 0
    if f.dry_run:
        print(f"[dry-run] would delete {len(deletable)} object(s) (nothing deleted)")
        return 0
    if not f.yes:
        try:
            with open("/dev/tty") as tty:
                sys.stderr.write(f"Delete {len(deletable)} object(s) from s3://{bucket}? [y/N] ")
                sys.stderr.flush()
                if tty.readline().strip().lower() not in ("y", "yes"):
                    print("Aborted."); return 1
        except OSError:
            print("Refusing to delete without confirmation (no tty). Use --yes.", file=sys.stderr)
            return 1
    deleted = 0
    for i in range(0, len(deletable), 1000):
        batch = [{"Key": k} for k in deletable[i:i + 1000]]
        resp = s3.delete_objects(Bucket=bucket, Delete={"Objects": batch, "Quiet": True})
        errs = resp.get("Errors") or []
        for e in errs:
            print("ERROR deleting", e.get("Key"), "-", e.get("Message"), file=sys.stderr)
        deleted += len(batch) - len(errs)
        print(f"deleted {min(i + 1000, len(deletable))}/{len(deletable)}")
    print(f"Done: deleted {deleted}/{len(deletable)} object(s).")
    return 0 if deleted == len(deletable) else 1


# ── CLI ──────────────────────────────────────────────────────────────────────
def add_stale_boms(p):
    p.add_argument("--stale-boms", dest="stale_boms", action="store_true",
                   help="build manifests (BOMs) none of whose entries match the store "
                        "any more, e.g. after a store wipe; rm removes them")
    p.add_argument("--manifests-dir", dest="manifests_dir", metavar="DIR",
                   help="with --stale-boms, check the BOM files of a local bits-manifests "
                        "checkout instead of the store's MANIFESTS/ (rm deletes the files)")


def add_selection(p):
    p.add_argument("--arch", action="append", metavar="ARCH", help="filter by architecture (repeatable/comma)")
    p.add_argument("--group", action="append", metavar="GROUP", help="filter by group")
    p.add_argument("--package", action="append", metavar="PKG", help="filter by package")
    p.add_argument("--version", action="append", metavar="VER", help="filter by version")
    p.add_argument("--bits-version", dest="bits_version", action="append", metavar="VER",
                   help="filter by the bits version that built the object (repeatable/comma)")
    p.add_argument("--bits-dist", dest="bits_dist", action="append", metavar="HASH",
                   help="filter by bits dist fingerprint (BITS_DIST_HASH); prefix match (repeatable/comma)")
    p.add_argument("--built-before", dest="built_before", metavar="WHEN",
                   help="keep only objects BUILT at or before WHEN (the recorded completed_at) — "
                        "an ISO-8601 time or a YYYY-MM-DD date. Unlike --before (which uses the S3 "
                        "upload time) this uses build provenance, so it purges a whole bits "
                        "generation regardless of when objects were (re-)uploaded.")
    p.add_argument("--certified-by", dest="certified_by", metavar="USER", help="filter by certifier")
    p.add_argument("--build", metavar="ID", help="filter by build-id substring")
    p.add_argument("--expired", action="store_true", help="only expired signed manifests")
    p.add_argument("--objects", action="store_true", help="resolve selection to raw store objects")
    p.add_argument("--links", action="store_true", help="with --objects, include named/dist symlinks")
    p.add_argument("--orphans", action="store_true", help="store tarballs not referenced by any manifest")
    p.add_argument("--broken-links", dest="broken_links", action="store_true",
                   help="named/dist links whose backing store object is missing")
    add_stale_boms(p)
    p.add_argument("--deep", action="store_true",
                   help="with --stale-boms, hash store objects that carry no recorded sha256 (slow)")
    p.add_argument("--raw", nargs="+", metavar="PATTERN", help="select raw keys by prefix or glob")
    p.add_argument("--older-than", dest="older_than", type=float, metavar="DAYS",
                   help="keep only objects older than DAYS")
    p.add_argument("--before", dest="before", metavar="WHEN",
                   help="keep only objects uploaded at or before WHEN — an ISO-8601 "
                        "time, a compact stamp YYYYMMDDTHHMMSSZ, or a manifest/BOM key "
                        "(or build_id) whose upload time is used. Purges a publication "
                        "'and earlier'; combine with --arch/--raw to scope the trees.")
    p.add_argument("--only-signed", action="store_true", help="consider only signed common manifests")
    p.add_argument("--only-boms", action="store_true", help="consider only per-build BOMs")
    p.add_argument("-l", "--long", action="store_true", help="show more detail (ls)")


def main():
    ap = argparse.ArgumentParser(prog=PROG, description="Inspect, verify and clean the bits S3 store.")
    sub = ap.add_subparsers(dest="verb")
    for verb in ("ls", "rm"):
        sp = sub.add_parser(verb, help="list" if verb == "ls" else "delete")
        add_selection(sp)
        if verb == "rm":
            sp.add_argument("--console", metavar="URL", default=None,
                            help="with --stale-boms --manifests-dir in a store bits-orchestrator "
                                 "certifies: the bits-console to request the decertify actions "
                                 "from (default: $BITS_CONSOLE_URL)")
            sp.add_argument("--console-cafile", dest="console_cafile", metavar="PEM", default=None,
                            help="CA bundle for the console when it is not in this host's trust store")
            sp.add_argument("--console-insecure", dest="console_insecure", action="store_true",
                            help="allow http:// / skip TLS verification to the console (testbed only)")
            sp.add_argument("-n", "--dry-run", action="store_true", help="show what would be deleted")
            sp.add_argument("--yes", action="store_true", help="skip the confirmation prompt")
            sp.add_argument("--force", action="store_true",
                            help="allow deleting verified-manifest objects / catch-all patterns")
    vp = sub.add_parser("verify", help="verify signed manifests against the store")
    vp.add_argument("--arch", action="append", metavar="ARCH", help="limit to architecture(s)")
    vp.add_argument("--deep", action="store_true", help="also re-hash every tarball (slow)")
    vp.add_argument("--orphans", action="store_true", help="also report unreferenced tarballs")
    add_stale_boms(vp)
    cp = sub.add_parser("cat", help="dump a store object's bytes to stdout")
    cp.add_argument("key", metavar="KEY",
                    help="S3 object key, e.g. MANIFESTS/rev-index/<arch>/<pkg>/<ver-rev> "
                         "or TARS/<arch>/store/<h2>/<hash>/<file>")

    # gc / stats: store-scoped ops folded in from the old top-level `bits gc` /
    # `bits store-stats`. Their handlers live in bits_helpers and build their own
    # S3 client, so the dest names below must match what they read.
    gp = sub.add_parser("gc", help="reachability GC of the shared store")
    gp.add_argument("--trust-manifest", dest="trustManifest", required=True, metavar="PATH",
                    help="signed common manifest whose hashes are the GC roots (must verify); "
                         "with several stores one per store, comma-separated, in the same order")
    gp.add_argument("--remote-store", "--store", dest="gcStore", metavar="URL", default=STORE,
                    help="S3 store URL/bucket to sweep; several, comma-separated, each in "
                         "turn (default: %(default)s)")
    gp.add_argument("-a", "--architecture", dest="architecture", metavar="ARCH", default=None,
                    help="architecture store tree to sweep (default: detected)")
    gp.add_argument("-w", "--work-dir", dest="workDir", metavar="WORKDIR", default=_WORKDIR,
                    help="bits work directory for the S3 client (default: %(default)s)")
    gp.add_argument("--grace-days", dest="graceDays", type=float, default=7.0, metavar="DAYS",
                    help="never sweep an object younger than DAYS (default: %(default)s)")
    gp.add_argument("--allow-empty", dest="allowEmpty", action="store_true", default=False,
                    help="permit sweeping when the verified manifest has zero roots (dangerous)")
    gp.add_argument("-n", "--dry-run", dest="dryRun", action="store_true", default=False,
                    help="report what would be swept without deleting")

    tp = sub.add_parser("stats", help="summarise store usage (per-arch + per-build/signed)")
    tp.add_argument("--remote-store", "--store", dest="storeStatsStore", metavar="URL", default=STORE,
                    help="S3 store URL/bucket to summarise (default: %(default)s)")
    tp.add_argument("--manifests", dest="manifests", metavar="PATH", nargs="*", default=None,
                    help="build-manifest JSON files/dirs (default: WORKDIR/MANIFESTS)")
    tp.add_argument("--trust-manifest", dest="trustManifest", metavar="PATH", default=None,
                    help="comma-separated signed common manifests marking which builds are signed")
    tp.add_argument("--tars-prefix", dest="tarsPrefix", metavar="PREFIX", default="TARS/",
                    help="store root prefix under which <arch>/store/... lives (default: %(default)s)")
    tp.add_argument("-o", "--out", dest="out", metavar="FILE", default="store.json",
                    help="path to write the store document (default: %(default)s)")
    tp.add_argument("--monitor-url", dest="monitorUrl", metavar="URL", default=None,
                    help="also POST Prometheus gauges here (falls back to $METRICS_URL)")
    tp.add_argument("-a", "--architecture", dest="architecture", metavar="ARCH", default=None,
                    help="architecture for store-path resolution (default: detected)")
    tp.add_argument("-w", "--work-dir", dest="workDir", metavar="WORKDIR", default=_WORKDIR,
                    help="bits work directory (S3 client + default MANIFESTS) (default: %(default)s)")

    up = sub.add_parser("upload", help="upload one built package's tarball to the S3 store")
    up.add_argument("package", metavar="PACKAGE", help="package name to upload (build it first)")
    up.add_argument("--version", dest="version", metavar="VER", default=None,
                    help="package version (default: the newest built)")
    up.add_argument("--remote-store", "--store", "--write-store", dest="uploadStore", metavar="URL",
                    default=STORE, help="S3 store to write to (default: %(default)s)")
    up.add_argument("-a", "--architecture", dest="architecture", metavar="ARCH", default=None,
                    help="architecture (default: detected)")
    up.add_argument("-w", "--work-dir", dest="workDir", metavar="WORKDIR", default=_WORKDIR,
                    help="bits work directory holding the built package (default: %(default)s)")
    up.add_argument("-n", "--dry-run", dest="dryRun", action="store_true", default=False,
                    help="report what would be uploaded without writing")

    argv = sys.argv[1:]
    if not argv or argv[0] not in ("ls", "rm", "verify", "cat", "gc", "stats", "upload", "-h", "--help"):
        argv = ["ls"] + argv                          # default verb
    args = ap.parse_args(argv)

    # Store-scoped verbs (gc/stats/upload) delegate to bits_helpers handlers and
    # do NOT use the ls/rm selection machinery — handle them before the
    # selection-attr normalisation below (which would csv-mangle package/version).
    if args.verb in ("gc", "stats", "upload"):
        from bits_helpers.publish import _normalize_s3_store
        from bits_helpers.sync import resolve_and_export_s3_config, s3_settings, store_urls
        resolve_and_export_s3_config()
        stores = store_urls(getattr(args, {"gc": "gcStore", "stats": "storeStatsStore",
                                           "upload": "uploadStore"}[args.verb], ""))
        for url in stores:                # each bucket's own credentials, else the run's
            bucket = _normalize_s3_store(url).split("://", 1)[-1].split("/", 1)[0]
            if not s3_settings(bucket)["access_key"]:
                sys.exit(f"{PROG}: no S3 credentials for {url} — put them in ~/.bits/s3keys "
                         "(or set $BITS_AWS_KEYS_FILE / AWS_ACCESS_KEY_ID, or "
                         "BITS_S3_<BUCKET>_ACCESS_KEY_ID for that bucket).")
        if not getattr(args, "architecture", None):
            from bits_helpers.arch import detectArch
            args.architecture = detectArch()
        if args.verb == "gc":
            from bits_helpers.gc import doGc
            doGc(args, ap)
            return 0
        if args.verb == "stats":
            from bits_helpers.store_stats import doStoreStats
            return doStoreStats(args, ap) or 0
        from bits_helpers.publish import _publish_s3       # upload, to every store given
        for url in stores:
            _publish_s3(args.package, args.version, args.architecture, args.workDir,
                        url, ap, dry_run=args.dryRun)
        return 0

    # Normalise multi-value filters; give safe defaults for attrs absent on a verb.
    for name in ("arch", "group", "package", "version", "bits_version", "bits_dist"):
        setattr(args, name, csv(getattr(args, name, None)))
    for name, dflt in (("certified_by", None), ("build", None), ("expired", False),
                       ("objects", False), ("links", False), ("orphans", False),
                       ("broken_links", False), ("stale_boms", False), ("manifests_dir", None),
                       ("raw", None), ("older_than", None), ("before", None),
                       ("bits_version", None), ("bits_dist", None), ("built_before", None),
                       ("only_signed", False),
                       ("only_boms", False), ("long", False), ("dry_run", False),
                       ("yes", False), ("force", False), ("deep", False),
                       ("console", None), ("console_cafile", None), ("console_insecure", False)):
        if not hasattr(args, name):
            setattr(args, name, dflt)
    args.before_dt = None
    args.built_before_dt = None

    if args.manifests_dir and not args.stale_boms:
        sys.exit(f"{PROG}: --manifests-dir only applies with --stale-boms")
    if args.deep and args.verb in ("ls", "rm") and not args.stale_boms:
        sys.exit(f"{PROG}: --deep only applies with --stale-boms (or to verify)")
    if args.stale_boms and args.verb in ("ls", "rm"):
        # --stale-boms selects whole BOMs by --arch/--build only. Refuse the other
        # selectors rather than silently ignore them and widen a delete.
        other = [opt for opt, val in (
            ("--group", args.group), ("--package", args.package), ("--version", args.version),
            ("--bits-version", args.bits_version), ("--bits-dist", args.bits_dist),
            ("--built-before", args.built_before), ("--certified-by", args.certified_by),
            ("--expired", args.expired), ("--objects", args.objects), ("--links", args.links),
            ("--orphans", args.orphans), ("--broken-links", args.broken_links),
            ("--raw", args.raw), ("--older-than", args.older_than), ("--before", args.before),
            ("--only-signed", args.only_signed), ("--only-boms", args.only_boms)) if val]
        if other:
            sys.exit(f"{PROG}: --stale-boms takes only --arch/--build to narrow it, not "
                     f"{', '.join(other)} (for one group, point --manifests-dir at "
                     "manifests/<group>)")
    if args.manifests_dir and not os.path.isdir(args.manifests_dir):
        sys.exit(f"{PROG}: --manifests-dir {args.manifests_dir!r} is not a directory")

    if not os.environ.get("AWS_ACCESS_KEY_ID"):
        sys.exit(f"{PROG}: no S3 credentials — put them in ~/.bits/s3keys "
                 "(or set $BITS_AWS_KEYS_FILE / AWS_ACCESS_KEY_ID).")

    # rm pre-flight guards (before touching S3).
    if args.verb == "rm":
        narrowed = any([args.raw, args.orphans, args.broken_links, args.stale_boms,
                        args.arch, args.group,
                        args.package, args.version, args.certified_by, args.build, args.expired,
                        args.bits_version, args.bits_dist, args.built_before])
        if not narrowed and not args.force:
            print(f"{PROG}: refusing to select the whole store for rm. Narrow it with a "
                  "filter (--arch/--package/…), --orphans, --broken-links, or --raw (or --force).",
                  file=sys.stderr)
            return 2
        for pat in (args.raw or []):
            p = pat.strip()
            if not args.force and (p in ("", "*", "/", "TARS", "TARS/", "MANIFESTS", "MANIFESTS/")
                                   or p.startswith("*")):
                print(f"{PROG}: refusing catch-all raw pattern {pat!r} without --force", file=sys.stderr)
                return 2

    s3, bucket = make_s3()
    print(f"# store: {STORE}  (bucket: {bucket})", file=sys.stderr)
    store = Store(s3, bucket)

    if args.verb == "cat":
        return do_cat(s3, bucket, args.key)

    args.marker = None
    if args.verb == "rm":
        # Whether bits-orchestrator owns the certified set: rm then leaves its
        # manifests alone and decertifies stale BOMs rather than removing them.
        try:
            args.marker = orchestrated.read(s3, bucket)
        except orchestrated.Unknown as exc:
            sys.exit(f"{PROG}: {exc}")

    # Resolve the absolute --before cutoff now that S3 is reachable (it may name a
    # manifest whose upload time we look up). Fail CLOSED: an unresolved cutoff
    # must never fall through to "no age filter" and widen a delete.
    if args.before:
        args.before_dt = resolve_before(s3, bucket, args.before)
        if args.before_dt is None:
            sys.exit(f"{PROG}: could not resolve --before {args.before!r} to a time "
                     "(expected ISO-8601, YYYYMMDDTHHMMSSZ, or an existing manifest key/build_id).")
        print(f"# --before {args.before!r} -> {args.before_dt.isoformat()}", file=sys.stderr)

    # Resolve the --built-before cutoff (build provenance, not S3 upload time).
    # Fail CLOSED like --before: an unparseable value must not silently widen a
    # delete. Accept an ISO-8601 timestamp or a YYYY-MM-DD / compact date.
    if args.built_before:
        bb = parse_iso(args.built_before)
        if bb is None:
            for fmt in ("%Y-%m-%d", "%Y%m%d", "%Y%m%dT%H%M%SZ"):
                try:
                    bb = datetime.datetime.strptime(args.built_before, fmt)
                    break
                except ValueError:
                    bb = None
        if bb is None:
            sys.exit(f"{PROG}: could not parse --built-before {args.built_before!r} "
                     "(expected ISO-8601 or YYYY-MM-DD).")
        args.built_before_dt = bb if bb.tzinfo else bb.replace(tzinfo=datetime.timezone.utc)
        print(f"# --built-before {args.built_before!r} -> {args.built_before_dt.isoformat()}",
              file=sys.stderr)

    if args.verb == "verify":
        return do_verify(store, s3, bucket, args)

    if args.stale_boms:
        keys = stale_bom_report(store, s3, bucket, args)
        if args.verb != "rm":
            return 0
        if args.manifests_dir and args.marker:
            return decertify_local(keys, args, args.marker)
        if args.manifests_dir:
            return delete_local_files(keys, args)
        if args.marker and keys:
            print(f"# the certified set ({orchestrated.describe(args.marker)}) still holds what "
                  "these copies certified: `--manifests-dir <a bits-manifests checkout>` asks "
                  "for it to be decertified", file=sys.stderr)
        return do_delete(s3, bucket, keys, store, args)

    # Resolve the object set for the chosen view.
    if args.raw:
        keys, title = select_raw(s3, bucket, args), "raw"
    elif args.broken_links:
        keys, title = select_broken_links(store, s3, bucket, args), "broken-links"
    elif args.orphans:
        keys = select_orphans(store, s3, bucket, args)
        keys.update(orphan_markers(s3, bucket, keys))   # + their dangling rev-index markers
        title = "orphans"
    elif args.objects or args.verb == "rm":
        keys, title = select_objects(store, s3, bucket, args), "objects"
    else:
        keys, title = None, None                      # manifest summary view

    if args.verb == "rm":
        return do_delete(s3, bucket, keys or {}, store, args)

    if keys is None:
        present_manifests(store, args)
    else:
        present_objects(keys, title)
    return 0


if __name__ == "__main__":
    sys.exit(main())
PY