fix(agent): route disk pressure to bounded repair

scripts/ops/docker-disk-pressure-retention-cleanup.py

@@ -2,9 +2,10 @@
 """Bounded Docker disk-pressure cleanup for host reboot recovery.
 
 This controller intentionally avoids `docker system prune`, volumes, containers,
-running images, databases, backups, and logs.  It only removes dangling images
-that are not referenced by any container, and can optionally run a bounded
-BuildKit cache cleanup with an explicit keep-storage floor.
+running images, databases, backups, and logs. It removes only images that are
+not referenced by any container. Tagged-image cleanup is explicit, age-bounded,
+and preserves a per-repository rollback window. BuildKit cleanup is separately
+opt-in and preserves an explicit storage floor.
 """
 
 from __future__ import annotations
@@ -13,14 +14,18 @@ import argparse
 import json
 import subprocess
 import sys
+from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Iterable
-
+from typing import Any
 
+UTC = timezone.utc  # noqa: UP017 - controller still runs on Python 3.9 hosts.
 DEFAULT_MIN_AGE_HOURS = 24
 DEFAULT_KEEP_DANGLING_NEWEST = 20
+DEFAULT_TAGGED_MIN_AGE_HOURS = 168
+DEFAULT_KEEP_TAGGED_NEWEST_PER_REPOSITORY = 3
+DEFAULT_MAX_IMAGE_REMOVALS = 100
 DEFAULT_BUILDER_KEEP_STORAGE = "30GB"
 
 
@@ -42,6 +47,30 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--host-label", default="")
     parser.add_argument("--min-age-hours", type=int, default=DEFAULT_MIN_AGE_HOURS)
     parser.add_argument("--keep-dangling-newest", type=int, default=DEFAULT_KEEP_DANGLING_NEWEST)
+    parser.add_argument(
+        "--include-tagged-images",
+        action="store_true",
+        help=(
+            "Also remove old, container-unreferenced tagged images while preserving "
+            "the newest images for every repository."
+        ),
+    )
+    parser.add_argument(
+        "--tagged-min-age-hours",
+        type=int,
+        default=DEFAULT_TAGGED_MIN_AGE_HOURS,
+    )
+    parser.add_argument(
+        "--keep-tagged-newest-per-repository",
+        type=int,
+        default=DEFAULT_KEEP_TAGGED_NEWEST_PER_REPOSITORY,
+    )
+    parser.add_argument(
+        "--max-image-removals",
+        type=int,
+        default=DEFAULT_MAX_IMAGE_REMOVALS,
+        help="Maximum image IDs removed by one controlled apply.",
+    )
     parser.add_argument(
         "--skip-dangling-images",
         action="store_true",
@@ -102,8 +131,8 @@ def parse_docker_datetime(value: str) -> datetime:
         text = f"{head}.{frac_text}{suffix}"
     parsed = datetime.fromisoformat(text)
     if parsed.tzinfo is None:
-        parsed = parsed.replace(tzinfo=timezone.utc)
-    return parsed.astimezone(timezone.utc)
+        parsed = parsed.replace(tzinfo=UTC)
+    return parsed.astimezone(UTC)
 
 
 def chunked(values: list[str], size: int) -> Iterable[list[str]]:
@@ -170,6 +199,45 @@ def get_dangling_images(docker_bin: str) -> list[ImageInfo]:
     return images
 
 
+def get_all_images(docker_bin: str) -> list[ImageInfo]:
+    image_ids = {
+        normalize_image_id(value)
+        for value in docker(
+            ["image", "ls", "--all", "--quiet", "--no-trunc"],
+            docker_bin,
+        ).stdout.split()
+        if normalize_image_id(value)
+    }
+    images: list[ImageInfo] = []
+    for group in chunked(sorted(image_ids), 100):
+        result = docker(["image", "inspect", *group], docker_bin)
+        payload = json.loads(result.stdout or "[]")
+        for item in payload:
+            image_id = normalize_image_id(str(item.get("Id") or ""))
+            if not image_id:
+                continue
+            tags = item.get("RepoTags") or []
+            images.append(
+                ImageInfo(
+                    image_id=image_id,
+                    created_at=parse_docker_datetime(str(item.get("Created") or "")),
+                    size_bytes=int(item.get("Size") or 0),
+                    repo_tags=tuple(str(tag) for tag in tags if tag),
+                )
+            )
+    return images
+
+
+def repository_from_tag(tag: str) -> str:
+    value = str(tag or "").strip()
+    if "@" in value:
+        return value.split("@", 1)[0]
+    tail = value.rsplit("/", 1)[-1]
+    if ":" not in tail:
+        return value
+    return value.rsplit(":", 1)[0]
+
+
 def select_dangling_image_removals(
     images: list[ImageInfo],
     protected_ids: set[str],
@@ -192,6 +260,45 @@ def select_dangling_image_removals(
     return sorted(dangling, key=lambda image: image.created_at)
 
 
+def select_tagged_image_removals(
+    images: list[ImageInfo],
+    protected_ids: set[str],
+    *,
+    now: datetime,
+    min_age_hours: int,
+    keep_newest_per_repository: int,
+) -> list[ImageInfo]:
+    unreferenced = [
+        image
+        for image in images
+        if normalize_image_id(image.image_id) not in protected_ids
+        and image.repo_tags
+    ]
+    repository_images: dict[str, list[ImageInfo]] = {}
+    for image in unreferenced:
+        for repository in {
+            repository_from_tag(tag) for tag in image.repo_tags if tag
+        }:
+            repository_images.setdefault(repository, []).append(image)
+
+    retained_ids: set[str] = set()
+    for repository_rows in repository_images.values():
+        repository_rows.sort(key=lambda image: image.created_at, reverse=True)
+        retained_ids.update(
+            normalize_image_id(image.image_id)
+            for image in repository_rows[:keep_newest_per_repository]
+        )
+
+    cutoff_seconds = min_age_hours * 3600
+    selected = [
+        image
+        for image in unreferenced
+        if normalize_image_id(image.image_id) not in retained_ids
+        and (now - image.created_at).total_seconds() >= cutoff_seconds
+    ]
+    return sorted(selected, key=lambda image: image.created_at)
+
+
 def summarize_images(images: list[ImageInfo]) -> dict[str, Any]:
     return {
         "count": len(images),
@@ -202,13 +309,29 @@ def summarize_images(images: list[ImageInfo]) -> dict[str, Any]:
     }
 
 
+def limit_image_removals(
+    images: list[ImageInfo],
+    *,
+    max_removals: int,
+) -> list[ImageInfo]:
+    return sorted(images, key=lambda image: image.created_at)[:max_removals]
+
+
 def remove_images(images: list[ImageInfo], docker_bin: str) -> list[str]:
     removed: list[str] = []
-    for group in chunked([image.image_id for image in images], 25):
-        if not group:
-            continue
+    dangling_ids = [image.image_id for image in images if not image.repo_tags]
+    for group in chunked(dangling_ids, 25):
         docker(["image", "rm", *group], docker_bin)
         removed.extend(group)
+    tagged_images = [image for image in images if image.repo_tags]
+    tagged_references = [
+        reference
+        for image in tagged_images
+        for reference in image.repo_tags
+    ]
+    for group in chunked(tagged_references, 25):
+        docker(["image", "rm", *group], docker_bin)
+    removed.extend(image.image_id for image in tagged_images)
     return removed
 
 
@@ -227,11 +350,11 @@ def builder_prune_command(args: argparse.Namespace) -> list[str]:
 
 
 def build_receipt(args: argparse.Namespace) -> dict[str, Any]:
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
     before = current_disk_bytes(args.disk_path)
     protected_ids = get_container_image_ids(args.docker_bin)
     dangling_images = get_dangling_images(args.docker_bin)
-    removal_candidates = (
+    dangling_removal_candidates = (
         []
         if args.skip_dangling_images
         else select_dangling_image_removals(
@@ -242,6 +365,29 @@ def build_receipt(args: argparse.Namespace) -> dict[str, Any]:
             keep_newest=args.keep_dangling_newest,
         )
     )
+    all_images = get_all_images(args.docker_bin) if args.include_tagged_images else []
+    tagged_removal_candidates = (
+        select_tagged_image_removals(
+            all_images,
+            protected_ids,
+            now=now,
+            min_age_hours=args.tagged_min_age_hours,
+            keep_newest_per_repository=args.keep_tagged_newest_per_repository,
+        )
+        if args.include_tagged_images
+        else []
+    )
+    candidates_by_id = {
+        normalize_image_id(image.image_id): image
+        for image in [*dangling_removal_candidates, *tagged_removal_candidates]
+    }
+    eligible_removal_candidates = sorted(
+        candidates_by_id.values(), key=lambda image: image.created_at
+    )
+    removal_candidates = limit_image_removals(
+        eligible_removal_candidates,
+        max_removals=args.max_image_removals,
+    )
     receipt: dict[str, Any] = {
         "schema_version": "awoooi_docker_disk_pressure_retention_cleanup_v1",
         "generated_at": now.isoformat(),
@@ -254,12 +400,21 @@ def build_receipt(args: argparse.Namespace) -> dict[str, Any]:
             "touches_databases": False,
             "touches_backups": False,
             "uses_docker_system_prune": False,
-            "removes_only_unreferenced_dangling_images": True,
+            "removes_only_container_unreferenced_images": True,
+            "tagged_image_cleanup_requires_explicit_flag": True,
+            "tagged_image_cleanup_preserves_repository_rollback_window": True,
+            "single_apply_image_removal_is_count_bounded": True,
             "builder_cache_cleanup_requires_explicit_flag": True,
         },
         "parameters": {
             "min_age_hours": args.min_age_hours,
             "keep_dangling_newest": args.keep_dangling_newest,
+            "include_tagged_images": args.include_tagged_images,
+            "tagged_min_age_hours": args.tagged_min_age_hours,
+            "keep_tagged_newest_per_repository": (
+                args.keep_tagged_newest_per_repository
+            ),
+            "max_image_removals": args.max_image_removals,
             "include_builder_cache": args.include_builder_cache,
             "builder_keep_storage": args.builder_keep_storage,
             "skip_dangling_images": args.skip_dangling_images,
@@ -267,7 +422,12 @@ def build_receipt(args: argparse.Namespace) -> dict[str, Any]:
         "disk_before": before,
         "protected_container_image_count": len(protected_ids),
         "dangling_image_total_count": len(dangling_images),
-        "dangling_image_removal_plan": summarize_images(removal_candidates),
+        "dangling_image_removal_plan": summarize_images(dangling_removal_candidates),
+        "tagged_image_removal_plan": summarize_images(tagged_removal_candidates),
+        "eligible_image_removal_plan": summarize_images(
+            eligible_removal_candidates
+        ),
+        "total_image_removal_plan": summarize_images(removal_candidates),
         "builder_cache_command": builder_prune_command(args)[1:] if args.include_builder_cache else None,
         "removed_image_ids": [],
         "builder_cache_cleanup_executed": False,
@@ -292,6 +452,15 @@ def main() -> int:
     if args.keep_dangling_newest < 0:
         print("keep-dangling-newest must be >= 0", file=sys.stderr)
         return 2
+    if args.tagged_min_age_hours <= 0:
+        print("tagged-min-age-hours must be > 0", file=sys.stderr)
+        return 2
+    if args.keep_tagged_newest_per_repository < 1:
+        print("keep-tagged-newest-per-repository must be >= 1", file=sys.stderr)
+        return 2
+    if args.max_image_removals < 1:
+        print("max-image-removals must be >= 1", file=sys.stderr)
+        return 2
     receipt = build_receipt(args)
     text = json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
     if args.output:

scripts/ops/tests/test_docker_disk_pressure_retention_cleanup.py

@@ -6,7 +6,6 @@ from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from types import SimpleNamespace
 
-
 ROOT = Path(__file__).resolve().parents[3]
 SCRIPT = ROOT / "scripts" / "ops" / "docker-disk-pressure-retention-cleanup.py"
 spec = importlib.util.spec_from_file_location("docker_disk_pressure_retention_cleanup", SCRIPT)
@@ -14,6 +13,7 @@ module = importlib.util.module_from_spec(spec)
 assert spec and spec.loader
 sys.modules[spec.name] = module
 spec.loader.exec_module(module)
+UTC = timezone.utc  # noqa: UP017 - mirrors supported Python 3.9 runtime.
 
 
 def image(image_id: str, created_at: datetime, tags: tuple[str, ...] = ()):
@@ -26,7 +26,7 @@ def image(image_id: str, created_at: datetime, tags: tuple[str, ...] = ()):
 
 
 def test_select_dangling_images_keeps_newest_and_protects_running_images() -> None:
-    now = datetime(2026, 7, 1, 12, tzinfo=timezone.utc)
+    now = datetime(2026, 7, 1, 12, tzinfo=UTC)
     images = [
         image("sha256:running", now - timedelta(hours=72)),
         image("oldest", now - timedelta(hours=72)),
@@ -73,7 +73,7 @@ def test_parse_docker_datetime_accepts_nanosecond_fraction() -> None:
 
 
 def test_summary_never_reports_volumes_or_container_cleanup_boundary() -> None:
-    now = datetime(2026, 7, 1, 12, tzinfo=timezone.utc)
+    now = datetime(2026, 7, 1, 12, tzinfo=UTC)
     selected = module.select_dangling_image_removals(
         [image("old", now - timedelta(days=3))],
         set(),
@@ -87,6 +87,88 @@ def test_summary_never_reports_volumes_or_container_cleanup_boundary() -> None:
     assert summary["sample_image_ids"] == ["old"]
 
 
+def test_tagged_cleanup_preserves_running_and_repository_rollback_window() -> None:
+    now = datetime(2026, 7, 14, 12, tzinfo=UTC)
+    images = [
+        image("running", now - timedelta(days=30), ("registry:5000/app:running",)),
+        image("latest", now - timedelta(days=1), ("registry:5000/app:latest",)),
+        image("rollback-1", now - timedelta(days=2), ("registry:5000/app:sha-1",)),
+        image("rollback-2", now - timedelta(days=3), ("registry:5000/app:sha-2",)),
+        image("expired", now - timedelta(days=14), ("registry:5000/app:sha-old",)),
+        image("other-repo", now - timedelta(days=14), ("registry:5000/other:only",)),
+    ]
+
+    selected = module.select_tagged_image_removals(
+        images,
+        {"running"},
+        now=now,
+        min_age_hours=7 * 24,
+        keep_newest_per_repository=3,
+    )
+
+    assert [item.image_id for item in selected] == ["expired"]
+
+
+def test_multi_repository_image_is_retained_when_any_repository_needs_it() -> None:
+    now = datetime(2026, 7, 14, 12, tzinfo=UTC)
+    shared = image(
+        "shared",
+        now - timedelta(days=30),
+        ("registry:5000/app:old", "registry:5000/critical:latest"),
+    )
+    selected = module.select_tagged_image_removals(
+        [shared, image("app-new", now - timedelta(days=1), ("registry:5000/app:new",))],
+        set(),
+        now=now,
+        min_age_hours=7 * 24,
+        keep_newest_per_repository=1,
+    )
+
+    assert selected == []
+
+
+def test_repository_parser_preserves_registry_port() -> None:
+    assert module.repository_from_tag("registry:5000/team/app:sha") == "registry:5000/team/app"
+
+
+def test_single_apply_image_removal_is_count_bounded() -> None:
+    now = datetime(2026, 7, 14, 12, tzinfo=UTC)
+    candidates = [
+        image(f"image-{index}", now - timedelta(days=index + 1))
+        for index in range(5)
+    ]
+
+    selected = module.limit_image_removals(candidates, max_removals=2)
+
+    assert [item.image_id for item in selected] == ["image-4", "image-3"]
+
+
+def test_remove_images_batches_tagged_references(monkeypatch) -> None:
+    calls: list[list[str]] = []
+
+    def fake_docker(args: list[str], _docker_bin: str):
+        calls.append(args)
+        return SimpleNamespace(stdout="", stderr="", returncode=0)
+
+    monkeypatch.setattr(module, "docker", fake_docker)
+    now = datetime(2026, 7, 14, 12, tzinfo=UTC)
+
+    removed = module.remove_images(
+        [
+            image("dangling", now),
+            image("tagged-a", now, ("repo-a:old", "repo-b:old")),
+            image("tagged-b", now, ("repo-c:old",)),
+        ],
+        "docker",
+    )
+
+    assert removed == ["dangling", "tagged-a", "tagged-b"]
+    assert calls == [
+        ["image", "rm", "dangling"],
+        ["image", "rm", "repo-a:old", "repo-b:old", "repo-c:old"],
+    ]
+
+
 def test_cli_exposes_builder_cache_only_flag() -> None:
     help_text = module.run_command(["python3", str(SCRIPT), "--help"]).stdout