Protect digest-pinned packages and refuse partial-inventory pruning

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e241-8285-7a63-8c0c-51c9cb824dc3

scripts/forgejo_package_prune.py

@@ -33,6 +33,30 @@ FORGEJO_IMAGE_RE = re.compile(
 )
 
 
+def protect_image(image: str, protected: set[tuple[str, str, str]]) -> str | None:
+    """Digest references conservatively protect all versions of their package.
+
+    Package APIs do not prove which tags or child manifests share a live digest.
+    Retaining the whole package avoids deleting live/rollback content through an
+    alias. The additive inventory intentionally keeps this protection until an
+    owner explicitly retires the reference.
+    """
+    ref, separator, digest = image.partition("@")
+    match = FORGEJO_IMAGE_RE.fullmatch(ref)
+    if not match:
+        if image.lower().startswith("forgejo.coulomb.social/"):
+            return "unrecognized Forgejo image reference"
+        return None
+    name = match.group("name")
+    if separator:
+        if not re.fullmatch(r"sha256:[0-9a-f]{64}", digest):
+            return "invalid Forgejo image digest"
+        protected.add(("container", name, "*"))
+    else:
+        protected.add(("container", name, match.group("tag") or "latest"))
+    return None
+
+
 @dataclass(frozen=True)
 class VersionRef:
     package_type: str
@@ -142,9 +166,9 @@ def collect_live_images_from_files(
             if not image or image.startswith("#"):
                 continue
             has_images = True
-            match = FORGEJO_IMAGE_RE.match(image)
-            if match and match.group("tag"):
-                protected.add(("container", match.group("name"), match.group("tag")))
+            error = protect_image(image, protected)
+            if error:
+                notes.append(f"{error} in live-images file: {path}")
         if not has_images:
             notes.append(f"live-images file empty: {path}")
     return protected, notes
@@ -181,14 +205,17 @@ def collect_live_cluster_versions(
         )
     except Exception as exc:  # noqa: BLE001
         return protected, [f"live-tag protection skipped: kubectl query failed ({exc})"]
+    notes: list[str] = []
     for line in result.stdout.splitlines():
         image = line.strip()
         if not image:
             continue
-        match = FORGEJO_IMAGE_RE.match(image)
-        if match and match.group("tag"):
-            protected.add(("container", match.group("name"), match.group("tag")))
-    return protected, []
+        error = protect_image(image, protected)
+        if error:
+            notes.append(error)
+    if not result.stdout.strip():
+        notes.append("live cluster image inventory empty")
+    return protected, notes
 
 
 def _api_request(
@@ -242,7 +269,9 @@ def list_packages(
         )
         url = f"{base_url.rstrip('/')}/api/v1/packages/{owner_q}?{query}"
         payload = _api_request("GET", url, token)
-        batch = payload if isinstance(payload, list) else []
+        if not isinstance(payload, list):
+            raise ValueError("invalid package inventory response")
+        batch = payload
         if not batch:
             break
         items.extend(batch)
@@ -321,7 +350,8 @@ def build_delete_plans(
                 if not version or version in keep:
                     continue
                 key = (package_type, name, version)
-                is_protected = key in protected
+                digest_protected = (package_type, name, "*") in protected
+                is_protected = key in protected or digest_protected
                 plans.append(
                     DeletePlan(
                         package_type=package_type,
@@ -329,7 +359,9 @@ def build_delete_plans(
                         version=version,
                         created_at=str(item.get("created_at") or ""),
                         protected=is_protected,
-                        reason="protected_production_tag" if is_protected else "beyond_retention_depth",
+                        reason=("protected_digest_package" if digest_protected else
+                                "protected_production_tag" if is_protected else
+                                "beyond_retention_depth"),
                     )
                 )
     return plans, errors
@@ -552,6 +584,11 @@ def main(argv: list[str] | None = None) -> int:
         protected=protected,
     )
 
+    # Never partially prune after an incomplete package or requested cluster scan.
+    if apply and (errors or protect_notes):
+        print("Refusing apply: incomplete inventory/protection coverage", file=sys.stderr)
+        return 2
+
     deleted: list[DeletePlan] = []
     for plan in plans:
         if plan.protected: