homelab apt-upgrade: add --safe (pct snapshot + vzdump fallback)

F3 from the apt-sweep backlog. `homelab apt-upgrade --safe ...` takes a
pre-upgrade snapshot per LXC before launching the apt run, so rollback is
trivial if anything regresses.

Snapshot strategy per target:
- LXC: try `pct snapshot <id> preupgrade_<timestamp>` first (CoW, near
  instant). If that refuses ("snapshot feature is not available" — the
  failure mode for LXCs with host bind-mounts like `mp0: /mnt/library`),
  fall back to `vzdump <id> --mode snapshot --storage local --compress
  zstd`. Slower but works on bind-mounted LXCs.
- hubris (PVE host): skipped intentionally; no host-level snapshot in v1.

If any snapshot fails the entire run refuses unless --force is passed.
Snapshot rollback hints are printed after launch so the operator has the
recovery path one copy/paste away.

Validated 2026-05-21: py_compile clean; --status mode works on bind-
mounted LXCs (skips snapshot path). Live snapshot test deferred to next
real fleet sweep to avoid leaving stale artifacts.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

bin/homelab

@@ -20,6 +20,7 @@ import shutil
 import subprocess
 import sys
 import urllib.request
+from datetime import datetime
 from pathlib import Path
 
 try:
@@ -1039,6 +1040,51 @@ def _audit_one(name: str, pve_id: str | None) -> dict | None:
         return None
 
 
+def _snapshot_target(name: str, pve_id: str | None, snap_name: str) -> tuple[str, str]:
+    """Take a pre-upgrade snapshot of a target.
+
+    Tries `pct snapshot` first (CoW, near-instant). Falls back to `vzdump
+    --mode snapshot` for LXCs where pct snapshot refuses due to host bind-
+    mounts. Hubris is skipped (no PVE-host-level snapshot supported here).
+
+    Returns (method, detail) where method is one of:
+      "pct"     -> pct snapshot succeeded; detail = snapshot name
+      "vzdump"  -> vzdump succeeded; detail = backup file path
+      "skip"    -> hubris (skipped intentionally)
+      "fail"    -> both failed; detail = error message
+    """
+    if pve_id is None:
+        return ("skip", "hubris (no host-level snapshot)")
+    # pct snapshot first
+    res = subprocess.run(hubris_ssh() + ["--", "pct", "snapshot", pve_id, snap_name,
+                                          "--description", f"homelab apt-upgrade --safe ({name})"],
+                         capture_output=True, text=True)
+    if res.returncode == 0:
+        return ("pct", snap_name)
+    err = (res.stderr or res.stdout or "").strip()
+    # Common refusal for LXCs with host bind-mounts.
+    bind_mount_refused = (
+        "snapshot feature is not available" in err.lower()
+        or "is not snapshottable" in err.lower()
+        or "snapshots are not supported" in err.lower()
+    )
+    if not bind_mount_refused:
+        return ("fail", err)
+    # Fall back to vzdump
+    res = subprocess.run(hubris_ssh() + ["--", "vzdump", pve_id,
+                                          "--mode", "snapshot",
+                                          "--storage", "local",
+                                          "--compress", "zstd",
+                                          "--notes-template", f"homelab apt-upgrade --safe ({name})"],
+                         capture_output=True, text=True)
+    if res.returncode == 0:
+        # Extract the file path from vzdump output (line like "creating archive '/var/lib/vz/dump/...vma.zst'").
+        m = _re.search(r"creating[^']*'([^']+)'", res.stdout)
+        path = m.group(1) if m else "(vzdump complete)"
+        return ("vzdump", path)
+    return ("fail", (res.stderr or res.stdout or "").strip())
+
+
 def cmd_apt_audit(args: argparse.Namespace) -> int:
     """Per-host pre-flight: dpkg state, holds, upgradable count, non-apt binaries, DNS health.
 
@@ -1116,6 +1162,24 @@ fi
             print("  or rerun with --force to skip the audit gate", file=sys.stderr)
             return 2
 
+    # Optional snapshot pass — LXC-only; hubris is skipped.
+    snap_results: dict[str, tuple[str, str]] = {}
+    if args.safe:
+        # pct snapshot names must match [a-zA-Z][a-zA-Z0-9_]* — underscores only.
+        snap_name = "preupgrade_" + datetime.now().strftime("%Y%m%d_%H%M")
+        print(f"snapshot pass ({snap_name}):")
+        any_fail = False
+        for name, pve_id in targets:
+            method, detail = _snapshot_target(name, pve_id, snap_name)
+            snap_results[name] = (method, detail)
+            print(f"  {name:<20} {method:<7} {detail}")
+            if method == "fail":
+                any_fail = True
+        if any_fail and not args.force:
+            print("refusing: at least one snapshot failed; rerun with --force to upgrade anyway",
+                  file=sys.stderr)
+            return 3
+
     # Launch on each target
     rc = 0
     for name, pve_id in targets:
@@ -1125,6 +1189,16 @@ fi
             rc = 1
             continue
         print(f"{name}: {res.stdout.strip()}")
+
+    # Reminder of snapshot rollback paths after launch.
+    if snap_results:
+        print()
+        print("snapshots created (rollback path on failure):")
+        for name, (method, detail) in snap_results.items():
+            if method == "pct":
+                print(f"  {name}: pct rollback <id> {detail}  (or pct delsnapshot <id> {detail})")
+            elif method == "vzdump":
+                print(f"  {name}: pct restore <id> {detail}  (or rm {detail} when no longer needed)")
     return rc
 
 
@@ -1202,8 +1276,10 @@ def main() -> int:
     grp.add_argument("--all", action="store_true", help="upgrade hubris + every LXC")
     sp.add_argument("--status", action="store_true",
                     help="show running screen sessions + tail the upgrade log on each target")
+    sp.add_argument("--safe", action="store_true",
+                    help="take a pre-upgrade snapshot per LXC (pct snapshot, vzdump fallback)")
     sp.add_argument("--force", action="store_true",
-                    help="skip the pre-flight dpkg-audit gate")
+                    help="skip the pre-flight dpkg-audit gate AND proceed past snapshot failures")
     sp.set_defaults(func=cmd_apt_upgrade)
 
     sp = sub.add_parser("nuke", help="shred /etc/age/key.txt + /opt/homelab-context on a host")