scripts/disk-gc-plan.py

@@ -42,6 +42,7 @@ DEFAULT_CONTRACT = REPO / "disk-gc-threat-response.prose.md"
 AMBER, RED, CRITICAL = 75, 85, 95
 
 IDENTITY_FIELDS = ("id", "guest", "vmid", "ct", "ctid", "hostname", "name", "ip")
+IDENTITY_TYPE_PREFIX = re.compile(r"^(?:lxc|qemu)/")
 UNIDENTIFIED_REASON = "unidentified target - refusing to schedule GC"
 
 
@@ -64,14 +65,40 @@ def load_report_only_guests(contract_path: pathlib.Path) -> list[dict]:
     )
 
 
+def _canonical_number(number: float) -> str:
+    if float(number).is_integer():
+        return str(int(number))
+    return str(number).strip().lower()
+
+
+def _normalize_identity(value: object) -> str:
+    """Canonicalise a guest identity so differently-encoded ids compare equal:
+    numeric and numeric-string ids collapse to an integer string, Proxmox
+    type prefixes and leading zeros are stripped, and hostnames/IPs are only
+    trimmed and lowercased."""
+    if isinstance(value, bool):
+        return str(value).strip().lower()
+    if isinstance(value, (int, float)):
+        return _canonical_number(float(value))
+    text = str(value).strip().lower()
+    text = IDENTITY_TYPE_PREFIX.sub("", text)
+    try:
+        return _canonical_number(float(text))
+    except ValueError:
+        return text
+
+
 def _keys(entry: dict) -> set[str]:
     """Guest/host identity keys, shared by exclusions and scan entries so the two
     sides of the gate can never key on different fields."""
     out: set[str] = set()
     for field in IDENTITY_FIELDS:
         value = entry.get(field)
-        if value is not None and str(value).strip():
-            out.add(str(value).strip().lower())
+        if value is None:
+            continue
+        key = _normalize_identity(value)
+        if key:
+            out.add(key)
     return out
 
 

tests/test_disk_gc_report_only.py

@@ -61,6 +61,20 @@ def test_exclusion_matches_on_any_identity_key(tmp_path):
         assert all(r["action"] == "report-only" for r in plan), (entry, plan)
 
 
+def test_exclusion_matches_encoded_identities(tmp_path):
+    """A differently-encoded CT 111 id must not slip past the gate to gc-executor."""
+    encodings = ({"id": 111.0},
+                 {"id": "111.0"},
+                 {"id": "lxc/111"},
+                 {"id": "qemu/111"},
+                 {"id": "0111"},
+                 {"id": "  111  "})
+    for alias in encodings:
+        plan = _plan([{**alias, "usage_pct": 97}], tmp_path)
+        assert plan, alias
+        assert plan[0]["action"] == "report-only", (alias, plan)
+
+
 def test_our_own_guests_still_get_gc(tmp_path):
     """acerpve .9 and amdpve .15 are ours — they must still be acted on."""
     plan = _plan([{"hostname": "acerpve", "ip": "192.168.68.9", "usage_pct": 77},