prune: merge exported live-image files into protection (multi-cluster)

ACTIVITY-WP-0020-T07: the kubectl scan only sees the prune host's own
cluster. New repeatable --live-images-file merges image refs exported
from other production clusters; missing file surfaces as WARN (reduced
coverage), forgejo-registry tags protected, other registries ignored.
5/5 tests.

Evidence gathered 2026-07-18: activity-core on railiance01 runs a
locally-imported image (activity-core:railiance01-prod), not a forgejo
registry tag — the largest would_delete set has no live registry
consumer. Live forgejo tags: state-hub:main-1cf949b (railiance01),
issue-core:0.2.1 + state-hub:f2e042a + vergabe-teilnahme:064d295
(coulombcore cluster).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

scripts/forgejo_package_prune.py

@@ -109,6 +109,35 @@ def collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]:
     return protected
 
 
+def collect_live_images_from_files(
+    paths: list[Path],
+) -> tuple[set[tuple[str, str, str]], list[str]]:
+    """Protect image tags listed in exported live-image files.
+
+    Each file holds one image ref per line (`kubectl get pods ... jsonpath`
+    output from another cluster). This closes the multi-cluster gap
+    (ACTIVITY-WP-0020-T07): the prune host's kubectl only sees its own
+    cluster, so every other production cluster exports its live images to a
+    file that is merged here. A missing file is a WARN, not a failure —
+    but it means reduced protection coverage, so the note must surface.
+    """
+    protected: set[tuple[str, str, str]] = set()
+    notes: list[str] = []
+    for raw_path in paths:
+        path = raw_path.expanduser()
+        if not path.is_file():
+            notes.append(f"live-images file missing: {path}")
+            continue
+        for line in path.read_text(encoding="utf-8").splitlines():
+            image = line.strip()
+            if not image or image.startswith("#"):
+                continue
+            match = FORGEJO_IMAGE_RE.match(image)
+            if match and match.group("tag"):
+                protected.add(("container", match.group("name"), match.group("tag")))
+    return protected, notes
+
+
 def collect_live_cluster_versions(
     *, kubectl: str = "kubectl", timeout: float = 60.0
 ) -> tuple[set[tuple[str, str, str]], list[str]]:
@@ -388,6 +417,18 @@ def main(argv: list[str] | None = None) -> int:
         help="Skip protecting image tags currently running in the cluster (kubectl)",
     )
     parser.set_defaults(protect_live=True)
+    parser.add_argument(
+        "--live-images-file",
+        dest="live_images_files",
+        type=Path,
+        action="append",
+        default=[],
+        help=(
+            "File with one image ref per line, exported from another "
+            "production cluster; repeatable. Tags matching the Forgejo "
+            "registry are protected in addition to the local kubectl scan."
+        ),
+    )
     args = parser.parse_args(argv)
 
     apply = bool(args.apply)
@@ -402,6 +443,13 @@ def main(argv: list[str] | None = None) -> int:
         print(f"Protected live cluster tags: {len(live)}", file=sys.stderr)
         for note in protect_notes:
             print(f"  WARN: {note}", file=sys.stderr)
+    if args.live_images_files:
+        file_live, file_notes = collect_live_images_from_files(args.live_images_files)
+        protected |= file_live
+        protect_notes.extend(file_notes)
+        print(f"Protected exported live tags: {len(file_live)}", file=sys.stderr)
+        for note in file_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)

tests/test_forgejo_package_prune.py

@@ -96,3 +96,26 @@ image:
 
 if __name__ == "__main__":
     unittest.main()
+
+def test_collect_live_images_from_files(tmp_path):
+    from scripts.forgejo_package_prune import collect_live_images_from_files
+
+    export = tmp_path / "live-coulombcore.txt"
+    export.write_text(
+        "# exported 2026-07-18\n"
+        "forgejo.coulomb.social/coulomb/issue-core:0.2.1\n"
+        "forgejo.coulomb.social/coulomb/state-hub:f2e042a\n"
+        "gitea.coulomb.social/coulomb/core-hub:3ed8531\n"
+        "nats:2.10-alpine\n"
+        "\n"
+    )
+    protected, notes = collect_live_images_from_files(
+        [export, tmp_path / "missing.txt"]
+    )
+
+    assert ("container", "issue-core", "0.2.1") in protected
+    assert ("container", "state-hub", "f2e042a") in protected
+    # non-forgejo registries and bare images are ignored
+    assert all(name != "core-hub" for (_, name, _) in protected)
+    assert len(protected) == 2
+    assert any("missing.txt" in n for n in notes)