Restore a container backup with a smaller root disk

Proxmox only grows disks. The restore dialog gets an optional "Root
disk GiB" field for container backups; with a size, the backup is
restored as a new guest with --rootfs <storage>:<N> (Proxmox's advanced
restore mode).

- A smaller disk is only allowed for a new VMID, checked in the dialog
  and again in cli.restore(): with force, Proxmox deletes the old disks
  before it unpacks, and a failed unpack then removes the guest.
- The size must be a positive integer (0 gives an unlimited subvol on
  ZFS/dir storage) and above the used root disk space of a running
  guest.
- The confirm screen is a separate DANGER screen with a red border: it
  says mount points are dropped and their backed-up files land in the
  root disk, raw lxc.* lines can come back under root@pam, root disk
  options are not carried over, and the used space is no fit check.
- Never --unique or ignore-unpack-errors.

Design: docs/plans/lxc-shrink.md (section A).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/plans/lxc-shrink.md

@@ -0,0 +1,121 @@
+# Shrink a container's root disk
+
+## Problem
+
+Proxmox only grows disks (`resize_vm` dies with "unable to shrink disk size"). A
+container whose rootfs was sized too large cannot be made smaller from the web UI or
+this TUI. Two routes exist, verified against pve-container 6.1.18 (master) source:
+
+1. API: restore a backup with an explicit `rootfs=<storage>:<GiB>` ("advanced restore
+   mode", pct.adoc). Reachable through cv4pve-cli (`do restore ct ... --rootfs S:N`
+   forwards the key; checked with `--dry-run`).
+2. Node shell: copy the files into a new, smaller volume and swap it in.
+
+Both are dangerous; the TUI must say so plainly.
+
+## Approach
+
+### A. TUI: smaller rootfs on restore (container backups only)
+
+- `RestoreScreen` gets an optional field "Root disk GiB" for lxc backups. Empty =
+  today's behaviour, unchanged.
+- With a size set, only **New guest** is allowed. Overwrite + size is refused: with
+  `force=1` PVE destroys every old volume before it extracts, and a failed extraction
+  then deletes the new disks and the config too; the container is gone.
+- `cli.restore()` gets `rootfs_gb: int | None`; when set it checks it with
+  `_positive` (a size of 0 makes PVE allocate an *unlimited* subvol on ZFS/dir) and
+  adds `--rootfs <storage>:<N>` (same storage as the dialog's choice). Never
+  `ignore-unpack-errors`.
+- The confirm screen for a shrink restore is a separate, red warning:
+  - every mpX of the backup is dropped from the new config; files of `backup=1`
+    mount points land inside the root disk and count against its size; `backup=0`
+    mount points are not in the backup at all;
+  - raw `lxc.*` lines (including `lxc.mount.entry` bind mounts) may come back when
+    cv4pve-cli runs as root@pam: do not run both guests at once;
+  - rootfs options (acl, quota, mountoptions, replicate) are not carried over;
+  - names the guest's *current* mount points, saying the backup may differ;
+  - shows rootfs used space when the guest is running (`status/current` `disk`;
+    0 when stopped, shown as "unknown"), and says this is not a fit check: the
+    backup may be older, mount point data and ext4 overhead (5 % reserve, inodes)
+    come on top;
+  - a too-small disk makes the restore task fail; the original is untouched because
+    this is always a new VMID.
+- Hard refusal only where it is certain: size not a positive integer, or size <=
+  used rootfs GiB of a running guest.
+- Known limit, unchanged from today's restore: `run(timeout=3600)` can give up while
+  the PVE task still runs.
+
+### B. Node script `scripts/pct-shrink-rootfs` (bash, run as root on the node)
+
+Usage: `pct-shrink-rootfs [--yes] <vmid> <new-size-GiB>`. Without `--yes` it only
+measures and prints the plan; measuring runs `pct mount`/`pct unmount`, which sets
+and clears `lock: mounted` (so even a dry run touches the config briefly).
+
+Refuses (exit non-zero, nothing changed) when: not root; `pct` missing; CT not
+stopped; config has any `[...]` section (snapshots, `[pve:pending]`); `lock:` set;
+`template: 1` or `protection: 1`; HA-managed (`ha-manager status`); a replication
+job exists (`pvesr list`); rootfs is not a volume; **any `mpN` or `lxc.mount.entry`**
+(rootfs-only CTs: no shadowed files, `df` measures rootfs alone, `pct mount` cannot
+fail on another storage); new size >= current size; size not a positive integer.
+
+Fit check after `pct mount` (`df -B1` and `df -i` on the rootfs): used bytes * 1.1 <=
+N GiB * 0.93 (5 % root reserve + ~2 % ext4 metadata), and used inodes * 1.1 <= the
+new disk's inodes (N * 65536, mkfs.ext4 default ratio). On any error after
+`pct mount` the trap runs `pct unmount`; if that fails it stops and prints the manual
+cleanup (`pct unmount`, `pct unlock`), touching nothing else.
+
+Steps (copy route, any storage):
+
+1. `pct set <id> --mp9 <storage>:<N>,mp=/.shrink-new` - PVE allocates and formats;
+   the script reads back mp9's volid and `size=`.
+2. `pct mount <id>`; `rsync -aHAX --numeric-ids --whole-file --sparse
+   --one-file-system --exclude=/.shrink-new` old rootfs -> new; `pct unmount <id>`.
+   On rsync failure: unmount, then free the new volume explicitly - `pct set
+   --delete mp9` only moves it to `unusedN`, so the script then deletes the
+   `unusedK` whose value is exactly the new volid (found by value, never by index).
+3. Swap, under PVE's config flock (`flock /run/lock/lxc/pve-config-<id>.lock`):
+   copy the config to `/root/pct-shrink-<id>-<timestamp>.conf`; rewrite the main
+   section with awk (exact string compare, no regex on volids) into
+   `<conf>.tmp.<pid>` and rename it, as PVE's `file_set_contents` does on pmxcfs;
+   refuse if rootfs or mp9 changed since the copy: `rootfs:` gets the
+   new volid with the old options and mp9's `size=`, the `mp9:` line goes, and
+   `unusedK: <old volid>` is added at the first free K. No `pct rescan`: the script
+   knows both volids and the size, so rescan's side effects (every orphan of the
+   VMID added as unused, `size=0` on pve-container < 6.1.9) are avoided.
+4. Print the exact `unusedK` key and value of the old rootfs: start the CT, check
+   it, then `pct set <id> --delete unusedK`. The script never deletes the old volume.
+
+The in-place ZFS `refquota` shortcut is left out: one route for every storage is
+less code to get wrong, and the copy keeps the original until the user deletes it.
+chattr flags (`+i`, `+a`) are not copied by rsync; the script says so.
+
+Tests: pytest runs the script with stub `pct`, `df`, `rsync`, `flock`,
+`ha-manager`, `pvesr` on `PATH` and a temp config dir (`PCT_CONF_DIR` override,
+default `/etc/pve/lxc`; the plan output prints the directory used). bats is not
+installed and CI only runs pytest. ShellCheck via the edit hook.
+
+## Out of scope
+
+- Shrinking mpX volumes, and containers that have any mount point (script).
+- Overwrite-in-place restore with a smaller disk.
+- Restoring back to the original VMID (second backup/restore cycle).
+- Running the node script from the TUI (it needs a root shell on the node).
+- Moving `backup=0` mount points to the new guest.
+
+## Steps
+
+- [x] `cli.restore(rootfs_gb=...)` adds `--rootfs S:N`; tests in `test_cli.py`.
+- [x] RestoreScreen: size field for lxc, refuse with overwrite, refuse <= used of a
+      running guest; shrink confirm text with the warnings above; tests in
+      `test_app.py`.
+- [x] `scripts/pct-shrink-rootfs` + `tests/test_pct_shrink.py`: each refusal, dry
+      run changes no config, happy path (exact new config, `unusedK` at first free
+      K with existing `unused0`), dir-style volid with `/` and `.`, rsync failure
+      frees the new volume by value, unmount failure stops.
+- [x] README: key table row and a warning paragraph; script usage.
+- [ ] Full `uv run pytest`, ruff, shellcheck green.
+
+## Open questions
+
+- Should a shrink restore pass `--unique 1` (new MACs)? Today's restore does not, it
+  warns instead. Proposal: keep that, the warning already covers it.

src/cv4pve_tui/app.py

@@ -56,6 +56,7 @@ class DialogScreen(ModalScreen):
     DialogScreen .row Label { width: 14; height: 3; content-align: left middle; }
     DialogScreen .row Select, DialogScreen .row Input { width: 1fr; }
     DialogScreen .row Select.mode { width: 16; }
+    DialogScreen #dialog.danger { border: heavy $error; }
     DialogScreen .sub {
         height: auto; border: round $accent; border-title-style: bold; padding: 0 1;
     }
@@ -72,12 +73,15 @@ class DialogScreen(ModalScreen):
 class ConfirmScreen(DialogScreen):
     BINDINGS = [("escape", "dismiss(False)", "No")]  # noqa: RUF012 - Textual idiom
 
-    def __init__(self, message: str, title: str = "Confirm") -> None:
+    def __init__(
+        self, message: str, title: str = "Confirm", danger: bool = False
+    ) -> None:
         super().__init__()
-        self.message, self.title_text = message, title
+        self.message, self.title_text, self.danger = message, title, danger
 
     def compose(self) -> ComposeResult:
-        with self.box(self.title_text):
+        with self.box(self.title_text) as box:
+            box.set_class(self.danger, "danger")
             yield Static(self.message, markup=False)
             with Horizontal(classes="buttons"):
                 yield Button("Yes", id="yes", variant="error")
@@ -706,7 +710,7 @@ class UpdatesScreen(ModalScreen):
 
 
 class RestoreScreen(DialogScreen):
-    """Restore one backup; dismisses with {overwrite, vmid, storage, start}, or None."""
+    """Restore one backup; dismisses with {overwrite, vmid, storage, start, rootfs_gb}, or None."""
 
     BINDINGS = [("escape", "dismiss(None)", "Cancel")]  # noqa: RUF012 - Textual idiom
 
@@ -735,6 +739,13 @@ class RestoreScreen(DialogScreen):
                     id="storage",
                 )
             yield Checkbox("Start after restore", id="start")
+            if "vzdump-lxc-" in self.backup["volid"]:
+                with Horizontal(classes="row"):
+                    yield Label("Root disk GiB")
+                    yield Input(
+                        placeholder="empty = as in the backup; smaller = DANGEROUS",
+                        id="rootfs",
+                    )
             with Horizontal(classes="row"):
                 yield Label("Confirm")
                 yield Input(placeholder="type the VMID to overwrite", id="confirm")
@@ -752,6 +763,21 @@ class RestoreScreen(DialogScreen):
             self.dismiss(None)
             return
         overwrite = self.query_one("#overwrite", RadioButton).value
+        rootfs_gb = None
+        for field in self.query("#rootfs"):
+            text = field.value.strip()
+            if text and (not text.isdecimal() or int(text) <= 0):
+                self.app.notify(
+                    "The root disk size must be a positive whole number of GiB",
+                    severity="error",
+                )
+                return
+            rootfs_gb = int(text) if text else None
+        if rootfs_gb and overwrite:  # force=1 destroys the old disks before extracting
+            self.app.notify(
+                "A smaller root disk is only possible as a new guest", severity="error"
+            )
+            return
         if overwrite:
             vmid = self.guest["vmid"]
             if self.query_one("#confirm", Input).value != str(vmid):
@@ -778,6 +804,7 @@ class RestoreScreen(DialogScreen):
                 "vmid": vmid,
                 "storage": self.query_one("#storage", Select).value,
                 "start": self.query_one("#start", Checkbox).value,
+                "rootfs_gb": rootfs_gb,
             }
         )
 
@@ -1198,8 +1225,9 @@ class InfoScreen(ModalScreen):
             if values["overwrite"] and cli.status(guest) != "stopped":
                 raise cli.CliError(f"{who} must be stopped to be overwritten")
             cli.restore(
-                guest, volid, vmid, values["storage"], values["overwrite"], start
-            )
+                guest, volid, vmid, values["storage"], values["overwrite"], start,
+                rootfs_gb=values["rootfs_gb"],
+            )  # fmt: skip
 
         def confirmed(ok: bool) -> None:
             if ok:
@@ -1215,12 +1243,61 @@ class InfoScreen(ModalScreen):
                 ),
                 confirmed,
             )
+        elif values["rootfs_gb"]:
+            self._confirm_shrink(what, text, values, confirmed)
         else:
             self.app.push_screen(
                 ConfirmScreen(f"Restore {what} as new guest {vmid}? {text}", "Restore"),
                 confirmed,
             )
 
+    @work(thread=True)
+    def _confirm_shrink(self, what: str, text: str, values: dict, confirmed) -> None:
+        size = values["rootfs_gb"]
+        try:  # live: the used space only exists while the guest runs
+            now = cli.current(self.guest)
+        except Exception as exc:  # noqa: BLE001 - surface every failure, never kill the app
+            self.app.call_from_thread(self.app.notify, _message(exc), severity="error")
+            return
+        used = int(now.get("disk") or 0) if now["status"] == "running" else 0
+        if used and size * 2**30 <= used:
+            self.app.call_from_thread(
+                self.app.notify,
+                f"{size} GiB is not above the {used / 2**30:.1f} GiB the running guest "
+                "already uses on its root disk",
+                severity="error",
+            )
+            return
+        mounts = ", ".join(
+            f"{k} ({m[1]})"
+            for k, v in sorted(self.config.items())
+            if re.fullmatch(r"mp\d+", k) and (m := re.search(r"\bmp=([^,]+)", str(v)))
+        )
+        message = (
+            f"DANGER: restore {what} as new guest {values['vmid']} with a root disk of "
+            f"only {size} GiB.\n\n"
+            "- Every mount point (mpN) of the backup is dropped from the new guest. "
+            "Files of mount points with backup=1 land inside the root disk and count "
+            "against its size; mount points with backup=0 are not in the backup at all.\n"
+            f"- The guest has these mount points now: {mounts or 'none'}. The backup "
+            "may differ.\n"
+            "- Raw lxc.* lines, including lxc.mount.entry bind mounts, may come back "
+            "when cv4pve-cli runs as root@pam: never run both guests at once.\n"
+            "- Root disk options (acl, quota, mountoptions, replicate) are not carried "
+            "over.\n"
+            f"- Used on the root disk now: "
+            f"{f'{used / 2**30:.1f} GiB' if used else 'unknown'}. This is not a fit "
+            "check: the backup may be older, mount point data and ext4 overhead (5 % "
+            "reserve, inodes) come on top.\n"
+            "- If the disk is too small the restore task fails. The original is "
+            f"untouched: this is always a new guest.\n\n{text}"
+        )
+        self.app.call_from_thread(
+            self.app.push_screen,
+            ConfirmScreen(message, "DANGER: smaller root disk", danger=True),
+            confirmed,
+        )
+
     @work(thread=True)
     def _confirm_overwrite(self, screen: ConfirmScreen, confirmed) -> None:
         try:  # live status: the guest table's status may be stale

src/cv4pve_tui/cli.py

@@ -215,8 +215,12 @@ def restore(
     storage: str,
     force: bool = False,
     start: bool = False,
+    rootfs_gb: int | None = None,
 ) -> None:
-    """Restore `volid` as `vmid`; `force` (overwrite) only for the guest's own VMID."""
+    """Restore `volid` as `vmid`; `force` (overwrite) only for the guest's own VMID.
+
+    `rootfs_gb` (container, new guest only) restores onto a root disk of that size.
+    """
     _positive(vmid, "vmid")
     if force and vmid != guest["vmid"]:
         raise ValueError(
@@ -230,11 +234,18 @@ def restore(
         )
     if not re.fullmatch(_STORAGE, storage):
         raise ValueError(f"invalid storage name: {storage!r}")
+    if rootfs_gb is not None:
+        # 0 would make PVE allocate an unlimited subvol on ZFS
+        _positive(rootfs_gb, "root disk size")
+        if force or backup_type != "lxc":
+            raise ValueError("a smaller root disk needs a container and a new guest")
     opts = ["--storage", storage]
     if force:
         opts += ["--force", "1"]
     if start:
         opts += ["--start", "1"]
+    if rootfs_gb is not None:
+        opts += ["--rootfs", f"{storage}:{rootfs_gb}"]
     run(
         "do", "restore", kind(guest), guest["node"], str(vmid), volid, *opts,
         "-y", "--wait", timeout=3600,

tests/test_app.py

@@ -105,6 +105,7 @@ class Fake:
         self.nextid, self.nextid_error = 105, None
         self.restore_store = ["local-lvm", "zfs-pool"]
         self.restore_calls, self.restore_error, self.restore_hook = [], None, None
+        self.restore_sizes = []  # rootfs_gb of every restore call
 
     def list_guests(self):
         self.list_calls += 1
@@ -298,8 +299,11 @@ class Fake:
     def restore_storages(self, guest):
         return list(self.restore_store)
 
-    def restore(self, guest, volid, vmid, storage, force=False, start=False):
+    def restore(
+        self, guest, volid, vmid, storage, force=False, start=False, rootfs_gb=None
+    ):
         self.restore_calls.append((guest["vmid"], volid, vmid, storage, force, start))
+        self.restore_sizes.append(rootfs_gb)
         if self.restore_hook:
             self.restore_hook()
         if self.restore_error:
@@ -4921,3 +4925,187 @@ async def test_keybar_follows_the_info_screen_bindings(fake):
         await open_info(app, pilot)
         _bar, text = keybar_text(app)
         assert "Restore backup" in text
+
+
+# --- shrink restore: smaller rootfs (docs/plans/lxc-shrink.md, A) ---------------------------
+
+CT_VOLID_SHRINK = "truenas-backup:backup/vzdump-lxc-100-new.tar.zst"
+CT_MPS = {
+    "rootfs": "zfs-pool:subvol-100-disk-0,size=8G",
+    "mp0": "local-lvm:vm-100-disk-1,mp=/data,size=4G",
+    "mp1": "/srv/share,mp=/share",
+}
+
+
+def ct_shrink_setup(fake, status="stopped", disk=0, mps=True):
+    fake.guests = [VM, dict(CT, status=status, disk=disk)]
+    fake.net_configs[100] = dict(CT_MPS) if mps else {"rootfs": CT_MPS["rootfs"]}
+    fake.disk_map = {"rootfs": "8G"}
+    fake.backup_list = [
+        {**BACKUPS[0], "volid": CT_VOLID_SHRINK, "notes": "backrest (manual)"}
+    ]
+
+
+async def open_ct_restore(app, pilot):
+    await open_restore(app, pilot, down=1)
+
+
+async def apply_shrink(app, pilot, size):
+    set_value(app, "#rootfs", size)
+    await press_button(app, pilot, "#apply")
+
+
+async def test_rootfs_field_exists_for_lxc_backup_and_is_empty(fake):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        assert app.screen.query_one("#rootfs", Input).value == ""
+
+
+async def test_rootfs_field_is_absent_for_vm_backup(fake):
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_restore(app, pilot)
+        assert not app.screen.query("#rootfs")
+
+
+async def test_lxc_restore_with_empty_size_is_unchanged(fake):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_new(app, pilot)
+        assert app.screen.title_text == "Restore"
+        await press_button(app, pilot, "#yes")
+        assert fake.restore_sizes == [None]
+
+
+async def test_shrink_confirm_is_a_danger_screen_naming_mount_points(fake):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, "4")
+        assert isinstance(app.screen, ConfirmScreen)
+        assert app.screen.title_text.startswith("DANGER")
+        assert app.screen.query_one("#dialog").has_class("danger")
+        text = text_of(app.screen)
+        assert text.startswith("DANGER")
+        for needle in ("mp0", "/data", "mp1", "/share", "may differ", "backup=0",
+                       "lxc.mount.entry", "root@pam", "acl", "unknown", "4 GiB",
+                       "105", "zfs-pool"):  # fmt: skip
+            assert needle in text, needle
+        assert fake.restore_calls == []
+
+
+async def test_shrink_confirm_without_mount_points_says_none(fake):
+    ct_shrink_setup(fake, mps=False)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, "4")
+        assert "mp0" not in text_of(app.screen)
+        assert "none" in text_of(app.screen).lower()
+
+
+async def test_shrink_confirmed_restores_new_guest_with_rootfs_gb(fake):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, " 4 ")
+        await press_button(app, pilot, "#yes")
+        assert fake.restore_calls == [
+            (100, CT_VOLID_SHRINK, 105, "zfs-pool", False, False)
+        ]
+        assert fake.restore_sizes == [4]
+        assert app.busy == set() and errors(app) == []
+
+
+@pytest.mark.parametrize("how", ["no", "escape"])
+async def test_shrink_confirm_declined_makes_no_call(fake, how):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, "4")
+        if how == "no":
+            await press_button(app, pilot, "#no")
+        else:
+            await pilot.press("escape")
+            await settle(app, pilot)
+        assert fake.restore_calls == [] and app.busy == set()
+
+
+@pytest.mark.parametrize("bad", ["0", "-3", "abc", "1.5", "4G", "1e1"])
+async def test_shrink_size_must_be_a_positive_integer(fake, bad):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, bad)
+        assert errors(app)
+        assert fake.restore_calls == []
+        assert not isinstance(app.screen, ConfirmScreen)
+
+
+async def test_shrink_with_overwrite_is_refused(fake):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        set_value(app, "#rootfs", "4")
+        await apply_overwrite(app, pilot, confirm="100")
+        assert errors(app)
+        assert fake.restore_calls == []
+        assert not isinstance(app.screen, ConfirmScreen)
+
+
+@pytest.mark.parametrize("size", ["1", "3"])
+async def test_shrink_size_at_or_below_used_of_running_guest_is_refused(fake, size):
+    ct_shrink_setup(fake, status="running", disk=3 * 2**30)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, size)
+        assert errors(app)
+        assert fake.restore_calls == []
+        assert not isinstance(app.screen, ConfirmScreen)
+
+
+async def test_shrink_size_above_used_of_running_guest_shows_used_space(fake):
+    ct_shrink_setup(fake, status="running", disk=3 * 2**30)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, "4")
+        assert isinstance(app.screen, ConfirmScreen)
+        text = text_of(app.screen)
+        assert "3.0 GiB" in text and "unknown" not in text
+        assert "not a fit check" in text
+
+
+@pytest.mark.parametrize(("status", "disk"), [("running", 0), ("stopped", 3 * 2**30)])
+async def test_shrink_is_not_refused_when_used_space_is_unknown_or_guest_stopped(
+    fake, status, disk
+):
+    ct_shrink_setup(fake, status=status, disk=disk)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        await apply_shrink(app, pilot, "1")
+        assert isinstance(app.screen, ConfirmScreen)
+        assert errors(app) == []
+
+
+async def test_shrink_status_failure_is_notified_without_confirm(fake):
+    ct_shrink_setup(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_ct_restore(app, pilot)
+        fake.current_error = cli_error("Error: boom")
+        await apply_shrink(app, pilot, "4")
+        assert errors(app)
+        assert fake.restore_calls == []
+        assert not isinstance(app.screen, ConfirmScreen)

tests/test_cli.py

@@ -1767,3 +1767,55 @@ def test_restore_refuses_a_backup_of_the_other_guest_type(fake, force):
             ct, "local:backup/vzdump-qemu-100-x.vma.zst", 100, "local-lvm", force=force
         )
     assert fake.calls == []
+
+
+# --- shrink restore: smaller rootfs (docs/plans/lxc-shrink.md, A) ------------
+
+
+def test_restore_ct_with_rootfs_gb_adds_rootfs_on_the_chosen_storage(fake):
+    cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=4)
+    assert fake.argv == [
+        "cv4pve-cli", "do", "restore", "ct", "mainframe", "105", CT_VOLID,
+        "--storage", "zfs", "--rootfs", "zfs:4", "-y", "--wait",
+    ]  # fmt: skip
+
+
+def test_restore_ct_with_rootfs_gb_and_start_exact_argv(fake):
+    cli.restore(CT, CT_VOLID, 105, "zfs", start=True, rootfs_gb=4)
+    assert fake.argv == [
+        "cv4pve-cli", "do", "restore", "ct", "mainframe", "105", CT_VOLID,
+        "--storage", "zfs", "--start", "1", "--rootfs", "zfs:4", "-y", "--wait",
+    ]  # fmt: skip
+
+
+def test_restore_without_rootfs_gb_argv_is_unchanged(fake):
+    cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=None)
+    assert "--rootfs" not in fake.argv
+    assert fake.argv[-4:] == ["--storage", "zfs", "-y", "--wait"]
+
+
+def test_restore_with_rootfs_gb_never_adds_unique_or_ignore_unpack_errors(fake):
+    cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=4)
+    joined = " ".join(fake.argv)
+    assert "--unique" not in joined and "ignore-unpack-errors" not in joined
+
+
+@pytest.mark.parametrize("bad", [0, -1, True, "4", 1.5])
+def test_restore_rootfs_gb_must_be_a_positive_int_and_never_runs(fake, bad):
+    # a size of 0 makes PVE allocate an unlimited subvol on ZFS/dir
+    with pytest.raises(ValueError):
+        cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=bad)
+    assert fake.calls == []
+
+
+def test_restore_rootfs_gb_with_force_raises_and_never_runs(fake):
+    # force=1 destroys the old volumes first; a failed extraction would lose the guest
+    with pytest.raises(ValueError):
+        cli.restore(CT, CT_VOLID, 100, "zfs", force=True, rootfs_gb=4)
+    assert fake.calls == []
+
+
+def test_restore_rootfs_gb_on_a_vm_backup_raises_and_never_runs(fake):
+    with pytest.raises(ValueError):
+        cli.restore(VM, VM_VOLID, 109, "local-lvm", rootfs_gb=4)
+    assert fake.calls == []