cv4pve-cli-tui/src/cv4pve_tui/app.py

"""Textual app for managing Proxmox guests through cv4pve-cli."""

import subprocess
import sys
from datetime import datetime

from rich.text import Text
from textual import work
from textual.app import App, ComposeResult, SuspendNotSupported
from textual.binding import Binding
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.screen import ModalScreen
from textual.widgets import (
    Button,
    Checkbox,
    DataTable,
    Footer,
    Header,
    Input,
    Label,
    Log,
    OptionList,
    Select,
    Static,
)
from textual.widgets.option_list import Option

from cv4pve_tui import cli


def _message(exc: Exception) -> str:
    return (
        exc.summary if isinstance(exc, cli.CliError) else f"{type(exc).__name__}: {exc}"
    )


class DialogScreen(ModalScreen):
    """A centred, titled box (#dialog) with labelled rows and right-aligned buttons."""

    # every widget in a row gets an explicit width: Horizontal alone squeezes Select to 1
    # and stretches Input to 100 %, pushing the rest off-screen
    DEFAULT_CSS = """
    DialogScreen { align: center middle; }
    DialogScreen #dialog {
        width: 80; max-width: 100%; height: auto; max-height: 100%; overflow-y: auto;
        border: round $accent; border-title-style: bold; padding: 0 1;
        background: $surface;
    }
    DialogScreen .row, DialogScreen .buttons { height: auto; }
    DialogScreen .buttons { align-horizontal: right; }
    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 .sub {
        height: auto; border: round $accent; border-title-style: bold; padding: 0 1;
    }
    """

    def box(self, title: str) -> Vertical:
        """The #dialog container; use as `with self.box(title):` inside compose."""
        box = Vertical(id="dialog")
        # titles carry guest names: never parse them as markup
        box.border_title = Text(title)
        return box


class ConfirmScreen(DialogScreen):
    BINDINGS = [("escape", "dismiss(False)", "No")]  # noqa: RUF012 - Textual idiom

    def __init__(self, message: str, title: str = "Confirm") -> None:
        super().__init__()
        self.message, self.title_text = message, title

    def compose(self) -> ComposeResult:
        with self.box(self.title_text):
            yield Static(self.message, markup=False)
            with Horizontal(classes="buttons"):
                yield Button("Yes", id="yes", variant="error")
                yield Button("No", id="no")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        self.dismiss(event.button.id == "yes")


class EditScreen(DialogScreen):
    """Dismisses with a dict of validated values, or None when cancelled."""

    BINDINGS = [("escape", "dismiss(None)", "Cancel")]  # noqa: RUF012 - Textual idiom

    def __init__(self, guest: dict, config: dict, disks: dict[str, str]) -> None:
        super().__init__()
        self.guest, self.config, self.disks = guest, config, disks

    def compose(self) -> ComposeResult:
        with self.box(f"Edit resources: {self.guest['name']} ({self.guest['vmid']})"):
            yield Static("Empty = unchanged")
            with Horizontal(classes="row"):
                yield Label("Cores")
                yield Input(
                    placeholder=f"now {self.config.get('cores', '?')}", id="cores"
                )
            with Horizontal(classes="row"):
                yield Label("Memory (MiB)")
                yield Input(
                    placeholder=f"now {self.config.get('memory', '?')}", id="memory"
                )
            with Horizontal(classes="row"):
                yield Label("Disk")
                yield Select(
                    [(f"{k} ({v})", k) for k, v in self.disks.items()],
                    prompt="disk",
                    id="disk",
                )
            with Horizontal(classes="row"):
                yield Label("Grow by (GiB)")
                yield Input(placeholder="grow by GiB", id="grow")
            with Horizontal(classes="buttons"):
                yield Button("Apply", id="apply", variant="primary")
                yield Button("Cancel", id="cancel")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id == "cancel":
            self.dismiss(None)
            return
        try:
            values = {name: self._number(name) for name in ("cores", "memory", "grow")}
        except ValueError as exc:
            self.app.notify(str(exc), severity="error")
            return
        disk = self.query_one("#disk", Select).value
        if values["grow"] is not None and disk is Select.NULL:
            self.app.notify("Choose a disk to grow", severity="error")
            return
        values["disk"] = None if disk is Select.NULL else disk
        self.dismiss(values)

    def _number(self, name: str) -> int | None:
        text = self.query_one(f"#{name}", Input).value.strip()
        if not text:
            return None
        if not text.isdecimal() or int(text) <= 0:
            raise ValueError(f"{name} must be a positive integer")
        return int(text)


class ContextScreen(DialogScreen):
    """Dismisses with the chosen context name, or None when cancelled."""

    BINDINGS = [("escape", "dismiss(None)", "Cancel")]  # noqa: RUF012 - Textual idiom

    def __init__(self, contexts: list[dict]) -> None:
        super().__init__()
        self.contexts = contexts

    def compose(self) -> ComposeResult:
        with self.box("Switch context"):
            yield OptionList(
                *(
                    Option(
                        f"{'*' if c['current'] else ' '} {c['name']}  {c['host']}",
                        id=c["name"],
                    )
                    for c in self.contexts
                ),
                id="contexts",
            )

    def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None:
        self.dismiss(event.option.id)


class BackupScreen(DialogScreen):
    """Dismisses with the chosen storage, or None when cancelled."""

    BINDINGS = [("escape", "dismiss(None)", "Cancel")]  # noqa: RUF012 - Textual idiom

    def __init__(self, guest: dict, storages: list[str], chosen: str) -> None:
        super().__init__()
        self.guest, self.storages, self.chosen = guest, storages, chosen

    def compose(self) -> ComposeResult:
        with self.box(f"Back up {self.guest['name']} ({self.guest['vmid']})"):
            yield Static(
                f"Back up {self.guest['name']} ({self.guest['vmid']}) now: mode snapshot, zstd",
                markup=False,
            )
            with Horizontal(classes="row"):
                yield Label("Storage")
                yield Select(
                    [(s, s) for s in self.storages],
                    value=self.chosen,
                    allow_blank=False,
                    id="storage",
                )
            with Horizontal(classes="buttons"):
                yield Button("Yes", id="yes", variant="primary")
                yield Button("No", id="no")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id == "yes":
            self.dismiss(self.query_one("#storage", Select).value)
        else:
            self.dismiss(None)


class SnapshotScreen(DialogScreen):
    """Dismisses with {name, description, vmstate}, or None when cancelled."""

    BINDINGS = [("escape", "dismiss(None)", "Cancel")]  # noqa: RUF012 - Textual idiom

    def __init__(self, guest: dict) -> None:
        super().__init__()
        self.guest = guest

    def compose(self) -> ComposeResult:
        with self.box(f"Take snapshot: {self.guest['name']} ({self.guest['vmid']})"):
            with Horizontal(classes="row"):
                yield Label("Name")
                yield Input(datetime.now().strftime("tui_%Y%m%d_%H%M"), id="snapname")
            with Horizontal(classes="row"):
                yield Label("Description")
                yield Input(placeholder="optional", id="snapdesc")
            if cli.kind(self.guest) == "vm":
                yield Checkbox("Include RAM (vmstate)", id="vmstate")
            with Horizontal(classes="buttons"):
                yield Button("Yes", id="yes", variant="primary")
                yield Button("No", id="no")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id != "yes":
            self.dismiss(None)
            return
        try:
            name = cli.valid_snapname(self.query_one("#snapname", Input).value)
            description = cli.valid_description(
                self.query_one("#snapdesc", Input).value.strip()
            )
        except ValueError as exc:
            self.app.notify(str(exc), severity="error")
            return
        vmstate = next(iter(self.query("#vmstate").results(Checkbox)), None)
        self.dismiss(
            {
                "name": name,
                "description": description,
                "vmstate": bool(vmstate and vmstate.value),
            }
        )


_IPV4_MODES = [("DHCP", "dhcp"), ("Static", "static"), ("None", "none")]
_IPV6_MODES = [("SLAAC", "auto"), *_IPV4_MODES]


class NetScreen(DialogScreen):
    """Edit bridge, VLAN, firewall and IP of one interface; confirms, then applies."""

    BINDINGS = [("escape", "dismiss", "Cancel")]  # noqa: RUF012 - Textual idiom

    def __init__(
        self, guest: dict, config: dict, bridges: list[str], running: bool
    ) -> None:
        super().__init__()
        self.guest, self.config, self.running = guest, config, running
        self.nets = cli.nets(config)
        self.is_vm = cli.kind(guest) == "vm"
        self.ip_enabled = not self.is_vm or cli.has_cloudinit(config)
        # a bridge the node no longer lists must stay selectable, or Select rejects the prefill
        used = {n.get("bridge") for n in self.nets.values()} - {None}
        self.bridges = sorted(set(bridges) | used)

    def compose(self) -> ComposeResult:
        with self.box(f"Edit network: {self.guest['name']} ({self.guest['vmid']})"):
            with Horizontal(classes="row"):
                yield Label("Interface")
                yield Select(
                    [(k, k) for k in self.nets],
                    value=next(iter(self.nets)),
                    allow_blank=False,
                    id="iface",
                )
            with Horizontal(classes="row"):
                yield Label("Bridge")
                yield Select(
                    [(b, b) for b in self.bridges], allow_blank=False, id="bridge"
                )
            with Horizontal(classes="row"):
                yield Label("VLAN")
                yield Input(placeholder="VLAN tag (empty = none)", id="tag")
            with Horizontal(classes="row"):
                yield Label("Firewall")
                yield Checkbox("", id="firewall")
            if not self.ip_enabled:
                yield Static(
                    "IP is configured inside the VM (no cloud-init drive)", id="ip_note"
                )
            for fam, modes in (("ipv4", _IPV4_MODES), ("ipv6", _IPV6_MODES)):
                with Vertical(id=f"{fam}_section", classes="sub") as section:
                    section.border_title = fam.replace("ip", "IP")
                    with Horizontal(classes="row"):
                        yield Select(modes, allow_blank=False, id=fam, classes="mode")
                        yield Input(placeholder="address/prefix", id=f"{fam}_cidr")
                        yield Input(placeholder="gateway", id=f"{fam}_gw")
            with Horizontal(classes="buttons"):
                yield Button("Apply", id="apply", variant="primary")
                yield Button("Cancel", id="cancel")

    def on_mount(self) -> None:
        for wid in ("ipv4", "ipv4_cidr", "ipv4_gw", "ipv6", "ipv6_cidr", "ipv6_gw"):
            self.query_one(f"#{wid}").disabled = not self.ip_enabled
        self._prefill()

    def on_select_changed(self, event: Select.Changed) -> None:
        if event.select.id == "iface":
            self._prefill()

    def _ip_cfg(self, iface: str) -> dict:
        """The dict holding the IP keys: the VM's ipconfigN, or the LXC net itself."""
        if self.is_vm:
            return cli.parse_net(str(self.config.get(f"ipconfig{iface[3:]}", "")))
        return self.nets[iface]

    def _prefill(self) -> None:
        iface = self.query_one("#iface", Select).value
        net = self.nets[iface]
        self.query_one("#bridge", Select).value = net.get("bridge", self.bridges[0])
        self.query_one("#tag", Input).value = net.get("tag", "")
        self.query_one("#firewall", Checkbox).value = net.get("firewall") == "1"
        for fam, family in (("ipv4", 4), ("ipv6", 6)):
            mode, cidr, gw = cli.ip_mode(self._ip_cfg(iface), family)
            self.query_one(f"#{fam}", Select).value = mode
            self.query_one(f"#{fam}_cidr", Input).value = cidr
            self.query_one(f"#{fam}_gw", Input).value = gw

    def _changes(self) -> tuple[dict, dict]:
        """(changes, old) as raw netN/ipconfigN strings, None = absent; ValueError on bad input."""
        iface = self.query_one("#iface", Select).value
        text = self.query_one("#tag", Input).value.strip()
        if text and not text.isdecimal():
            raise ValueError("VLAN tag must be 1-4094")
        current = self.nets[iface]
        link = (
            self.query_one("#bridge", Select).value,
            int(text) if text else None,
            self.query_one("#firewall", Checkbox).value,
        )
        was = (current.get("bridge"), int(current["tag"]) if current.get("tag") else None,
               current.get("firewall") == "1")  # fmt: skip
        # untouched fields are not rewritten (e.g. firewall=0 stays as it is)
        net = cli.apply_link(current, *link) if link != was else dict(current)
        ip_key = f"ipconfig{iface[3:]}"
        ip_old = self._ip_cfg(iface) if self.is_vm else {}
        ip = ip_old if self.is_vm else net  # the dict that holds the IP keys
        if self.ip_enabled:
            for fam, family in (("ipv4", 4), ("ipv6", 6)):
                mode, cidr, gw = (
                    self.query_one(f"#{fam}{s}").value for s in ("", "_cidr", "_gw")
                )
                wanted = (mode, cidr.strip(), gw.strip())
                if wanted != cli.ip_mode(
                    ip, family
                ):  # untouched: e.g. ip6=manual stays
                    ip = cli.apply_ip(ip, family, *wanted)
        if not self.is_vm:
            net = ip
        changes, old = {}, {}
        if net != self.nets[iface]:
            changes[iface] = cli.format_net(net)
            old[iface] = str(self.config[iface])
        if self.is_vm and ip != ip_old:
            changes[ip_key] = cli.format_net(ip) or None
            old[ip_key] = self.config.get(ip_key)
        return changes, old

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id == "cancel":
            self.dismiss()
            return
        try:
            changes, old = self._changes()
        except ValueError as exc:
            self.app.notify(str(exc), severity="error")
            return
        if not changes:
            self.app.notify("No change")
            return
        lines = [
            f"{k}\n  from {old[k] or '(not set)'}\n  to   {v or '(removed)'}"
            for k, v in changes.items()
        ]
        notes = []
        if self.running and any(k.startswith("net") for k in changes):
            notes.append(
                "Link changes apply immediately and can cut the guest's network."
            )
        if (
            self.running
            and not self.is_vm
            and any(_ip_changed(old[k], v) for k, v in changes.items())
        ):  # seen live: static -> dhcp left the CT without IPv4 until it was rebooted
            notes.append(
                "IP changes on a running container may need a reboot of the container "
                "(e.g. DHCP is not restarted)."
            )
        if any(k.startswith("ipconfig") for k in changes):
            notes.append("IP settings (cloud-init) apply at next boot.")
        who = f"{self.guest['name']} ({self.guest['vmid']})"
        self.app.push_screen(
            ConfirmScreen(
                "\n".join([f"Change network of {who}?", *lines, *notes]),
                "Network change",
            ),
            lambda ok: (
                self.app.start_net_change(self.guest, changes, old, self)
                if ok
                else None
            ),
        )


class TaskLogScreen(DialogScreen):
    """The log of one task, as plain text (task output is never parsed as markup)."""

    BINDINGS = [("escape", "dismiss", "Close")]  # noqa: RUF012 - Textual idiom
    DEFAULT_CSS = """
    TaskLogScreen #dialog { width: 120; height: 90%; }
    TaskLogScreen Log { height: 1fr; }
    """

    def __init__(self, entry: dict) -> None:
        super().__init__()
        self.entry = entry

    def compose(self) -> ComposeResult:
        t = self.entry
        with self.box(f"Log: {t.get('type', '')} {t.get('id') or t.get('node', '')}"):
            yield Log(id="tasklog")

    def on_mount(self) -> None:
        self._load_worker()

    @work(thread=True)
    def _load_worker(self) -> None:
        try:
            text = cli.task_log(self.entry["upid"])
        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
        self.app.call_from_thread(self.query_one("#tasklog", Log).write, text)


def _duration(task: dict) -> str:
    end, start = task.get("endtime"), task.get("starttime")
    if not end:
        return "running"
    if not start:
        return "-"  # finished, but the start is unknown
    secs = max(0, int(end - start))
    if secs >= 3600:
        return f"{secs // 3600}h {secs % 3600 // 60}m"
    return f"{secs // 60}m {secs % 60}s" if secs >= 60 else f"{secs}s"


class TasksScreen(ModalScreen):
    """Recent cluster tasks, or those of one guest (vmid); Enter shows a task's log."""

    BINDINGS = [  # noqa: RUF012 - Textual idiom
        ("escape", "dismiss", "Close"),
        ("g", "reload", "Refresh"),
    ]
    DEFAULT_CSS = """
    TasksScreen { align: center middle; }
    TasksScreen #tasks_section {
        width: 100%; height: 100%; border: round $accent; border-title-style: bold;
        padding: 0 1; background: $surface;
    }
    TasksScreen DataTable { height: 1fr; }
    """

    def __init__(self, guest: dict | None = None) -> None:
        super().__init__()
        self.guest = guest

    def compose(self) -> ComposeResult:
        title = "Tasks"
        if self.guest:
            title += f": {self.guest['name']} ({self.guest['vmid']})"
        with Vertical(id="tasks_section") as section:
            section.border_title = Text(title)  # guest names: never markup
            yield DataTable(id="tasks", cursor_type="row")
            yield Static("", id="tasks_note", markup=False)
        yield Footer()

    def on_mount(self) -> None:
        self.rows: list[dict] = []
        self.query_one("#tasks", DataTable).add_columns(
            "", "Start", "Duration", "Node", "Type", "VMID", "User", "Status"
        )
        self.action_reload()

    def action_reload(self) -> None:
        self._load_worker()

    @work(thread=True)
    def _load_worker(self) -> None:
        try:
            tasks = cli.tasks()
        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
        if self.guest:
            tasks = [t for t in tasks if t.get("id") == str(self.guest["vmid"])]
        self.app.call_from_thread(self._fill, tasks)

    def _fill(self, tasks: list[dict]) -> None:
        self.rows = tasks
        table = self.query_one("#tasks", DataTable)
        table.clear()
        for t in tasks:
            # running = no endtime yet; a finished task without a status text is "unknown"
            status = "running" if not t.get("endtime") else t.get("status") or "unknown"
            _add_plain(
                table,
                "!" if status not in ("OK", "running") else "",
                _when(t.get("starttime")),
                _duration(t),
                t.get("node", ""),
                t.get("type", ""),
                t.get("id") or "-",
                t.get("user", ""),
                status,
            )
        self.query_one("#tasks_note", Static).update("" if tasks else "No tasks")

    def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
        self.app.push_screen(TaskLogScreen(self.rows[event.cursor_row]))


def _collect_updates() -> tuple[list[dict], list[str]]:
    """Runs in a worker thread: (rows over all online nodes, one note per failed node)."""
    rows, notes = [], []
    for node in cli.nodes():
        try:
            rows += [{**p, "node": node} for p in cli.package_updates(node)]
        except Exception as exc:  # noqa: BLE001 - a failing node must not hide the others
            notes.append(f"{node}: unavailable: {_message(exc)}")
    return rows, notes


class UpdatesScreen(ModalScreen):
    """Pending package updates of every online node (read-only)."""

    BINDINGS = [  # noqa: RUF012 - Textual idiom
        ("escape", "dismiss", "Close"),
        ("g", "reload", "Refresh"),
    ]
    DEFAULT_CSS = """
    UpdatesScreen { align: center middle; }
    UpdatesScreen #updates_section {
        width: 100%; height: 100%; border: round $accent; border-title-style: bold;
        padding: 0 1; background: $surface;
    }
    UpdatesScreen DataTable { height: 1fr; }
    """

    def compose(self) -> ComposeResult:
        with Vertical(id="updates_section") as section:
            section.border_title = "Package updates"
            yield DataTable(id="updates", cursor_type="row")
            yield Static("", id="updates_note", markup=False)
        yield Footer()

    def on_mount(self) -> None:
        self.query_one("#updates", DataTable).add_columns(
            "Node", "Package", "Current", "New", "Origin", "Priority"
        )
        self.action_reload()

    def action_reload(self) -> None:
        self._load_worker()

    @work(thread=True)
    def _load_worker(self) -> None:
        gen = self.app.generation
        try:
            rows, notes = _collect_updates()
        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
        self.app.call_from_thread(self._fill, rows, notes)
        self.app.call_from_thread(
            self.app._set_count, None if notes else len(rows), gen
        )

    def _fill(self, rows: list[dict], notes: list[str]) -> None:
        table = self.query_one("#updates", DataTable)
        table.clear()
        for p in rows:
            _add_plain(
                table,
                p["node"],
                p.get("Package", ""),
                p.get("OldVersion", ""),
                p.get("Version", ""),
                p.get("Origin", ""),
                p.get("Priority", ""),
            )
        lines = [] if rows else ["No pending updates"]
        lines += notes
        lines.append("Lists are as of each node's last apt update; it is not run here.")
        self.query_one("#updates_note", Static).update("\n".join(lines))


class InfoScreen(ModalScreen):
    """Backups, backup job and network of one guest (read-only), plus backup now."""

    BINDINGS = [  # noqa: RUF012 - Textual idiom
        ("escape", "dismiss", "Close"),
        ("g", "reload", "Refresh"),
        ("b", "backup", "Backup now"),
        ("a", "toggle_all", "All backups"),
        ("n", "net", "Edit network"),
        ("u", "undo", "Undo network"),
        ("e", "edit", "Edit resources"),
        ("p", "snapshot", "Take snapshot"),
        ("R", "rollback", "Roll back"),
        ("D", "delete_snapshot", "Delete snapshot"),
        ("l", "tasks", "Tasks"),
    ]
    DEFAULT_CSS = """
    InfoScreen DataTable { height: auto; max-height: 12; }
    InfoScreen .section {
        height: auto; border: round $accent; border-title-style: bold;
        padding: 0 1; margin-bottom: 1;
    }
    InfoScreen .label { text-style: bold; margin-top: 1; }
    InfoScreen .label.first { margin-top: 0; }
    """

    def __init__(self, guest: dict) -> None:
        super().__init__()
        self.guest = guest
        self.storages: list[str] = []
        self.jobs: list[dict] = []
        self.backups: list[dict] = []
        self.show_all = False
        self.config: dict = {}
        self.running = False
        self.snaps: list[dict] = []

    def compose(self) -> ComposeResult:
        with VerticalScroll(id="info_body"):  # four sections do not fit short terminals
            yield Static(f"{self.guest['name']} ({self.guest['vmid']})", markup=False)
            with Vertical(id="resources_section", classes="section") as res:
                res.border_title = "Resources"
                yield DataTable(id="resources", cursor_type="none")
            with Vertical(id="backup_section", classes="section") as backup:
                backup.border_title = "Backups"
                yield Static("", id="jobs", markup=False)
                yield DataTable(id="backups")
                yield Static("", id="hidden", markup=False)
            with Vertical(id="snapshots_section", classes="section") as snaps:
                snaps.border_title = "Snapshots"
                yield DataTable(id="snapshots", cursor_type="row")
                yield Static("", id="snap_note", markup=False)
            with Vertical(id="network_section", classes="section") as network:
                network.border_title = "Network"
                yield Static("Configured", classes="label first")
                yield DataTable(id="nets")
                yield Static("In use", classes="label")
                yield Static("", id="unavailable", markup=False)
                yield DataTable(id="addresses")
        yield Footer()

    def on_mount(self) -> None:
        self.query_one("#resources", DataTable).add_columns(
            "Resource", "Configured", "In use"
        )
        self.query_one("#backups", DataTable).add_columns(
            "Date", "Storage", "Size (GiB)", "Format", "Notes"
        )
        self.query_one("#snapshots", DataTable).add_columns(
            "Name",
            "Date",
            "Description",
            *(["RAM"] if cli.kind(self.guest) == "vm" else []),
        )
        self.query_one("#nets", DataTable).add_columns(
            "Iface", "Bridge", "VLAN", "MAC", "IPv4", "IPv6", "Firewall"
        )
        self.query_one("#addresses", DataTable).add_columns(
            "Interface", "MAC", "IPv4", "IPv6"
        )
        self.action_reload()

    def action_reload(self) -> None:
        self._load_worker()

    def action_tasks(self) -> None:
        self.app.push_screen(TasksScreen(self.guest))

    @work(thread=True)
    def _load_worker(self) -> None:
        self.load()

    def load(self) -> None:
        """Runs in a worker thread."""
        guest = self.guest
        try:
            config = cli.guest_config(guest)
            nets = cli.nets(config)
            storages = cli.backup_storages(guest)
            jobs = cli.backup_jobs()
            backups = cli.backups(guest, storages)
        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
        addresses, unavailable, usage = [], "", {}
        try:  # live status: the guest table's status may be stale
            usage = cli.current(guest)
            running = usage.get("status") == "running"
        except Exception as exc:  # noqa: BLE001 - shown inline, the rest of the screen still works
            running, unavailable = False, f"unavailable: {_message(exc)}"
        if running:
            try:
                addresses = cli.interfaces(guest)
            except Exception as exc:  # noqa: BLE001 - shown inline, the rest of the screen still works
                unavailable = f"unavailable: {_message(exc)}"
        snaps, snap_note = [], ""
        try:
            snaps = cli.snapshots(guest)
        except Exception as exc:  # noqa: BLE001 - shown inline, the rest of the screen still works
            snap_note = f"unavailable: {_message(exc)}"
        self.running = running
        self.app.call_from_thread(
            self._show,
            (config, nets, storages, jobs, backups, addresses, unavailable, usage),
            snaps,
            snap_note,
        )

    def _show(self, loaded: tuple, snaps: list[dict], snap_note: str) -> None:
        if not self.is_attached:  # closed while loading
            return
        try:
            self._fill_screen(*loaded)
            self._show_snapshots(snaps, snap_note)
        except Exception as exc:  # noqa: BLE001 - odd API data must not kill the app
            self.app.notify(f"info: {_message(exc)}", severity="error")

    def _fill_screen(
        self, config, nets, storages, jobs, backups, addresses, unavailable, usage
    ) -> None:
        self.storages, self.jobs, self.config = storages, jobs, config
        covering = cli.covering_jobs(jobs, self.guest)
        self.query_one("#jobs", Static).update(
            "; ".join(_job_text(j) for j in covering)
            or "No backup job covers this guest"
        )
        table = self.query_one("#resources", DataTable).clear()
        in_use = _in_use(usage, cli.kind(self.guest)) if self.running else {}
        for name, value in _resources(config, cli.kind(self.guest)):
            _add_plain(table, name, value, in_use.get(name, "-"))
        self.backups = backups
        self._show_backups()
        table = self.query_one("#nets", DataTable).clear()
        for name, net in nets.items():
            if cli.kind(self.guest) == "vm":  # IP settings are cloud-init's ipconfigN
                mac = next(iter(net.values()), "-")
                ip = cli.parse_net(str(config.get(f"ipconfig{name[3:]}", "")))
            else:
                mac, ip = net.get("hwaddr", "-"), net
            _add_plain(
                table,
                name,
                net.get("bridge", "-"),
                net.get("tag", "-"),
                mac,
                ip.get("ip", "-"),
                ip.get("ip6", "-"),
                "yes" if net.get("firewall") == "1" else "no",
            )
        note = self.query_one("#unavailable", Static)
        note.update(unavailable)
        note.display = bool(unavailable)  # no empty line when the addresses are known
        table = self.query_one("#addresses", DataTable).clear()
        for i in addresses:
            _add_plain(
                table,
                i["name"],
                i["mac"] or "-",
                ", ".join(i["ipv4"]),
                ", ".join(i["ipv6"]),
            )

    def _show_snapshots(self, snaps: list[dict], note: str) -> None:
        self.snaps = snaps
        table = self.query_one("#snapshots", DataTable).clear()
        vm = cli.kind(self.guest) == "vm"
        for s in snaps:
            _add_plain(
                table,
                s["name"],
                _when(s.get("snaptime")),
                s.get("description") or "",
                *(["yes" if s.get("vmstate") else "no"] if vm else []),
            )
        self.query_one("#snap_note", Static).update(
            note or ("" if snaps else "No snapshots")
        )

    def _show_backups(self) -> None:
        name = self.guest["name"]
        shown = [
            b for b in self.backups if self.show_all or not _earlier_guest(b, name)
        ]
        table = self.query_one("#backups", DataTable).clear()
        for b in shown:
            _add_plain(
                table,
                datetime.fromtimestamp(b["ctime"]).strftime("%Y-%m-%d %H:%M")
                if b.get("ctime")
                else "-",
                b["storage"],
                f"{b['size'] / 2**30:.1f}" if b.get("size") is not None else "-",
                b.get("format") or "-",
                (b.get("notes") or "")
                + (" (earlier guest)" if _earlier_guest(b, name) else ""),
            )
        hidden = len(self.backups) - len(shown)
        self.query_one("#hidden", Static).update(
            f"{hidden} backup(s) of an earlier guest with this VMID hidden - press a to show"
            if hidden
            else ""
        )

    def action_edit(self) -> None:
        self.app.edit_guest(self.guest)

    def action_toggle_all(self) -> None:
        self.show_all = not self.show_all
        self._show_backups()

    def _refuse_if_busy(self) -> bool:
        busy = self.guest["vmid"] in self.app.busy
        if busy:
            self.app.notify(
                f"{self.guest['name']} ({self.guest['vmid']}) is busy",
                severity="warning",
            )
        return busy

    def action_net(self) -> None:
        if cli.nets(self.config) and not self._refuse_if_busy():
            self._open_net()

    @work(thread=True)
    def _open_net(self) -> None:
        try:
            bridges = cli.bridges(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
        self.app.call_from_thread(
            self.app.push_screen,
            NetScreen(self.guest, self.config, bridges, self.running),
        )

    def action_undo(self) -> None:
        old = self.app.undo.get(self.guest["vmid"])
        if old is None:
            self.app.notify("No network change to undo")
            return
        if self._refuse_if_busy():
            return
        lines = [f"{k} = {v or '(remove)'}" for k, v in old.items()]
        self.app.push_screen(
            ConfirmScreen(
                "\n".join(["Restore the previous network settings?", *lines]),
                "Undo network change",
            ),
            lambda ok: self.app.start_net_change(self.guest, old, None) if ok else None,
        )

    def _snapshot_under_cursor(self) -> dict | None:
        """The snapshot at the cursor; None (with a notify) when there is none or the guest is busy."""
        row = self.query_one("#snapshots", DataTable).cursor_row
        if not 0 <= row < len(self.snaps):
            self.app.notify("No snapshots")
            return None
        return None if self._refuse_if_busy() else self.snaps[row]

    def action_snapshot(self) -> None:
        if self._refuse_if_busy():
            return
        self.app.push_screen(SnapshotScreen(self.guest), self._take_snapshot)

    def _take_snapshot(self, values: dict | None) -> None:
        if values:
            self.app.start_job(
                self.guest,
                "Snapshot",
                lambda: cli.take_snapshot(
                    self.guest,
                    values["name"],
                    values["description"],
                    values["vmstate"],
                ),
            )

    def action_rollback(self) -> None:
        snap = self._snapshot_under_cursor()
        if snap is None:
            return
        text = (
            f"Roll back to snapshot {snap['name']} ({_when(snap.get('snaptime'))})? "
            "Everything changed since then is lost. "
            "If the guest is running, it is started again afterwards."
        )

        def job() -> None:
            # status read right before the rollback (not at screen load); if it cannot be
            # read the rollback does not run, rather than leave a running guest stopped
            start = cli.status(self.guest) == "running"
            cli.rollback_snapshot(self.guest, snap["name"], start)

        self.app.push_screen(
            ConfirmScreen(text, "Roll back"),
            lambda ok: (
                self.app.start_job(
                    self.guest,
                    "Rollback",
                    job,
                )
                if ok
                else None
            ),
        )

    def action_delete_snapshot(self) -> None:
        snap = self._snapshot_under_cursor()
        if snap is None:
            return
        self.app.push_screen(
            ConfirmScreen(f"Delete snapshot {snap['name']}?", "Delete snapshot"),
            lambda ok: (
                self.app.start_job(
                    self.guest,
                    "Delete snapshot",
                    lambda: cli.delete_snapshot(self.guest, snap["name"]),
                )
                if ok
                else None
            ),
        )

    def action_backup(self) -> None:
        if not self.storages:
            self.app.notify("No backup storage available", severity="warning")
            return
        chosen = cli.job_storage(self.jobs, self.guest)
        if chosen not in self.storages:
            chosen = self.storages[0]
        self.app.push_screen(
            BackupScreen(self.guest, self.storages, chosen),
            lambda storage: (
                self.app.start_backup(self.guest, storage) if storage else None
            ),
        )


def _when(epoch: int | None) -> str:
    return datetime.fromtimestamp(epoch).strftime("%Y-%m-%d %H:%M") if epoch else "-"


def _ip_changed(old: str | None, new: str | None) -> bool:
    before, after = cli.parse_net(old or ""), cli.parse_net(new or "")
    return any(before.get(k) != after.get(k) for k in ("ip", "gw", "ip6", "gw6"))


def _earlier_guest(backup: dict, name: str) -> bool:
    """VMIDs are reused: notes (the job's {{guestname}}) naming another guest mean a deleted one."""
    notes = backup.get("notes") or ""
    return bool(notes) and notes != name and not notes.startswith(f"{name} ")


def _job_text(job: dict) -> str:
    keep = (job.get("prune-backups") or {}).get("keep-last")
    return f"{job['schedule']} to {job['storage']}" + (
        f", keep-last {keep}" if keep else ""
    )


class GuestTable(DataTable):
    """The guest table; shows its Enter (row selected -> info screen) in the footer."""

    BINDINGS = [Binding("enter", "select_cursor", "Info")]  # noqa: RUF012 - Textual idiom


class GuestApp(App):
    BINDINGS = [  # noqa: RUF012 - Textual idiom
        ("s", "act('start')", "Start"),
        ("S", "act('shutdown')", "Shutdown"),
        ("x", "act('stop')", "Stop"),
        ("r", "act('reboot')", "Reboot"),
        ("t", "shell", "Shell"),
        ("g", "reload", "Refresh"),
        ("c", "context", "Context"),
        ("l", "tasks", "Tasks"),
        ("U", "updates", "Updates"),
        ("q", "quit", "Quit"),
    ]

    def __init__(self) -> None:
        super().__init__()
        self.guests: dict[str, dict] = {}
        self.busy: set[int] = set()
        self.undo: dict[int, dict] = {}  # vmid -> values before the last network change
        # bumped on every context switch; replies started earlier come from the old cluster
        self.generation = 0
        self.switching = False
        self.context_name = ""
        self.unprotected: set[int] = set()  # VMIDs no backup job covers
        self.package_count: int | None = None  # pending package updates; None = unknown

    CSS = """
    #guests_section {
        border: round $accent; border-title-style: bold; padding: 0 1;
    }
    #guests_section DataTable { height: 1fr; }
    """

    def compose(self) -> ComposeResult:
        yield Header()
        with Vertical(id="guests_section") as section:
            section.border_title = "Guests"
            yield GuestTable(id="guests", cursor_type="row")
        yield Footer()

    def on_mount(self) -> None:
        # one place for every sub_title change: mount, switch and failed switch
        self.watch(self, "sub_title", self._show_context_frame)
        self._check_version()
        table = self.query_one("#guests", DataTable)
        table.add_columns(
            "",  # "!" = no backup job covers the guest
            "VMID",
            "Name",
            "Type",
            "Node",
            "Status",
            "Cores",
            "Memory (MiB)",
            "Disk (GiB)",
            "CPU %",
            "Mem %",
        )
        self._show_context()
        self.action_reload()
        self._count_worker()

    def action_tasks(self) -> None:
        self.push_screen(TasksScreen())

    def action_updates(self) -> None:
        self.push_screen(UpdatesScreen())

    @work(thread=True)
    def _count_worker(self) -> None:
        self._refresh_count()

    def _refresh_count(self) -> None:
        """Runs in a worker thread: any failure leaves the count unknown, silently."""
        gen = self.generation
        try:
            rows, notes = _collect_updates()
            count = None if notes else len(rows)
        except Exception:  # noqa: BLE001 - the count is optional; no notify on failure
            count = None
        self.call_from_thread(self._set_count, count, gen)

    def _set_count(self, count: int | None, gen: int) -> None:
        if gen != self.generation:  # fetched before a context switch: old cluster
            return
        self.package_count = count
        self._update_frame()

    @work(thread=True)
    def _check_version(self) -> None:
        try:
            found = cli.version()
        except Exception as exc:  # noqa: BLE001 - every later call fails too; say why once
            self.call_from_thread(
                self.notify,
                f"cv4pve-cli is not runnable: {_message(exc)}",
                severity="error",
            )
            return
        if found != cli.TESTED_VERSION:
            self.call_from_thread(
                self.notify,
                f"cv4pve-cli {found} found; this app was tested with {cli.TESTED_VERSION}. "
                "Some screens may misread its output.",
                severity="warning",
            )

    def _show_context_frame(self, sub_title: str) -> None:
        self.context_name = sub_title.removeprefix("context: ")
        self._update_frame()

    def _update_frame(self) -> None:
        n = len(self.unprotected & {g["vmid"] for g in self.guests.values()})
        count = f" · {n} without backup job" if n else ""
        if self.package_count:
            n = self.package_count
            count += f" · {n} package update{'s' if n != 1 else ''}"
        self.query_one("#guests_section").border_subtitle = self.context_name + count

    @work(thread=True)
    def _show_context(self) -> None:
        self._read_context()

    def _read_context(self) -> None:
        """Runs in a worker thread: show the CLI's current context in the Header."""
        gen = self.generation
        try:
            name = next(c["name"] for c in cli.contexts() if c["current"])
        except StopIteration:
            name = "?"
        except Exception as exc:  # noqa: BLE001 - any failure must reach the user, not kill the app
            self.call_from_thread(self.notify, _message(exc), severity="error")
            name = "?"
        self.call_from_thread(self._set_context_title, name, gen)

    def _set_context_title(self, name: str, gen: int) -> None:
        if gen == self.generation:
            self.sub_title = f"context: {name}"

    def _selected(self) -> dict | None:
        table = self.query_one("#guests", DataTable)
        if not table.row_count:
            return None
        return self.guests[
            table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value
        ]

    def _fill(
        self, guests: list[dict], gen: int, unprotected: set[int] | None = None
    ) -> None:
        if gen != self.generation:  # listed before a context switch: old cluster
            return
        self.unprotected = unprotected or set()
        table = self.query_one("#guests", DataTable)
        current = self._selected()
        table.clear()
        self.guests = {str(g["vmid"]): g for g in guests}
        for g in guests:
            _add_plain(table,
                "!" if g["vmid"] in self.unprotected else "",
                g["vmid"], g["name"], g["type"], g["node"], g["status"],
                g.get("maxcpu", "-"), _scaled(g.get("maxmem"), 2**20), _scaled(g.get("maxdisk"), 2**30),
                *_percents(g),
                key=str(g["vmid"]),
            )  # fmt: skip
        if current and str(current["vmid"]) in self.guests:
            table.move_cursor(row=table.get_row_index(str(current["vmid"])))
        self._update_frame()

    def _reload(self, acted: dict | None = None) -> None:
        """Runs in a worker thread. `acted` gets its live status: the listing lags."""
        gen = self.generation
        try:
            guests = cli.list_guests()
            for g in guests:
                if acted and g["vmid"] == acted["vmid"]:
                    live = cli.current(g)  # status and usage of the acted guest lag too
                    g.update(
                        {
                            k: live[k]
                            for k in ("status", "cpu", "mem", "maxmem")
                            if k in live
                        }
                    )
        except Exception as exc:  # noqa: BLE001 - any failure must reach the user, not kill the app
            self.call_from_thread(self.notify, _message(exc), severity="error")
            return
        unprotected = None
        try:
            unprotected = cli.not_backed_up()
        except Exception as exc:  # noqa: BLE001 - the listing still works without it
            self.call_from_thread(
                self.notify,
                f"backup coverage unknown: {_message(exc)}",
                severity="warning",
            )
        self.call_from_thread(self._fill, guests, gen, unprotected)

    def action_reload(self) -> None:
        if self.switching:
            self.notify("Switching context; refresh after it ends", severity="warning")
            return
        self._reload_worker()

    @work(thread=True)
    def _reload_worker(self) -> None:
        self._reload()

    def action_context(self) -> None:
        if self.busy or self.switching:
            self.notify(
                "Busy (guest action or context switch); try again when it ends",
                severity="warning",
            )
            return
        self._open_context()

    @work(thread=True)
    def _open_context(self) -> None:
        try:
            contexts = cli.contexts()
        except Exception as exc:  # noqa: BLE001 - surface every failure, never kill the app
            self.call_from_thread(self.notify, _message(exc), severity="error")
            return
        current = next((c["name"] for c in contexts if c["current"]), None)
        self.call_from_thread(
            self.push_screen,
            ContextScreen(contexts),
            lambda name: self._context_chosen(name, current),
        )

    def _context_chosen(self, name: str | None, current: str | None) -> None:
        if name is None or name == current:
            return
        self.query_one(
            "#guests", DataTable
        ).clear()  # old cluster's guests: not actionable
        self.guests = {}
        self.undo.clear()  # VMIDs repeat across clusters: never undo onto another cluster
        self.package_count = None
        self._update_frame()
        self.generation += 1
        self.switching = True
        self._switch_context(name)

    @work(thread=True)
    def _switch_context(self, name: str) -> None:
        try:
            try:
                cli.use_context(name)
            except Exception as exc:  # noqa: BLE001 - surface every failure, never kill the app
                self.call_from_thread(self.notify, _message(exc), severity="error")
                self._read_context()
            else:
                self.call_from_thread(setattr, self, "sub_title", f"context: {name}")
            self._reload()
            self._refresh_count()
        finally:
            self.call_from_thread(setattr, self, "switching", False)

    def start_job(self, guest: dict, label: str, job) -> None:
        if guest["vmid"] in self.busy:
            self.notify(
                f"{guest['name']} ({guest['vmid']}) is busy", severity="warning"
            )
            return
        self.busy.add(guest["vmid"])
        self.notify(f"{label} {guest['name']} ({guest['vmid']}) ...")
        self._run_job(guest, label, job)

    @work(thread=True)
    def _run_job(self, guest: dict, label: str, job) -> None:
        try:
            job()
        except Exception as exc:  # noqa: BLE001 - surface every failure, never kill the app
            self.call_from_thread(
                self.notify, f"{label} failed: {_message(exc)}", severity="error"
            )
        else:
            self.call_from_thread(
                self.notify, f"{label} done: {guest['name']} ({guest['vmid']})"
            )
        finally:
            self._reload_infos(
                guest
            )  # e.g. edited cores show up in an open info screen
            # before un-busying, so no context switch can start mid-reload
            self._reload(acted=guest)
            self.busy.discard(guest["vmid"])

    def action_act(self, name: str) -> None:
        guest = self._selected()
        if guest is None:
            return
        run = lambda: cli.action(name, guest)
        if name == "start":
            self.start_job(guest, name, run)
            return
        who = f"{guest['name']} ({guest['vmid']})"
        text = f"{name.capitalize()} {who}?"
        if name == "stop":
            text = f"Hard power-off {who}? The guest is killed without shutting down."
        self.push_screen(
            ConfirmScreen(text, name.capitalize()),
            lambda ok: self.start_job(guest, name, run) if ok else None,
        )

    def edit_guest(self, guest: dict) -> None:
        """Opened from the info screen (key e)."""
        if guest["vmid"] in self.busy:
            self.notify(
                f"{guest['name']} ({guest['vmid']}) is busy", severity="warning"
            )
            return
        self._open_edit(guest)

    def action_info(self) -> None:
        guest = self._selected()
        if guest is not None:
            self.push_screen(InfoScreen(guest))

    def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
        if event.data_table.id == "guests":  # Enter on a guest row
            self.action_info()

    def start_backup(self, guest: dict, storage: str) -> None:
        who = f"{guest['name']} ({guest['vmid']})"
        if guest["vmid"] in self.busy:
            self.notify(f"{who} is busy", severity="warning")
            return
        self.busy.add(guest["vmid"])
        self.notify(f"Backup started: {who} to {storage}")
        self._run_backup(guest, storage)

    @work(thread=True)
    def _run_backup(self, guest: dict, storage: str) -> None:
        who = f"{guest['name']} ({guest['vmid']})"
        try:
            cli.backup_now(guest, storage)
        except Exception as exc:  # noqa: BLE001 - surface every failure, never kill the app
            self.call_from_thread(
                self.notify, f"Backup failed: {_message(exc)}", severity="error"
            )
        else:
            self.call_from_thread(self.notify, f"Backup done: {who}")
        finally:
            self._reload_infos(guest)
            self.busy.discard(guest["vmid"])

    def _reload_infos(self, guest: dict) -> None:
        """Runs in a worker thread: reload the guest's open InfoScreens (call while still busy)."""
        for screen in list(self.screen_stack):
            if isinstance(screen, InfoScreen) and screen.guest["vmid"] == guest["vmid"]:
                screen.load()

    def start_net_change(
        self, guest: dict, changes: dict, entry: dict | None, screen=None
    ) -> None:
        """`entry` becomes the guest's undo entry on success (None clears it).

        `screen` (the NetScreen) is closed on success and stays open on failure.
        """
        who = f"{guest['name']} ({guest['vmid']})"
        if guest["vmid"] in self.busy:
            self.notify(f"{who} is busy", severity="warning")
            return
        self.busy.add(guest["vmid"])
        self.notify(f"Network change started: {who}")
        self._run_net(guest, changes, entry, screen)

    @work(thread=True)
    def _run_net(self, guest: dict, changes: dict, entry: dict | None, screen) -> None:
        who = f"{guest['name']} ({guest['vmid']})"
        done = False
        try:
            cli.set_net(guest, changes)
        except Exception as exc:  # noqa: BLE001 - surface every failure, never kill the app
            self.call_from_thread(
                self.notify, f"Network change failed: {_message(exc)}", severity="error"
            )
        else:
            done = True
            if entry is None:
                self.undo.pop(guest["vmid"], None)
            else:
                self.undo[guest["vmid"]] = entry
            self.call_from_thread(self.notify, f"Network change done: {who}")
        finally:
            self._reload_infos(guest)
            if done and screen is not None:
                self.call_from_thread(self._close_net, screen)
            self.busy.discard(guest["vmid"])

    def _close_net(self, screen) -> None:
        if self.screen is screen:  # the user may have left it already
            screen.dismiss()

    def action_shell(self) -> None:
        guest = self._selected()
        if guest is None:
            return
        who = f"{guest['name']} ({guest['vmid']})"
        if guest["type"] != "lxc":
            self.notify(f"{who} is not a container", severity="warning")
            return
        if guest["status"] != "running":
            self.notify(f"{who} is not running", severity="warning")
            return
        if guest["vmid"] in self.busy:
            self.notify(f"{who} is busy", severity="warning")
            return
        cmd = ["ssh", "-t", f"root@{guest['node']}", "pct", "enter", str(guest["vmid"])]
        error = None
        try:
            with self.suspend():
                try:  # inside: Textual's suspend() skips the resume if its body raises
                    _reset_terminal()
                    code = subprocess.run(cmd, check=False).returncode
                except OSError as exc:
                    error = _message(exc)
        except SuspendNotSupported as exc:
            error = _message(exc)
        else:
            if (
                error is None and code == 255
            ):  # other codes are the shell's own last status
                error = "ssh failed (exit 255)"
        if error:
            self.notify(f"Shell in {who} failed: {error}", severity="error")
        self.action_reload()

    @work(thread=True)
    def _open_edit(self, guest: dict) -> None:
        try:
            config = cli.guest_config(guest)
            disks = cli.disks(config, cli.kind(guest))
        except Exception as exc:  # noqa: BLE001 - surface every failure, never kill the app
            self.call_from_thread(self.notify, _message(exc), severity="error")
            return
        self.call_from_thread(
            self.push_screen,
            EditScreen(guest, config, disks),
            lambda v: self._edit_done(guest, v),
        )

    def _edit_done(self, guest: dict, values: dict | None) -> None:
        if not values or not any(values[k] for k in ("cores", "memory", "grow")):
            return
        if values["grow"]:
            who = f"{guest['name']} ({guest['vmid']})"
            self.push_screen(
                ConfirmScreen(
                    f"Grow {values['disk']} of {who} by {values['grow']} GiB? This cannot be undone.",
                    "Grow disk",
                ),
                lambda ok: (
                    self._apply(guest, values) if ok else None
                ),  # "No" cancels the whole edit
            )
        else:
            self._apply(guest, values)

    def _apply(self, guest: dict, values: dict) -> None:
        cores, memory, grow = values["cores"], values["memory"], values["grow"]

        def job() -> None:
            if cores or memory:
                cli.set_resources(guest, cores=cores, memory_mb=memory)
            if grow:
                cli.resize_disk(guest, values["disk"], grow)
            if cores or memory:
                try:
                    pending = cli.pending_keys(guest)
                except Exception as exc:  # noqa: BLE001 - only a notice; the change itself succeeded
                    self.call_from_thread(
                        self.notify,
                        f"pending check failed: {_message(exc)}",
                        severity="warning",
                    )
                    return
                if {"cores", "memory"} & set(pending):
                    self.call_from_thread(self.notify, "Change applies at next reboot")

        self.start_job(guest, "edit", job)


def _reset_terminal() -> None:
    """Full terminal reset (ESC c, what `reset` sends): clean screen, no leftover TUI modes."""
    sys.stdout.write("\033c")
    sys.stdout.flush()


def _resources(config: dict, kind: str) -> list[tuple[str, str]]:
    """Configured CPU, memory (and CT swap) and disks, as (name, value) rows."""
    cores = config.get("cores")
    if cores is None and kind == "ct":
        cpu = "unlimited"  # no cores limit: the container may use all host cores
    else:
        cores = int(cores or 1)
        cpu = f"{cores} core{'s' if cores != 1 else ''}"
        if kind == "vm":
            sockets = int(config.get("sockets") or 1)
            cpu += f" × {sockets} socket{'s' if sockets != 1 else ''}"
    rows = [("CPU", cpu)]
    if config.get("memory") is not None:
        rows.append(("Memory", f"{config['memory']} MiB"))
    if kind == "ct" and config.get("swap") is not None:
        rows.append(("Swap", f"{config['swap']} MiB"))
    rows += [(f"Disk {name}", size) for name, size in cli.disks(config, kind).items()]
    return rows


def _pct(part, whole) -> float | None:
    return part / whole * 100 if part is not None and whole else None


def _percents(g: dict) -> list[str]:
    """CPU % and Mem % cells of the listing; "-" unless running and the field is known."""
    if g["status"] != "running":
        return ["-", "-"]
    cpu = g["cpu"] * 100 if g.get("cpu") is not None else None
    mem = _pct(g.get("mem"), g.get("maxmem"))
    return ["-" if v is None else f"{v:.1f}" for v in (cpu, mem)]


def _in_use(usage: dict, kind: str) -> dict[str, str]:
    """In-use cells of a running guest by Resource name; missing fields stay absent ("-")."""
    out = {}
    if usage.get("cpu") is not None:
        out["CPU"] = f"{usage['cpu'] * 100:.1f} %"
    if usage.get("mem") is not None:
        pct = _pct(usage["mem"], usage.get("maxmem"))
        out["Memory"] = f"{usage['mem'] / 2**20:.0f} MiB" + (
            f" ({pct:.0f} %)" if pct is not None else ""
        )
    if kind == "ct":
        if usage.get("swap") is not None:
            out["Swap"] = f"{usage['swap'] / 2**20:.0f} MiB"
        if usage.get("disk") is not None:
            pct = _pct(usage["disk"], usage.get("maxdisk"))
            out["Disk rootfs"] = f"{usage['disk'] / 2**30:.1f} GiB" + (
                f" ({pct:.0f} %)" if pct is not None else ""
            )
    return out


def _add_plain(table: DataTable, *cells, **kwargs) -> None:
    """DataTable parses str cells as markup; names and notes like "a[/b]" must stay literal."""
    table.add_row(*(Text(c) if isinstance(c, str) else c for c in cells), **kwargs)


def _scaled(value, unit: int):
    return "-" if value is None else round(value / unit)


def main() -> None:
    GuestApp().run()