Project pinned digest-safe retention tool through existing GitOps resources

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

k8s/gitops/platform-source.json

@@ -0,0 +1,7 @@
+{
+  "schema_version": 1,
+  "repository": "coulomb/railiance-platform",
+  "revision": "743def17bec7f32600c8340b8e6b520430b88897",
+  "path": "scripts/forgejo_package_prune.py",
+  "sha256": "fa3d4cc8cdcc14ecce509cf6865c6e4726d722f84814b97d40686d68d6ee5bb1"
+}

k8s/gitops/runtime.yaml

@@ -1183,6 +1183,293 @@ data:
         evidence: []
         gaps:
           - "Add explicit ops inventory probes and evidence events."
+  forgejo_package_prune.py: "#!/usr/bin/env python3\n\"\"\"Forgejo package retention\
+    \ prune \u2014 keep newest N versions per package.\"\"\"\n\nfrom __future__ import\
+    \ annotations\n\nimport argparse\nimport json\nimport os\nimport re\nimport shutil\n\
+    import subprocess\nimport sys\nimport time\nimport urllib.error\nimport urllib.parse\n\
+    import urllib.request\nfrom dataclasses import dataclass\nfrom datetime import\
+    \ datetime\nfrom pathlib import Path\nfrom typing import Any\n\nDEFAULT_BASE =\
+    \ \"https://forgejo.coulomb.social\"\nDEFAULT_OWNER = \"coulomb\"\nDEFAULT_TYPES\
+    \ = (\"container\", \"pypi\", \"npm\", \"generic\")\nDEFAULT_MAX_VERSIONS = 3\n\
+    DEFAULT_APPS_ROOT = Path.home() / \"railiance-apps\"\nDEFAULT_FORGEJO_ADMIN_BAO_PATH\
+    \ = \"platform/workloads/forgejo/forgejo-admin\"\nDEFAULT_FORGEJO_ADMIN_BAO_FIELD\
+    \ = \"API_TOKEN\"\nLEGACY_FORGEJO_TOKEN_FILE = Path(\"/tmp/forgejo-tegwick-api-token\"\
+    )\nFORGEJO_IMAGE_RE = re.compile(\n    r\"^forgejo\\.coulomb\\.social/(?:coulomb/)?(?P<name>[^:/]+)(?::(?P<tag>[^/\\\
+    s]+))?$\",\n    re.IGNORECASE,\n)\n\n\ndef protect_image(image: str, protected:\
+    \ set[tuple[str, str, str]]) -> str | None:\n    \"\"\"Digest references conservatively\
+    \ protect all versions of their package.\n\n    Package APIs do not prove which\
+    \ tags or child manifests share a live digest.\n    Retaining the whole package\
+    \ avoids deleting live/rollback content through an\n    alias. The additive inventory\
+    \ intentionally keeps this protection until an\n    owner explicitly retires the\
+    \ reference.\n    \"\"\"\n    ref, separator, digest = image.partition(\"@\")\n\
+    \    match = FORGEJO_IMAGE_RE.fullmatch(ref)\n    if not match:\n        if image.lower().startswith(\"\
+    forgejo.coulomb.social/\"):\n            return \"unrecognized Forgejo image reference\"\
+    \n        return None\n    name = match.group(\"name\")\n    if separator:\n \
+    \       if not re.fullmatch(r\"sha256:[0-9a-f]{64}\", digest):\n            return\
+    \ \"invalid Forgejo image digest\"\n        protected.add((\"container\", name,\
+    \ \"*\"))\n    else:\n        protected.add((\"container\", name, match.group(\"\
+    tag\") or \"latest\"))\n    return None\n\n\n@dataclass(frozen=True)\nclass VersionRef:\n\
+    \    package_type: str\n    name: str\n    version: str\n\n    def key(self) ->\
+    \ tuple[str, str, str]:\n        return (self.package_type, self.name, self.version)\n\
+    \n\n@dataclass(frozen=True)\nclass DeletePlan:\n    package_type: str\n    name:\
+    \ str\n    version: str\n    created_at: str\n    protected: bool\n    reason:\
+    \ str\n\n\ndef _parse_created_at(value: str | None) -> datetime:\n    if not value:\n\
+    \        return datetime.min\n    try:\n        return datetime.fromisoformat(value.replace(\"\
+    Z\", \"+00:00\"))\n    except ValueError:\n        return datetime.min\n\n\ndef\
+    \ collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]:\n\
+    \    protected: set[tuple[str, str, str]] = set()\n    if not apps_root.is_dir():\n\
+    \        return protected\n\n    patterns = [\n        apps_root / \"helm\" /\
+    \ \"*-values.yaml\",\n        apps_root / \"charts\" / \"*\" / \"values.yaml\"\
+    ,\n    ]\n    paths: list[Path] = []\n    for pattern in patterns:\n        paths.extend(sorted(pattern.parent.glob(pattern.name)))\n\
+    \n    try:\n        import yaml  # type: ignore\n    except ImportError:\n   \
+    \     yaml = None\n\n    for path in paths:\n        text = path.read_text(encoding=\"\
+    utf-8\")\n        if yaml is not None:\n            try:\n                data\
+    \ = yaml.safe_load(text) or {}\n            except Exception:\n              \
+    \  data = {}\n            image = data.get(\"image\") if isinstance(data, dict)\
+    \ else None\n            if isinstance(image, dict):\n                repo = str(image.get(\"\
+    repository\") or \"\").strip()\n                tag = str(image.get(\"tag\") or\
+    \ \"\").strip()\n                if repo and tag:\n                    match =\
+    \ FORGEJO_IMAGE_RE.match(repo) or FORGEJO_IMAGE_RE.match(\n                  \
+    \      f\"{repo}:{tag}\"\n                    )\n                    if match:\n\
+    \                        name = match.group(\"name\")\n                      \
+    \  protected.add((\"container\", name, tag))\n            continue\n\n       \
+    \ repo_match = re.search(\n            r\"repository:\\s*forgejo\\.coulomb\\.social/coulomb/([^\\\
+    s]+)\",\n            text,\n            re.IGNORECASE,\n        )\n        tag_match\
+    \ = re.search(r'^\\s*tag:\\s*\"?([^\"\\s#]+)\"?\\s*$', text, re.MULTILINE)\n \
+    \       if repo_match and tag_match:\n            protected.add((\"container\"\
+    , repo_match.group(1), tag_match.group(1)))\n\n    return protected\n\n\ndef collect_live_images_from_files(\n\
+    \    paths: list[Path],\n) -> tuple[set[tuple[str, str, str]], list[str]]:\n \
+    \   \"\"\"Protect image tags listed in exported live-image files.\n\n    Each\
+    \ file holds one image ref per line (`kubectl get pods ... jsonpath`\n    output\
+    \ from another cluster). This closes the multi-cluster gap\n    (ACTIVITY-WP-0020-T07):\
+    \ the prune host's kubectl only sees its own\n    cluster, so every other production\
+    \ cluster exports its live images to a\n    file that is merged here. Unavailable\
+    \ or empty exports produce notes;\n    main refuses apply when any requested export\
+    \ cannot provide coverage.\n    \"\"\"\n    protected: set[tuple[str, str, str]]\
+    \ = set()\n    notes: list[str] = []\n    for raw_path in paths:\n        path\
+    \ = raw_path.expanduser()\n        if not path.is_file():\n            notes.append(f\"\
+    live-images file missing: {path}\")\n            continue\n        try:\n    \
+    \        lines = path.read_text(encoding=\"utf-8\").splitlines()\n        except\
+    \ (OSError, UnicodeError):\n            notes.append(f\"live-images file unreadable:\
+    \ {path}\")\n            continue\n        has_images = False\n        for line\
+    \ in lines:\n            image = line.strip()\n            if not image or image.startswith(\"\
+    #\"):\n                continue\n            has_images = True\n            error\
+    \ = protect_image(image, protected)\n            if error:\n                notes.append(f\"\
+    {error} in live-images file: {path}\")\n        if not has_images:\n         \
+    \   notes.append(f\"live-images file empty: {path}\")\n    return protected, notes\n\
+    \n\ndef collect_live_cluster_versions(\n    *, kubectl: str = \"kubectl\", timeout:\
+    \ float = 60.0\n) -> tuple[set[tuple[str, str, str]], list[str]]:\n    \"\"\"\
+    Protect image tags currently running in the cluster (best-effort).\n\n    Enumerates\
+    \ all pod container images across namespaces via kubectl and\n    protects any\
+    \ `forgejo.coulomb.social/coulomb/<name>:<tag>`. This closes the\n    gap where\
+    \ a live deployment pins a tag not declared in Helm values (e.g.\n    CI-deployed\
+    \ apps). Failures (no kubectl, no cluster access) return an empty\n    set with\
+    \ a note \u2014 pruning a reachable registry must not hard-depend on\n    cluster\
+    \ access, but the note surfaces reduced protection coverage.\n    \"\"\"\n   \
+    \ protected: set[tuple[str, str, str]] = set()\n    if shutil.which(kubectl) is\
+    \ None:\n        return protected, [\"live-tag protection skipped: kubectl not\
+    \ found\"]\n    jsonpath = (\n        \"{range .items[*]}\"\n        \"{range\
+    \ .spec.containers[*]}{.image}{'\\\\n'}{end}\"\n        \"{range .spec.initContainers[*]}{.image}{'\\\
+    \\n'}{end}\"\n        \"{end}\"\n    )\n    try:\n        result = subprocess.run(\n\
+    \            [kubectl, \"get\", \"pods\", \"--all-namespaces\", \"-o\", f\"jsonpath={jsonpath}\"\
+    ],\n            capture_output=True,\n            text=True,\n            timeout=timeout,\n\
+    \            check=True,\n        )\n    except Exception as exc:  # noqa: BLE001\n\
+    \        return protected, [f\"live-tag protection skipped: kubectl query failed\
+    \ ({exc})\"]\n    notes: list[str] = []\n    for line in result.stdout.splitlines():\n\
+    \        image = line.strip()\n        if not image:\n            continue\n \
+    \       error = protect_image(image, protected)\n        if error:\n         \
+    \   notes.append(error)\n    if not result.stdout.strip():\n        notes.append(\"\
+    live cluster image inventory empty\")\n    return protected, notes\n\n\ndef _api_request(\n\
+    \    method: str,\n    url: str,\n    token: str,\n    *,\n    timeout: float\
+    \ = 60.0,\n    retries: int = 2,\n) -> Any:\n    req = urllib.request.Request(\n\
+    \        url,\n        method=method,\n        headers={\n            \"Authorization\"\
+    : f\"token {token}\",\n            \"Accept\": \"application/json\",\n       \
+    \ },\n    )\n    last_exc: Exception | None = None\n    for attempt in range(retries\
+    \ + 1):\n        try:\n            with urllib.request.urlopen(req, timeout=timeout)\
+    \ as resp:\n                body = resp.read().decode(\"utf-8\")\n           \
+    \     return json.loads(body) if body else None\n        except urllib.error.HTTPError:\n\
+    \            raise  # 4xx/5xx are real responses \u2014 surface them, do not retry\n\
+    \        except (urllib.error.URLError, TimeoutError, OSError) as exc:\n     \
+    \       # Transient: connection reset, read timeout, TLS handshake timeout.\n\
+    \            last_exc = exc\n            if attempt < retries:\n             \
+    \   time.sleep(2 * (attempt + 1))\n                continue\n            raise\n\
+    \    if last_exc:  # pragma: no cover - defensive\n        raise last_exc\n\n\n\
+    def list_packages(\n    base_url: str,\n    token: str,\n    owner: str,\n   \
+    \ package_type: str,\n) -> list[dict[str, Any]]:\n    owner_q = urllib.parse.quote(owner)\n\
+    \    items: list[dict[str, Any]] = []\n    page = 1\n    page_size = 50\n    while\
+    \ True:\n        query = urllib.parse.urlencode(\n            {\"limit\": page_size,\
+    \ \"page\": page, \"type\": package_type}\n        )\n        url = f\"{base_url.rstrip('/')}/api/v1/packages/{owner_q}?{query}\"\
+    \n        payload = _api_request(\"GET\", url, token)\n        if not isinstance(payload,\
+    \ list):\n            raise ValueError(\"invalid package inventory response\"\
+    )\n        batch = payload\n        if not batch:\n            break\n       \
+    \ items.extend(batch)\n        if len(batch) < page_size:\n            break\n\
+    \        page += 1\n    return items\n\n\ndef delete_version(\n    base_url: str,\n\
+    \    token: str,\n    owner: str,\n    package_type: str,\n    name: str,\n  \
+    \  version: str,\n    *,\n    dry_run: bool,\n) -> None:\n    owner_q = urllib.parse.quote(owner)\n\
+    \    type_q = urllib.parse.quote(package_type)\n    name_q = urllib.parse.quote(name,\
+    \ safe=\"\")\n    version_q = urllib.parse.quote(version, safe=\"\")\n    path\
+    \ = f\"/api/v1/packages/{owner_q}/{type_q}/{name_q}/{version_q}\"\n    if dry_run:\n\
+    \        return\n    url = f\"{base_url.rstrip('/')}{path}\"\n    _api_request(\"\
+    DELETE\", url, token)\n\n\ndef build_delete_plans(\n    *,\n    base_url: str,\n\
+    \    token: str,\n    owner: str,\n    package_types: list[str],\n    max_versions:\
+    \ int,\n    protected: set[tuple[str, str, str]],\n) -> tuple[list[DeletePlan],\
+    \ list[str]]:\n    plans: list[DeletePlan] = []\n    errors: list[str] = []\n\n\
+    \    for package_type in package_types:\n        try:\n            packages =\
+    \ list_packages(base_url, token, owner, package_type)\n        except urllib.error.HTTPError\
+    \ as exc:\n            errors.append(f\"list {package_type}: HTTP {exc.code}\"\
+    )\n            continue\n        except Exception as exc:  # noqa: BLE001\n  \
+    \          errors.append(f\"list {package_type}: {exc}\")\n            continue\n\
+    \n        # Forgejo's package list endpoint returns one entry per (name, version).\n\
+    \        # Group by package name; each group is that package's version set \u2014\
+    \ there\n        # is no separate per-package \"/versions\" endpoint.\n      \
+    \  by_name: dict[str, list[dict[str, Any]]] = {}\n        for package in packages:\n\
+    \            name = str(package.get(\"name\") or package.get(\"package_name\"\
+    ) or \"\")\n            if not name:\n                continue\n            by_name.setdefault(name,\
+    \ []).append(package)\n\n        for name, versions in by_name.items():\n    \
+    \        sorted_versions = sorted(\n                versions,\n              \
+    \  key=lambda item: _parse_created_at(str(item.get(\"created_at\") or \"\")),\n\
+    \                reverse=True,\n            )\n            keep = {\n        \
+    \        str(item.get(\"version\") or \"\")\n                for item in sorted_versions[:max_versions]\n\
+    \                if str(item.get(\"version\") or \"\")\n            }\n      \
+    \      for item in sorted_versions[max_versions:]:\n                version =\
+    \ str(item.get(\"version\") or \"\")\n                if not version or version\
+    \ in keep:\n                    continue\n                key = (package_type,\
+    \ name, version)\n                digest_protected = (package_type, name, \"*\"\
+    ) in protected\n                is_protected = key in protected or digest_protected\n\
+    \                plans.append(\n                    DeletePlan(\n            \
+    \            package_type=package_type,\n                        name=name,\n\
+    \                        version=version,\n                        created_at=str(item.get(\"\
+    created_at\") or \"\"),\n                        protected=is_protected,\n   \
+    \                     reason=(\"protected_digest_package\" if digest_protected\
+    \ else\n                                \"protected_production_tag\" if is_protected\
+    \ else\n                                \"beyond_retention_depth\"),\n       \
+    \             )\n                )\n    return plans, errors\n\n\ndef _read_token_file(path:\
+    \ Path) -> str:\n    return path.read_text(encoding=\"utf-8\").strip()\n\n\ndef\
+    \ _truthy_env(name: str) -> bool:\n    return os.environ.get(name, \"\").strip().lower()\
+    \ in {\"1\", \"true\", \"yes\", \"on\"}\n\n\ndef _load_token_from_file_env() ->\
+    \ str | None:\n    for env_name in (\"FORGEJO_TOKEN_FILE\", \"FORGEJO_ADMIN_TOKEN_FILE\"\
+    ):\n        raw_path = os.environ.get(env_name, \"\").strip()\n        if not\
+    \ raw_path:\n            continue\n        path = Path(raw_path).expanduser()\n\
+    \        if not path.is_file():\n            raise SystemExit(f\"ERROR: {env_name}\
+    \ points to a missing file: {path}\")\n        token = _read_token_file(path)\n\
+    \        if token:\n            return token\n        raise SystemExit(f\"ERROR:\
+    \ {env_name} points to an empty file: {path}\")\n    return None\n\n\ndef _load_token_from_openbao()\
+    \ -> tuple[str | None, str | None]:\n    bao_bin = os.environ.get(\"FORGEJO_ADMIN_BAO_CLI\"\
+    , \"bao\").strip() or \"bao\"\n    bao_path = (\n        os.environ.get(\"FORGEJO_ADMIN_BAO_PATH\"\
+    , DEFAULT_FORGEJO_ADMIN_BAO_PATH).strip()\n        or DEFAULT_FORGEJO_ADMIN_BAO_PATH\n\
+    \    )\n    bao_field = (\n        os.environ.get(\"FORGEJO_ADMIN_BAO_FIELD\"\
+    , DEFAULT_FORGEJO_ADMIN_BAO_FIELD).strip()\n        or DEFAULT_FORGEJO_ADMIN_BAO_FIELD\n\
+    \    )\n    if shutil.which(bao_bin) is None:\n        return None, f\"{bao_bin}\
+    \ CLI not found\"\n    try:\n        result = subprocess.run(\n            [bao_bin,\
+    \ \"kv\", \"get\", f\"-field={bao_field}\", bao_path],\n            capture_output=True,\n\
+    \            text=True,\n            check=True,\n        )\n    except subprocess.CalledProcessError\
+    \ as exc:\n        detail = exc.stderr.strip() or exc.stdout.strip() or f\"exit\
+    \ {exc.returncode}\"\n        return None, f\"{bao_bin} kv get failed: {detail}\"\
+    \n    except OSError as exc:\n        return None, f\"{bao_bin} invocation failed:\
+    \ {exc}\"\n    token = result.stdout.strip()\n    if not token:\n        return\
+    \ None, f\"{bao_bin} kv get returned an empty {bao_field} field\"\n    return\
+    \ token, None\n\n\ndef _token_help_message(bao_error: str | None) -> str:\n  \
+    \  lines = [\n        \"ERROR: Forgejo API token required (read:package + write:package).\"\
+    ,\n        \"  Primary: bao login -method=oidc -path=netkingdom role=forgejo-admin-workload-kv-read\"\
+    ,\n        f\"  Default lane: {DEFAULT_FORGEJO_ADMIN_BAO_PATH} field {DEFAULT_FORGEJO_ADMIN_BAO_FIELD}\"\
+    ,\n        \"  Override lane via FORGEJO_ADMIN_BAO_PATH / FORGEJO_ADMIN_BAO_FIELD\
+    \ if needed.\",\n        \"  Break-glass: set FORGEJO_TOKEN / FORGEJO_ADMIN_TOKEN,\
+    \ or set FORGEJO_TOKEN_FILE / FORGEJO_ADMIN_TOKEN_FILE.\",\n        \"  Legacy\
+    \ /tmp fallback stays opt-in only via FORGEJO_ALLOW_LEGACY_FILE_FALLBACK=1.\"\
+    ,\n        \"  See: railiance-platform/docs/forgejo-package-prune.md\",\n    ]\n\
+    \    if bao_error:\n        lines.insert(3, f\"  OpenBao lookup failed: {bao_error}\"\
+    )\n    return \"\\n\".join(lines)\n\n\ndef load_token() -> str:\n    for env_name\
+    \ in (\"FORGEJO_TOKEN\", \"FORGEJO_ADMIN_TOKEN\"):\n        token = os.environ.get(env_name,\
+    \ \"\").strip()\n        if token:\n            return token\n    token = _load_token_from_file_env()\n\
+    \    if token:\n        return token\n    token, bao_error = _load_token_from_openbao()\n\
+    \    if token:\n        return token\n\n    if _truthy_env(\"FORGEJO_ALLOW_LEGACY_FILE_FALLBACK\"\
+    ):\n        if LEGACY_FORGEJO_TOKEN_FILE.is_file():\n            token = _read_token_file(LEGACY_FORGEJO_TOKEN_FILE)\n\
+    \            if token:\n                return token\n            suffix = f\"\
+    legacy file {LEGACY_FORGEJO_TOKEN_FILE} is empty\"\n        else:\n          \
+    \  suffix = f\"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is missing\"\n        bao_error\
+    \ = f\"{bao_error}; {suffix}\" if bao_error else suffix\n\n    raise SystemExit(_token_help_message(bao_error))\n\
+    \n\ndef emit_summary(\n    *,\n    owner: str,\n    max_versions: int,\n    package_types:\
+    \ list[str],\n    protected: set[tuple[str, str, str]],\n    plans: list[DeletePlan],\n\
+    \    errors: list[str],\n    apply: bool,\n    deleted: list[DeletePlan],\n) ->\
+    \ dict[str, Any]:\n    would_delete = [p for p in plans if not p.protected]\n\
+    \    skipped_protected = [p for p in plans if p.protected]\n    return {\n   \
+    \     \"kind\": \"forgejo_package_prune\",\n        \"owner\": owner,\n      \
+    \  \"max_versions\": max_versions,\n        \"package_types\": package_types,\n\
+    \        \"protected_count\": len(protected),\n        \"candidate_count\": len(would_delete),\n\
+    \        \"skipped_protected_count\": len(skipped_protected),\n        \"deleted_count\"\
+    : len(deleted),\n        \"apply\": apply,\n        \"would_delete\": [\n    \
+    \        {\n                \"type\": p.package_type,\n                \"name\"\
+    : p.name,\n                \"version\": p.version,\n                \"created_at\"\
+    : p.created_at,\n            }\n            for p in would_delete\n        ],\n\
+    \        \"skipped_protected\": [\n            {\n                \"type\": p.package_type,\n\
+    \                \"name\": p.name,\n                \"version\": p.version,\n\
+    \                \"reason\": p.reason,\n            }\n            for p in skipped_protected\n\
+    \        ],\n        \"deleted\": [\n            {\n                \"type\":\
+    \ p.package_type,\n                \"name\": p.name,\n                \"version\"\
+    : p.version,\n            }\n            for p in deleted\n        ],\n      \
+    \  \"errors\": errors,\n    }\n\n\ndef main(argv: list[str] | None = None) ->\
+    \ int:\n    parser = argparse.ArgumentParser(description=\"Prune old Forgejo package\
+    \ versions\")\n    parser.add_argument(\"--owner\", default=DEFAULT_OWNER)\n \
+    \   parser.add_argument(\"--base-url\", default=os.environ.get(\"FORGEJO_BASE_URL\"\
+    , DEFAULT_BASE))\n    parser.add_argument(\"--max-versions\", type=int, default=DEFAULT_MAX_VERSIONS)\n\
+    \    parser.add_argument(\n        \"--types\",\n        default=\",\".join(DEFAULT_TYPES),\n\
+    \        help=\"Comma-separated package types\",\n    )\n    parser.add_argument(\"\
+    --apps-root\", type=Path, default=DEFAULT_APPS_ROOT)\n    parser.add_argument(\"\
+    --apply\", action=\"store_true\")\n    parser.add_argument(\"--json\", action=\"\
+    store_true\", help=\"Emit JSON summary on stdout\")\n    parser.add_argument(\n\
+    \        \"--no-protect-live\",\n        dest=\"protect_live\",\n        action=\"\
+    store_false\",\n        help=\"Skip protecting image tags currently running in\
+    \ the cluster (kubectl)\",\n    )\n    parser.set_defaults(protect_live=True)\n\
+    \    parser.add_argument(\n        \"--live-images-file\",\n        dest=\"live_images_files\"\
+    ,\n        type=Path,\n        action=\"append\",\n        default=[],\n     \
+    \   help=(\n            \"File with one image ref per line, exported from another\
+    \ \"\n            \"production cluster; repeatable. Tags matching the Forgejo\
+    \ \"\n            \"registry are protected in addition to the local kubectl scan.\"\
+    \n        ),\n    )\n    args = parser.parse_args(argv)\n\n    apply = bool(args.apply)\n\
+    \    dry_run = not apply\n    package_types = [part.strip() for part in args.types.split(\"\
+    ,\") if part.strip()]\n    file_live, file_notes = collect_live_images_from_files(args.live_images_files)\n\
+    \    if apply and file_notes:\n        for note in file_notes:\n            print(f\"\
+    \  ERROR: {note}\", file=sys.stderr)\n        print(\"Refusing apply: requested\
+    \ live-image inventory is unavailable or empty\", file=sys.stderr)\n        return\
+    \ 2\n    token = load_token()\n    protected = collect_protected_versions(args.apps_root.expanduser())\n\
+    \    protect_notes: list[str] = []\n    if args.protect_live:\n        live, protect_notes\
+    \ = collect_live_cluster_versions()\n        protected |= live\n        print(f\"\
+    Protected live cluster tags: {len(live)}\", file=sys.stderr)\n        for note\
+    \ in protect_notes:\n            print(f\"  WARN: {note}\", file=sys.stderr)\n\
+    \    if args.live_images_files:\n        protected |= file_live\n        protect_notes.extend(file_notes)\n\
+    \        print(f\"Protected exported live tags: {len(file_live)}\", file=sys.stderr)\n\
+    \        for note in file_notes:\n            print(f\"  WARN: {note}\", file=sys.stderr)\n\
+    \n    print(f\"Forgejo package prune \u2014 owner={args.owner} keep={args.max_versions}\"\
+    , file=sys.stderr)\n    print(f\"Protected production tags: {len(protected)}\"\
+    , file=sys.stderr)\n\n    plans, errors = build_delete_plans(\n        base_url=args.base_url,\n\
+    \        token=token,\n        owner=args.owner,\n        package_types=package_types,\n\
+    \        max_versions=max(1, args.max_versions),\n        protected=protected,\n\
+    \    )\n\n    # Never partially prune after an incomplete package or requested\
+    \ cluster scan.\n    if apply and (errors or protect_notes):\n        print(\"\
+    Refusing apply: incomplete inventory/protection coverage\", file=sys.stderr)\n\
+    \        return 2\n\n    deleted: list[DeletePlan] = []\n    for plan in plans:\n\
+    \        if plan.protected:\n            print(\n                f\"  skip protected\
+    \  {plan.package_type}/{plan.name}:{plan.version}\",\n                file=sys.stderr,\n\
+    \            )\n            continue\n        if dry_run:\n            print(\n\
+    \                f\"  would delete    {plan.package_type}/{plan.name}:{plan.version}\"\
+    ,\n                file=sys.stderr,\n            )\n            continue\n   \
+    \     try:\n            delete_version(\n                args.base_url,\n    \
+    \            token,\n                args.owner,\n                plan.package_type,\n\
+    \                plan.name,\n                plan.version,\n                dry_run=False,\n\
+    \            )\n            deleted.append(plan)\n            print(\n       \
+    \         f\"  deleted         {plan.package_type}/{plan.name}:{plan.version}\"\
+    ,\n                file=sys.stderr,\n            )\n        except urllib.error.HTTPError\
+    \ as exc:\n            errors.append(f\"delete {plan.package_type}/{plan.name}:{plan.version}:\
+    \ HTTP {exc.code}\")\n        except Exception as exc:  # noqa: BLE001\n     \
+    \       errors.append(f\"delete {plan.package_type}/{plan.name}:{plan.version}:\
+    \ {exc}\")\n\n    summary = emit_summary(\n        owner=args.owner,\n       \
+    \ max_versions=args.max_versions,\n        package_types=package_types,\n    \
+    \    protected=protected,\n        plans=plans,\n        errors=errors,\n    \
+    \    apply=apply,\n        deleted=deleted,\n    )\n    summary[\"live_protection\"\
+    ] = args.protect_live\n    summary[\"protection_notes\"] = protect_notes\n\n \
+    \   if args.json or not sys.stdout.isatty():\n        print(json.dumps(summary,\
+    \ indent=2))\n    else:\n        print(json.dumps(summary, indent=2))\n\n    return\
+    \ 1 if errors else 0\n\n\nif __name__ == \"__main__\":\n    raise SystemExit(main())\n"
 kind: ConfigMap
 metadata:
   name: actcore-ops-service-inventory
@@ -1343,6 +1630,7 @@ spec:
   template:
     metadata:
       annotations:
+        activity-core/retention-sha256: fa3d4cc8cdcc14ecce509cf6865c6e4726d722f84814b97d40686d68d6ee5bb1
         kubectl.kubernetes.io/restartedAt: '2026-09-05T20:48:57+02:00'
       labels:
         app.kubernetes.io/name: actcore-worker
@@ -1397,6 +1685,10 @@ spec:
         - mountPath: /var/custodian/runtime/prompts
           name: custodian-runtime
           readOnly: true
+        - mountPath: /opt/railiance-platform/scripts/forgejo_package_prune.py
+          name: ops-service-inventory
+          subPath: forgejo_package_prune.py
+          readOnly: true
         - mountPath: /opt/railiance-platform
           name: railiance-platform
           readOnly: true

k8s/railiance/20-runtime.yaml

@@ -1079,6 +1079,655 @@ data:
         evidence: []
         gaps:
           - "Add explicit ops inventory probes and evidence events."
+  forgejo_package_prune.py: |
+    #!/usr/bin/env python3
+    """Forgejo package retention prune — keep newest N versions per package."""
+
+    from __future__ import annotations
+
+    import argparse
+    import json
+    import os
+    import re
+    import shutil
+    import subprocess
+    import sys
+    import time
+    import urllib.error
+    import urllib.parse
+    import urllib.request
+    from dataclasses import dataclass
+    from datetime import datetime
+    from pathlib import Path
+    from typing import Any
+
+    DEFAULT_BASE = "https://forgejo.coulomb.social"
+    DEFAULT_OWNER = "coulomb"
+    DEFAULT_TYPES = ("container", "pypi", "npm", "generic")
+    DEFAULT_MAX_VERSIONS = 3
+    DEFAULT_APPS_ROOT = Path.home() / "railiance-apps"
+    DEFAULT_FORGEJO_ADMIN_BAO_PATH = "platform/workloads/forgejo/forgejo-admin"
+    DEFAULT_FORGEJO_ADMIN_BAO_FIELD = "API_TOKEN"
+    LEGACY_FORGEJO_TOKEN_FILE = Path("/tmp/forgejo-tegwick-api-token")
+    FORGEJO_IMAGE_RE = re.compile(
+        r"^forgejo\.coulomb\.social/(?:coulomb/)?(?P<name>[^:/]+)(?::(?P<tag>[^/\s]+))?$",
+        re.IGNORECASE,
+    )
+
+
+    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
+        name: str
+        version: str
+
+        def key(self) -> tuple[str, str, str]:
+            return (self.package_type, self.name, self.version)
+
+
+    @dataclass(frozen=True)
+    class DeletePlan:
+        package_type: str
+        name: str
+        version: str
+        created_at: str
+        protected: bool
+        reason: str
+
+
+    def _parse_created_at(value: str | None) -> datetime:
+        if not value:
+            return datetime.min
+        try:
+            return datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return datetime.min
+
+
+    def collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]:
+        protected: set[tuple[str, str, str]] = set()
+        if not apps_root.is_dir():
+            return protected
+
+        patterns = [
+            apps_root / "helm" / "*-values.yaml",
+            apps_root / "charts" / "*" / "values.yaml",
+        ]
+        paths: list[Path] = []
+        for pattern in patterns:
+            paths.extend(sorted(pattern.parent.glob(pattern.name)))
+
+        try:
+            import yaml  # type: ignore
+        except ImportError:
+            yaml = None
+
+        for path in paths:
+            text = path.read_text(encoding="utf-8")
+            if yaml is not None:
+                try:
+                    data = yaml.safe_load(text) or {}
+                except Exception:
+                    data = {}
+                image = data.get("image") if isinstance(data, dict) else None
+                if isinstance(image, dict):
+                    repo = str(image.get("repository") or "").strip()
+                    tag = str(image.get("tag") or "").strip()
+                    if repo and tag:
+                        match = FORGEJO_IMAGE_RE.match(repo) or FORGEJO_IMAGE_RE.match(
+                            f"{repo}:{tag}"
+                        )
+                        if match:
+                            name = match.group("name")
+                            protected.add(("container", name, tag))
+                continue
+
+            repo_match = re.search(
+                r"repository:\s*forgejo\.coulomb\.social/coulomb/([^\s]+)",
+                text,
+                re.IGNORECASE,
+            )
+            tag_match = re.search(r'^\s*tag:\s*"?([^"\s#]+)"?\s*$', text, re.MULTILINE)
+            if repo_match and tag_match:
+                protected.add(("container", repo_match.group(1), tag_match.group(1)))
+
+        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. Unavailable or empty exports produce notes;
+        main refuses apply when any requested export cannot provide coverage.
+        """
+        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
+            try:
+                lines = path.read_text(encoding="utf-8").splitlines()
+            except (OSError, UnicodeError):
+                notes.append(f"live-images file unreadable: {path}")
+                continue
+            has_images = False
+            for line in lines:
+                image = line.strip()
+                if not image or image.startswith("#"):
+                    continue
+                has_images = True
+                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
+
+
+    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})"]
+        notes: list[str] = []
+        for line in result.stdout.splitlines():
+            image = line.strip()
+            if not image:
+                continue
+            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(
+        method: str,
+        url: str,
+        token: str,
+        *,
+        timeout: float = 60.0,
+        retries: int = 2,
+    ) -> Any:
+        req = urllib.request.Request(
+            url,
+            method=method,
+            headers={
+                "Authorization": f"token {token}",
+                "Accept": "application/json",
+            },
+        )
+        last_exc: Exception | None = None
+        for attempt in range(retries + 1):
+            try:
+                with urllib.request.urlopen(req, timeout=timeout) as resp:
+                    body = resp.read().decode("utf-8")
+                    return json.loads(body) if body else None
+            except urllib.error.HTTPError:
+                raise  # 4xx/5xx are real responses — surface them, do not retry
+            except (urllib.error.URLError, TimeoutError, OSError) as exc:
+                # Transient: connection reset, read timeout, TLS handshake timeout.
+                last_exc = exc
+                if attempt < retries:
+                    time.sleep(2 * (attempt + 1))
+                    continue
+                raise
+        if last_exc:  # pragma: no cover - defensive
+            raise last_exc
+
+
+    def list_packages(
+        base_url: str,
+        token: str,
+        owner: str,
+        package_type: str,
+    ) -> list[dict[str, Any]]:
+        owner_q = urllib.parse.quote(owner)
+        items: list[dict[str, Any]] = []
+        page = 1
+        page_size = 50
+        while True:
+            query = urllib.parse.urlencode(
+                {"limit": page_size, "page": page, "type": package_type}
+            )
+            url = f"{base_url.rstrip('/')}/api/v1/packages/{owner_q}?{query}"
+            payload = _api_request("GET", url, token)
+            if not isinstance(payload, list):
+                raise ValueError("invalid package inventory response")
+            batch = payload
+            if not batch:
+                break
+            items.extend(batch)
+            if len(batch) < page_size:
+                break
+            page += 1
+        return items
+
+
+    def delete_version(
+        base_url: str,
+        token: str,
+        owner: str,
+        package_type: str,
+        name: str,
+        version: str,
+        *,
+        dry_run: bool,
+    ) -> None:
+        owner_q = urllib.parse.quote(owner)
+        type_q = urllib.parse.quote(package_type)
+        name_q = urllib.parse.quote(name, safe="")
+        version_q = urllib.parse.quote(version, safe="")
+        path = f"/api/v1/packages/{owner_q}/{type_q}/{name_q}/{version_q}"
+        if dry_run:
+            return
+        url = f"{base_url.rstrip('/')}{path}"
+        _api_request("DELETE", url, token)
+
+
+    def build_delete_plans(
+        *,
+        base_url: str,
+        token: str,
+        owner: str,
+        package_types: list[str],
+        max_versions: int,
+        protected: set[tuple[str, str, str]],
+    ) -> tuple[list[DeletePlan], list[str]]:
+        plans: list[DeletePlan] = []
+        errors: list[str] = []
+
+        for package_type in package_types:
+            try:
+                packages = list_packages(base_url, token, owner, package_type)
+            except urllib.error.HTTPError as exc:
+                errors.append(f"list {package_type}: HTTP {exc.code}")
+                continue
+            except Exception as exc:  # noqa: BLE001
+                errors.append(f"list {package_type}: {exc}")
+                continue
+
+            # Forgejo's package list endpoint returns one entry per (name, version).
+            # Group by package name; each group is that package's version set — there
+            # is no separate per-package "/versions" endpoint.
+            by_name: dict[str, list[dict[str, Any]]] = {}
+            for package in packages:
+                name = str(package.get("name") or package.get("package_name") or "")
+                if not name:
+                    continue
+                by_name.setdefault(name, []).append(package)
+
+            for name, versions in by_name.items():
+                sorted_versions = sorted(
+                    versions,
+                    key=lambda item: _parse_created_at(str(item.get("created_at") or "")),
+                    reverse=True,
+                )
+                keep = {
+                    str(item.get("version") or "")
+                    for item in sorted_versions[:max_versions]
+                    if str(item.get("version") or "")
+                }
+                for item in sorted_versions[max_versions:]:
+                    version = str(item.get("version") or "")
+                    if not version or version in keep:
+                        continue
+                    key = (package_type, name, version)
+                    digest_protected = (package_type, name, "*") in protected
+                    is_protected = key in protected or digest_protected
+                    plans.append(
+                        DeletePlan(
+                            package_type=package_type,
+                            name=name,
+                            version=version,
+                            created_at=str(item.get("created_at") or ""),
+                            protected=is_protected,
+                            reason=("protected_digest_package" if digest_protected else
+                                    "protected_production_tag" if is_protected else
+                                    "beyond_retention_depth"),
+                        )
+                    )
+        return plans, errors
+
+
+    def _read_token_file(path: Path) -> str:
+        return path.read_text(encoding="utf-8").strip()
+
+
+    def _truthy_env(name: str) -> bool:
+        return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+    def _load_token_from_file_env() -> str | None:
+        for env_name in ("FORGEJO_TOKEN_FILE", "FORGEJO_ADMIN_TOKEN_FILE"):
+            raw_path = os.environ.get(env_name, "").strip()
+            if not raw_path:
+                continue
+            path = Path(raw_path).expanduser()
+            if not path.is_file():
+                raise SystemExit(f"ERROR: {env_name} points to a missing file: {path}")
+            token = _read_token_file(path)
+            if token:
+                return token
+            raise SystemExit(f"ERROR: {env_name} points to an empty file: {path}")
+        return None
+
+
+    def _load_token_from_openbao() -> tuple[str | None, str | None]:
+        bao_bin = os.environ.get("FORGEJO_ADMIN_BAO_CLI", "bao").strip() or "bao"
+        bao_path = (
+            os.environ.get("FORGEJO_ADMIN_BAO_PATH", DEFAULT_FORGEJO_ADMIN_BAO_PATH).strip()
+            or DEFAULT_FORGEJO_ADMIN_BAO_PATH
+        )
+        bao_field = (
+            os.environ.get("FORGEJO_ADMIN_BAO_FIELD", DEFAULT_FORGEJO_ADMIN_BAO_FIELD).strip()
+            or DEFAULT_FORGEJO_ADMIN_BAO_FIELD
+        )
+        if shutil.which(bao_bin) is None:
+            return None, f"{bao_bin} CLI not found"
+        try:
+            result = subprocess.run(
+                [bao_bin, "kv", "get", f"-field={bao_field}", bao_path],
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+        except subprocess.CalledProcessError as exc:
+            detail = exc.stderr.strip() or exc.stdout.strip() or f"exit {exc.returncode}"
+            return None, f"{bao_bin} kv get failed: {detail}"
+        except OSError as exc:
+            return None, f"{bao_bin} invocation failed: {exc}"
+        token = result.stdout.strip()
+        if not token:
+            return None, f"{bao_bin} kv get returned an empty {bao_field} field"
+        return token, None
+
+
+    def _token_help_message(bao_error: str | None) -> str:
+        lines = [
+            "ERROR: Forgejo API token required (read:package + write:package).",
+            "  Primary: bao login -method=oidc -path=netkingdom role=forgejo-admin-workload-kv-read",
+            f"  Default lane: {DEFAULT_FORGEJO_ADMIN_BAO_PATH} field {DEFAULT_FORGEJO_ADMIN_BAO_FIELD}",
+            "  Override lane via FORGEJO_ADMIN_BAO_PATH / FORGEJO_ADMIN_BAO_FIELD if needed.",
+            "  Break-glass: set FORGEJO_TOKEN / FORGEJO_ADMIN_TOKEN, or set FORGEJO_TOKEN_FILE / FORGEJO_ADMIN_TOKEN_FILE.",
+            "  Legacy /tmp fallback stays opt-in only via FORGEJO_ALLOW_LEGACY_FILE_FALLBACK=1.",
+            "  See: railiance-platform/docs/forgejo-package-prune.md",
+        ]
+        if bao_error:
+            lines.insert(3, f"  OpenBao lookup failed: {bao_error}")
+        return "\n".join(lines)
+
+
+    def load_token() -> str:
+        for env_name in ("FORGEJO_TOKEN", "FORGEJO_ADMIN_TOKEN"):
+            token = os.environ.get(env_name, "").strip()
+            if token:
+                return token
+        token = _load_token_from_file_env()
+        if token:
+            return token
+        token, bao_error = _load_token_from_openbao()
+        if token:
+            return token
+
+        if _truthy_env("FORGEJO_ALLOW_LEGACY_FILE_FALLBACK"):
+            if LEGACY_FORGEJO_TOKEN_FILE.is_file():
+                token = _read_token_file(LEGACY_FORGEJO_TOKEN_FILE)
+                if token:
+                    return token
+                suffix = f"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is empty"
+            else:
+                suffix = f"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is missing"
+            bao_error = f"{bao_error}; {suffix}" if bao_error else suffix
+
+        raise SystemExit(_token_help_message(bao_error))
+
+
+    def emit_summary(
+        *,
+        owner: str,
+        max_versions: int,
+        package_types: list[str],
+        protected: set[tuple[str, str, str]],
+        plans: list[DeletePlan],
+        errors: list[str],
+        apply: bool,
+        deleted: list[DeletePlan],
+    ) -> dict[str, Any]:
+        would_delete = [p for p in plans if not p.protected]
+        skipped_protected = [p for p in plans if p.protected]
+        return {
+            "kind": "forgejo_package_prune",
+            "owner": owner,
+            "max_versions": max_versions,
+            "package_types": package_types,
+            "protected_count": len(protected),
+            "candidate_count": len(would_delete),
+            "skipped_protected_count": len(skipped_protected),
+            "deleted_count": len(deleted),
+            "apply": apply,
+            "would_delete": [
+                {
+                    "type": p.package_type,
+                    "name": p.name,
+                    "version": p.version,
+                    "created_at": p.created_at,
+                }
+                for p in would_delete
+            ],
+            "skipped_protected": [
+                {
+                    "type": p.package_type,
+                    "name": p.name,
+                    "version": p.version,
+                    "reason": p.reason,
+                }
+                for p in skipped_protected
+            ],
+            "deleted": [
+                {
+                    "type": p.package_type,
+                    "name": p.name,
+                    "version": p.version,
+                }
+                for p in deleted
+            ],
+            "errors": errors,
+        }
+
+
+    def main(argv: list[str] | None = None) -> int:
+        parser = argparse.ArgumentParser(description="Prune old Forgejo package versions")
+        parser.add_argument("--owner", default=DEFAULT_OWNER)
+        parser.add_argument("--base-url", default=os.environ.get("FORGEJO_BASE_URL", DEFAULT_BASE))
+        parser.add_argument("--max-versions", type=int, default=DEFAULT_MAX_VERSIONS)
+        parser.add_argument(
+            "--types",
+            default=",".join(DEFAULT_TYPES),
+            help="Comma-separated package types",
+        )
+        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)
+        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)
+        dry_run = not apply
+        package_types = [part.strip() for part in args.types.split(",") if part.strip()]
+        file_live, file_notes = collect_live_images_from_files(args.live_images_files)
+        if apply and file_notes:
+            for note in file_notes:
+                print(f"  ERROR: {note}", file=sys.stderr)
+            print("Refusing apply: requested live-image inventory is unavailable or empty", file=sys.stderr)
+            return 2
+        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)
+        if 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)
+
+        plans, errors = build_delete_plans(
+            base_url=args.base_url,
+            token=token,
+            owner=args.owner,
+            package_types=package_types,
+            max_versions=max(1, args.max_versions),
+            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:
+                print(
+                    f"  skip protected  {plan.package_type}/{plan.name}:{plan.version}",
+                    file=sys.stderr,
+                )
+                continue
+            if dry_run:
+                print(
+                    f"  would delete    {plan.package_type}/{plan.name}:{plan.version}",
+                    file=sys.stderr,
+                )
+                continue
+            try:
+                delete_version(
+                    args.base_url,
+                    token,
+                    args.owner,
+                    plan.package_type,
+                    plan.name,
+                    plan.version,
+                    dry_run=False,
+                )
+                deleted.append(plan)
+                print(
+                    f"  deleted         {plan.package_type}/{plan.name}:{plan.version}",
+                    file=sys.stderr,
+                )
+            except urllib.error.HTTPError as exc:
+                errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: HTTP {exc.code}")
+            except Exception as exc:  # noqa: BLE001
+                errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: {exc}")
+
+        summary = emit_summary(
+            owner=args.owner,
+            max_versions=args.max_versions,
+            package_types=package_types,
+            protected=protected,
+            plans=plans,
+            errors=errors,
+            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))
+        else:
+            print(json.dumps(summary, indent=2))
+
+        return 1 if errors else 0
+
+
+    if __name__ == "__main__":
+        raise SystemExit(main())
 kind: ConfigMap
 metadata:
   name: actcore-ops-service-inventory
@@ -1528,6 +2177,7 @@ spec:
   template:
     metadata:
       annotations:
+        activity-core/retention-sha256: fa3d4cc8cdcc14ecce509cf6865c6e4726d722f84814b97d40686d68d6ee5bb1
         kubectl.kubernetes.io/restartedAt: '2026-09-05T20:48:57+02:00'
       labels:
         app.kubernetes.io/name: actcore-worker
@@ -1582,6 +2232,10 @@ spec:
         - mountPath: /var/custodian/runtime/prompts
           name: custodian-runtime
           readOnly: true
+        - mountPath: /opt/railiance-platform/scripts/forgejo_package_prune.py
+          name: ops-service-inventory
+          subPath: forgejo_package_prune.py
+          readOnly: true
         - mountPath: /opt/railiance-platform
           name: railiance-platform
           readOnly: true