feat(forgejo-package-prune): protect image tags running live in the cluster

Adds collect_live_cluster_versions() — enumerates pod container images via
kubectl and protects any forgejo.coulomb.social/coulomb/<name>:<tag>. Closes the
gap where CI-deployed apps (e.g. state-hub) pin a live tag absent from Helm
values. On by default (--no-protect-live to skip); emits protection_notes +
live_protection in the JSON summary so consumers can detect reduced coverage.
Verified against production: protects state-hub/vergabe/issue-core live tags.

ACTIVITY-WP-0020 T07 (partial — see workplan note re multi-cluster coverage).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

scripts/forgejo_package_prune.py

@@ -7,6 +7,8 @@ import argparse
 import json
 import os
 import re
+import shutil
+import subprocess
 import sys
 import time
 import urllib.error
@@ -107,6 +109,47 @@ def collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]:
     return protected
 
 
+def collect_live_cluster_versions(
+    *, kubectl: str = "kubectl", timeout: float = 60.0
+) -> tuple[set[tuple[str, str, str]], list[str]]:
+    """Protect image tags currently running in the cluster (best-effort).
+
+    Enumerates all pod container images across namespaces via kubectl and
+    protects any `forgejo.coulomb.social/coulomb/<name>:<tag>`. This closes the
+    gap where a live deployment pins a tag not declared in Helm values (e.g.
+    CI-deployed apps). Failures (no kubectl, no cluster access) return an empty
+    set with a note — pruning a reachable registry must not hard-depend on
+    cluster access, but the note surfaces reduced protection coverage.
+    """
+    protected: set[tuple[str, str, str]] = set()
+    if shutil.which(kubectl) is None:
+        return protected, ["live-tag protection skipped: kubectl not found"]
+    jsonpath = (
+        "{range .items[*]}"
+        "{range .spec.containers[*]}{.image}{'\\n'}{end}"
+        "{range .spec.initContainers[*]}{.image}{'\\n'}{end}"
+        "{end}"
+    )
+    try:
+        result = subprocess.run(
+            [kubectl, "get", "pods", "--all-namespaces", "-o", f"jsonpath={jsonpath}"],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=True,
+        )
+    except Exception as exc:  # noqa: BLE001
+        return protected, [f"live-tag protection skipped: kubectl query failed ({exc})"]
+    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, []
+
+
 def _api_request(
     method: str,
     url: str,
@@ -338,6 +381,13 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--apps-root", type=Path, default=DEFAULT_APPS_ROOT)
     parser.add_argument("--apply", action="store_true")
     parser.add_argument("--json", action="store_true", help="Emit JSON summary on stdout")
+    parser.add_argument(
+        "--no-protect-live",
+        dest="protect_live",
+        action="store_false",
+        help="Skip protecting image tags currently running in the cluster (kubectl)",
+    )
+    parser.set_defaults(protect_live=True)
     args = parser.parse_args(argv)
 
     apply = bool(args.apply)
@@ -345,6 +395,13 @@ def main(argv: list[str] | None = None) -> int:
     package_types = [part.strip() for part in args.types.split(",") if part.strip()]
     token = load_token()
     protected = collect_protected_versions(args.apps_root.expanduser())
+    protect_notes: list[str] = []
+    if args.protect_live:
+        live, protect_notes = collect_live_cluster_versions()
+        protected |= live
+        print(f"Protected live cluster tags: {len(live)}", file=sys.stderr)
+        for note in protect_notes:
+            print(f"  WARN: {note}", file=sys.stderr)
 
     print(f"Forgejo package prune — owner={args.owner} keep={args.max_versions}", file=sys.stderr)
     print(f"Protected production tags: {len(protected)}", file=sys.stderr)
@@ -402,6 +459,8 @@ def main(argv: list[str] | None = None) -> int:
         apply=apply,
         deleted=deleted,
     )
+    summary["live_protection"] = args.protect_live
+    summary["protection_notes"] = protect_notes
 
     if args.json or not sys.stdout.isatty():
         print(json.dumps(summary, indent=2))

tests/test_forgejo_package_prune.py

@@ -64,5 +64,35 @@ image:
         self.assertEqual(protected_plans, {("vergabe-teilnahme", "old-prod")})
 
 
+    def test_collect_live_cluster_versions_protects_forgejo_tags(self) -> None:
+        images = "\n".join([
+            "forgejo.coulomb.social/coulomb/state-hub:f2e042a",
+            "forgejo.coulomb.social/coulomb/vergabe-teilnahme:064d295",
+            "registry.k8s.io/pause:3.9",              # unrelated registry → ignored
+            "gitea.coulomb.social/coulomb/old-app:x",  # non-forgejo → ignored
+            "",
+        ])
+        with mock.patch.object(prune.shutil, "which", return_value="/usr/bin/kubectl"), \
+             mock.patch.object(
+                 prune.subprocess, "run",
+                 return_value=mock.Mock(stdout=images, returncode=0),
+             ):
+            protected, notes = prune.collect_live_cluster_versions()
+        self.assertEqual(notes, [])
+        self.assertEqual(
+            protected,
+            {
+                ("container", "state-hub", "f2e042a"),
+                ("container", "vergabe-teilnahme", "064d295"),
+            },
+        )
+
+    def test_collect_live_cluster_versions_skips_without_kubectl(self) -> None:
+        with mock.patch.object(prune.shutil, "which", return_value=None):
+            protected, notes = prune.collect_live_cluster_versions()
+        self.assertEqual(protected, set())
+        self.assertTrue(notes and "kubectl not found" in notes[0])
+
+
 if __name__ == "__main__":
     unittest.main()