Add context listing and switching to the TUI

Press c to list cv4pve-cli contexts and switch between them. The
active context is always shown in the header.

- Switching runs `cv4pve-cli config use`, which changes the global
  current context. cv4pve-cli 2.4.0 has no per-call context option.
- `config list` has no JSON output, so its text lines are parsed.
- The guest table is cleared as soon as another context is chosen, so
  the old cluster's guests cannot be acted on.
- A generation counter drops guest listings that started before a
  switch, and g and c are refused while a switch runs. Without this,
  a reload already in flight could refill the table with the old
  cluster's guests, and VMIDs collide across clusters.

103 tests pass. The switch was tested live from mainframe to
mainframe-edge and back.

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

.gitignore

@@ -0,0 +1,4 @@
+.venv/
+__pycache__/
+*.pyc
+.pytest_cache/

README.md

@@ -1,2 +1,40 @@
 # cv4pve-cli-tui
 
+Textual TUI for [cv4pve-cli](https://github.com/Corsinvest/cv4pve-cli): list Proxmox VE
+guests (VMs and LXC), start/shutdown/stop/reboot them, and change cores, memory and
+disk size.
+
+It shells out to `cv4pve-cli` and uses its current context (`cv4pve-cli config use <name>`);
+it stores no credentials itself. Tested with cv4pve-cli 2.4.0.
+
+## Run
+
+```sh
+uv run cv4pve-tui
+```
+
+| Key | Action |
+|-----|--------|
+| `s` | start |
+| `S` | shutdown (confirm) |
+| `x` | stop - hard power-off (confirm) |
+| `r` | reboot (confirm) |
+| `e` | edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) |
+| `g` | refresh |
+| `c` | switch cv4pve-cli context (cluster) |
+| `q` | quit |
+
+Switching context runs `cv4pve-cli config use`, which changes cv4pve-cli's global current
+context: other shells and scripts follow, and it stays after the app exits.
+
+Empty fields in the edit dialog are left unchanged. If a CPU/memory change cannot be
+hot-plugged, the app tells you it applies at the next reboot.
+
+## Development
+
+```sh
+uv sync
+uv run pytest
+```
+
+Tests mock `cv4pve-cli`; they never touch a cluster. Design: `docs/plans/guest-tui-mvp.md`.

docs/plans/guest-tui-mvp.md

@@ -0,0 +1,181 @@
+# Guest TUI MVP
+
+## Problem
+
+Manage Proxmox guests (VMs and LXC containers) from the terminal without remembering
+`cv4pve-cli` syntax: list them, start/shutdown/stop/reboot, and change CPU cores,
+memory and disk size.
+
+## Approach
+
+Thin subprocess wrapper around `cv4pve-cli` (tested against 2.4.0) plus a Textual app.
+The app never handles credentials: it uses whatever context `cv4pve-cli config use`
+selected.
+
+Observed CLI behaviour the design relies on:
+
+- `get guests -o json` -> list of dicts with `vmid`, `type` (`qemu`|`lxc`), `node`,
+  `name`, `status`, `maxcpu`, `maxmem` (bytes), `maxdisk` (bytes), ...
+- `show vm|ct <node> <vmid> -o json` -> list of `{"key": ..., "value": ...}`.
+- `api get /nodes/<node>/<qemu|lxc>/<vmid>/pending -o json` -> list of
+  `{"key", "value", "pending", "delete"}`; `pending` non-null means not yet applied.
+- Async actions print a UPID and exit 0 **even when the task fails**. With `--wait`
+  the CLI waits for the task and exits non-zero (5) with the error text on failure.
+  Every mutating call therefore uses `--wait`.
+- `do stop` refuses to run without `-y` (verified); `-y` is accepted by all four
+  lifecycle actions, so it is always passed.
+- Errors: non-zero exit, message on stdout/stderr. Sync failures are multi-line
+  (`Error: ...` / `Call: ...` / `Alias: ...`). Success is judged by exit code only,
+  never by parsing UPID text.
+- Each call costs ~0.5 s (.NET startup), so calls run in worker threads.
+- `get guests` (`/cluster/resources`) lags a few seconds behind actions; after an
+  action the app reads the acted-on guest's live status from
+  `api get /nodes/<node>/<qemu|lxc>/<vmid>/status/current` (key/value list).
+
+### Module `cv4pve_tui/cli.py`
+
+- `CliError(Exception)` - `str()` is the CLI's combined, stripped output;
+  `.summary` is one line for the UI: the first non-empty line that is not a UPID,
+  `Error:`-prefix stripped, plus the following indented detail line if the first
+  line ends with `:`.
+- `run(*args) -> str` - `subprocess.run(["cv4pve-cli", *args], capture_output=True,
+  text=True, stdin=DEVNULL, timeout=330)`; non-zero exit -> `CliError`.
+- `kind(guest) -> "vm" | "ct"` - from `type` (`qemu` -> `vm`, `lxc` -> `ct`).
+- `list_guests() -> list[dict]` - sorted by `vmid`.
+- `guest_config(guest) -> dict` - key/value list flattened to a dict.
+- `disks(config, kind) -> dict[str, str]` - resizable disks -> size string.
+  VM: `scsiN|virtioN|sataN|ideN` with a `size=`, excluding `media=cdrom` and
+  cloud-init. CT: `rootfs`, `mpN`.
+- `ACTIONS = ("start", "shutdown", "stop", "reboot")`;
+  `action(name, guest)` -> `do <name> <kind> <node> <vmid> -y --wait`;
+  unknown name -> `ValueError`.
+- `set_resources(guest, cores=None, memory_mb=None)` ->
+  `set <kind> config <node> <vmid> [--cores N] [--memory M] --wait`; values must be
+  positive ints; nothing to set -> `ValueError`.
+- `resize_disk(guest, disk, grow_gb)` -> `do resize <kind> disk <node> <vmid> <disk>
+  +<grow_gb>G --wait`; `grow_gb` must be a positive int (growth only - Proxmox cannot
+  shrink, and an increment cannot shrink by construction).
+- `pending_keys(guest) -> list[str]` - keys with a non-null `pending`.
+
+### Module `cv4pve_tui/app.py`
+
+- `DataTable` of guests: VMID, name, type, node, status, cores, memory (MiB),
+  disk (GiB). Cursor row = selected guest.
+- Keys: `s` start, `S` shutdown, `x` stop (hard), `r` reboot, `e` edit,
+  `g` refresh, `q` quit.
+- `shutdown`, `stop`, `reboot` open a confirm modal naming the guest; `stop` says it
+  is a hard power-off. `start` needs no confirmation.
+- `e` opens an edit modal: cores, memory MiB, disk select + "grow by GiB".
+  Empty field = unchanged. Disk growth gets its own confirm (irreversible).
+- After `set_resources`, if `pending_keys` contains `cores` or `memory`, notify that
+  the change applies at next reboot.
+- All CLI calls in thread workers; results/errors via `notify` (errors show
+  `CliError.summary`); table refreshes after each action.
+- A "working" notice is shown when an action starts (`--wait` can block minutes),
+  and a second action on a guest that is already busy is refused.
+- Action keys on an empty table do nothing (no crash).
+
+### UI contract (shared by tests and implementation)
+
+- `app.py` does `from cv4pve_tui import cli` and always calls `cli.<function>(...)`,
+  so tests monkeypatch `cv4pve_tui.cli.<function>`.
+- `class GuestApp(App)`; `def main()` runs it.
+- Guest table: `DataTable` with id `guests`, one row per guest, row key = str(vmid),
+  in `list_guests()` order. Loaded on mount.
+- `ConfirmScreen(ModalScreen[bool])`: buttons `#yes` / `#no`; its text names the guest
+  (name and vmid); for `stop` it contains the words "hard power-off".
+- `EditScreen(ModalScreen)`: `Input#cores`, `Input#memory`, `Select#disk`,
+  `Input#grow`, buttons `#apply` / `#cancel`. Inputs are pre-filled with nothing
+  (empty = unchanged); placeholders show current values. Disk options come from
+  `cli.disks(cli.guest_config(guest), cli.kind(guest))`.
+  On apply: if cores/memory given -> `cli.set_resources(guest, cores=..., memory_mb=...)`
+  (only the given ones, as ints); if grow given -> `ConfirmScreen` first, then
+  `cli.resize_disk(guest, disk, grow_gb)`. Non-integer or <= 0 input -> `notify`
+  with severity "error" and no CLI call.
+- Thread workers (`@work(thread=True)` or `run_worker(..., thread=True)`); tests wait
+  with `await app.workers.wait_for_complete()` then `await pilot.pause()`.
+
+## Out of scope
+
+Create/delete, snapshots, migration, console, backup, in-app context switching,
+sockets/vCPU/hotplug settings, live auto-refresh.
+
+## Steps
+
+- [x] Project skeleton (`uv`, `pyproject.toml`, entry point `cv4pve-tui`) -
+      `uv run python -c "import cv4pve_tui"` succeeds.
+- [x] Tests for `cli.py` with `subprocess.run` mocked - fail before implementation.
+- [x] `cli.py` - tests pass.
+- [x] Tests for the app with Textual `run_test` and `cli` functions mocked.
+- [x] `app.py` - tests pass.
+- [x] Read-only smoke test against the real cluster (listing, edit dialog disk options,
+      pending check).
+- [x] Live edit (cores/memory), shutdown, start, reboot on CT 104 (mainframe), restored
+      to baseline afterwards. Disk resize not run live (irreversible).
+
+## Open questions
+
+None blocking.
+
+---
+
+# Context switching
+
+## Problem
+
+Several cv4pve-cli contexts (clusters) exist; switch between them inside the TUI
+instead of quitting and running `cv4pve-cli config use`.
+
+## Decision
+
+Option A: switching runs `cv4pve-cli config use <name>`, which changes the **global**
+current context (all shells and scripts follow) and stays after the app exits. cv4pve-cli
+2.4.0 has no per-call context flag or env var (verified), so there is no app-local option.
+
+Observed CLI behaviour:
+
+- `config list` has no `-o json`. Text output, one line per context; current one
+  prefixed with `* `, others with two spaces:
+  `* mainframe            mainframe.iot.home.arpa:8006   (api-token)`
+  `  mainframe-edge       mainframe-edge.home.arpa:8006   (api-token)`
+- `config use <name>` -> exit 0 on success; unknown name -> exit 3,
+  `Error: Context '<name>' not found.`, current context unchanged.
+
+### `cli.py`
+
+- `contexts() -> list[dict]` - `run("config", "list")`, each line parsed to
+  `{"name", "host", "auth", "current": bool}` (regex
+  `^(\*)?\s*(\S+)\s+(\S+)\s+\((.*)\)\s*$`); non-matching lines are ignored.
+- `use_context(name)` - `run("config", "use", name)`.
+
+### `app.py` (UI contract)
+
+- The active context is always visible: `app.sub_title == f"context: {name}"`
+  (shown in the Header). Set on mount from `cli.contexts()` and after every switch.
+  If `contexts()` fails: error notify, sub_title `context: ?`.
+- Key `c` opens `ContextScreen(ModalScreen[str | None])` with `OptionList#contexts`,
+  one option per context, option id = context name, current one marked with `*`.
+  Selecting an option dismisses with its name; `escape` dismisses with `None`.
+- `c` while any guest is busy -> warning notify, no screen (VMIDs collide across
+  clusters, so the busy guard would point at the wrong cluster).
+- Selecting the current context or dismissing -> no CLI call.
+- A reload started before a switch must not refill the table (generation counter);
+  `g` and `c` are refused while a switch runs; the post-action reload finishes before
+  the guest stops being busy.
+- Selecting another context: table cleared immediately (old cluster's guests must not
+  stay actionable), then in a worker `cli.use_context(name)`, sub_title updated, guests
+  reloaded. On `CliError`: error notify with summary, sub_title re-read from
+  `cli.contexts()`, guests reloaded.
+
+## Out of scope
+
+Adding/editing/deleting contexts, `config verify`, restoring the original context on exit.
+
+## Steps
+
+- [x] Tests for `contexts`/`use_context` and the UI contract above - red first.
+- [x] Implementation - full suite green, ruff clean.
+- [x] Live: open the list, switch to `mainframe-edge` and back to `mainframe`
+      (only `config use` and read-only listing; no guest actions); final context
+      is `mainframe`.
+- [ ] Push `feat/guest-tui-mvp`, merge to `main` (user-approved).

pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "cv4pve-tui"
+version = "0.1.0"
+description = "Textual TUI for cv4pve-cli: manage Proxmox VE guests"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = ["textual>=8.2"]
+
+[project.scripts]
+cv4pve-tui = "cv4pve_tui.app:main"
+
+[dependency-groups]
+dev = ["pytest>=8", "pytest-asyncio>=1.0"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+asyncio_mode = "auto"

src/cv4pve_tui/__init__.py

@@ -0,0 +1 @@
+"""Textual TUI for cv4pve-cli."""

src/cv4pve_tui/app.py

@@ -0,0 +1,398 @@
+"""Textual app for managing Proxmox guests through cv4pve-cli."""
+
+from textual import work
+from textual.app import App, ComposeResult
+from textual.containers import Horizontal, Vertical
+from textual.screen import ModalScreen
+from textual.widgets import (
+    Button,
+    DataTable,
+    Footer,
+    Header,
+    Input,
+    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 ConfirmScreen(ModalScreen[bool]):
+    BINDINGS = [("escape", "dismiss(False)", "No")]  # noqa: RUF012 - Textual idiom
+
+    def __init__(self, message: str) -> None:
+        super().__init__()
+        self.message = message
+
+    def compose(self) -> ComposeResult:
+        with Vertical():
+            yield Static(self.message)
+            with Horizontal():
+                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(ModalScreen):
+    """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 Vertical():
+            yield Static(
+                f"Edit {self.guest['name']} ({self.guest['vmid']}); empty = unchanged"
+            )
+            yield Input(
+                placeholder=f"cores (now {self.config.get('cores', '?')})", id="cores"
+            )
+            yield Input(
+                placeholder=f"memory MiB (now {self.config.get('memory', '?')})",
+                id="memory",
+            )
+            yield Select(
+                [(f"{k} ({v})", k) for k, v in self.disks.items()],
+                prompt="disk",
+                id="disk",
+            )
+            yield Input(placeholder="grow by GiB", id="grow")
+            with Horizontal():
+                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(ModalScreen[str | None]):
+    """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 Vertical():
+            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 GuestApp(App):
+    BINDINGS = [  # noqa: RUF012 - Textual idiom
+        ("s", "act('start')", "Start"),
+        ("S", "act('shutdown')", "Shutdown"),
+        ("x", "act('stop')", "Stop"),
+        ("r", "act('reboot')", "Reboot"),
+        ("e", "edit", "Edit"),
+        ("g", "reload", "Refresh"),
+        ("c", "context", "Context"),
+        ("q", "quit", "Quit"),
+    ]
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.guests: dict[str, dict] = {}
+        self.busy: set[int] = set()
+        # bumped on every context switch; replies started earlier come from the old cluster
+        self.generation = 0
+        self.switching = False
+
+    def compose(self) -> ComposeResult:
+        yield Header()
+        yield DataTable(id="guests", cursor_type="row")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        table = self.query_one("#guests", DataTable)
+        table.add_columns(
+            "VMID",
+            "Name",
+            "Type",
+            "Node",
+            "Status",
+            "Cores",
+            "Memory (MiB)",
+            "Disk (GiB)",
+        )
+        self._show_context()
+        self.action_reload()
+
+    @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) -> None:
+        if gen != self.generation:  # listed before a context switch: old cluster
+            return
+        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:
+            table.add_row(
+                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),
+                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"])))
+
+    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"]:
+                    g["status"] = cli.status(g)
+        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
+        self.call_from_thread(self._fill, guests, gen)
+
+    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.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()
+        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(
+                acted=guest
+            )  # before un-busying, so no context switch can start mid-reload
+            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),
+            lambda ok: self._start_job(guest, name, run) if ok else None,
+        )
+
+    def action_edit(self) -> None:
+        guest = self._selected()
+        if guest is None:
+            return
+        if guest["vmid"] in self.busy:
+            self.notify(
+                f"{guest['name']} ({guest['vmid']}) is busy", severity="warning"
+            )
+            return
+        self._open_edit(guest)
+
+    @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."
+                ),
+                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 _scaled(value, unit: int):
+    return "-" if value is None else round(value / unit)
+
+
+def main() -> None:
+    GuestApp().run()

src/cv4pve_tui/cli.py

@@ -0,0 +1,158 @@
+"""Thin subprocess wrapper around cv4pve-cli."""
+
+import json
+import re
+import subprocess
+
+
+class CliError(Exception):
+    """cv4pve-cli failed; str() is its combined, stripped output."""
+
+    @property
+    def summary(self) -> str:
+        """One line for the UI: first non-UPID line, plus its indented detail line."""
+        lines = str(self).splitlines()
+        for i, line in enumerate(lines):
+            text = line.strip()
+            if not text or text.startswith("UPID:"):
+                continue
+            text = text.removeprefix("Error:").strip()
+            if text.endswith(":") and i + 1 < len(lines) and lines[i + 1][:1].isspace():
+                text += " " + lines[i + 1].strip()
+            return text
+        return str(self)
+
+
+def run(*args: str) -> str:
+    try:
+        proc = subprocess.run(
+            ["cv4pve-cli", *args],
+            check=False,
+            capture_output=True,
+            text=True,
+            stdin=subprocess.DEVNULL,
+            timeout=330,
+        )
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        raise CliError(str(exc)) from exc
+    if proc.returncode != 0:
+        raise CliError(
+            "\n".join(s.strip() for s in (proc.stdout, proc.stderr) if s.strip())
+        )
+    return proc.stdout
+
+
+def kind(guest: dict) -> str:
+    return "vm" if guest["type"] == "qemu" else "ct"
+
+
+def _api_type(guest: dict) -> str:
+    return "qemu" if guest["type"] == "qemu" else "lxc"
+
+
+def list_guests() -> list[dict]:
+    guests = json.loads(run("get", "guests", "-o", "json"))
+    for g in guests:
+        g["name"] = (
+            g.get("name") or f"VM {g['vmid']}"
+        )  # unnamed guests have no/null name
+    return sorted(guests, key=lambda g: g["vmid"])
+
+
+def guest_config(guest: dict) -> dict:
+    out = run("show", kind(guest), guest["node"], str(guest["vmid"]), "-o", "json")
+    return {item["key"]: item["value"] for item in json.loads(out)}
+
+
+_VM_DISK = re.compile(r"(scsi|virtio|sata|ide)\d+")
+_CT_DISK = re.compile(r"rootfs|mp\d+")
+
+
+def disks(config: dict, kind: str) -> dict[str, str]:
+    pattern = _VM_DISK if kind == "vm" else _CT_DISK
+    result = {}
+    for key, value in config.items():
+        if not pattern.fullmatch(key):
+            continue
+        value = str(value)
+        size = re.search(r"size=(\w+)", value)
+        if size and "media=cdrom" not in value and "cloudinit" not in value:
+            result[key] = size.group(1)
+    return result
+
+
+ACTIONS = ("start", "shutdown", "stop", "reboot")
+
+
+def action(name: str, guest: dict) -> None:
+    if name not in ACTIONS:
+        raise ValueError(f"unknown action: {name}")
+    run("do", name, kind(guest), guest["node"], str(guest["vmid"]), "-y", "--wait")
+
+
+def _positive(value, what: str) -> int:
+    if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
+        raise ValueError(f"{what} must be a positive integer")
+    return value
+
+
+def set_resources(
+    guest: dict, cores: int | None = None, memory_mb: int | None = None
+) -> None:
+    opts = []
+    if cores is not None:
+        opts += ["--cores", str(_positive(cores, "cores"))]
+    if memory_mb is not None:
+        opts += ["--memory", str(_positive(memory_mb, "memory"))]
+    if not opts:
+        raise ValueError("nothing to set")
+    run(
+        "set", kind(guest), "config", guest["node"], str(guest["vmid"]), *opts, "--wait"
+    )
+
+
+def resize_disk(guest: dict, disk: str, grow_gb: int) -> None:
+    _positive(grow_gb, "grow")
+    run(
+        "do",
+        "resize",
+        kind(guest),
+        "disk",
+        guest["node"],
+        str(guest["vmid"]),
+        disk,
+        f"+{grow_gb}G",
+        "--wait",
+    )
+
+
+def pending_keys(guest: dict) -> list[str]:
+    path = f"/nodes/{guest['node']}/{_api_type(guest)}/{guest['vmid']}/pending"
+    items = json.loads(run("api", "get", path, "-o", "json"))
+    return [i["key"] for i in items if i.get("pending") is not None]
+
+
+def status(guest: dict) -> str:
+    """Live status; get guests (/cluster/resources) lags a few seconds behind actions."""
+    path = f"/nodes/{guest['node']}/{_api_type(guest)}/{guest['vmid']}/status/current"
+    return {
+        i["key"]: i["value"] for i in json.loads(run("api", "get", path, "-o", "json"))
+    }["status"]
+
+
+_CONTEXT_LINE = re.compile(r"^(\*)?\s*(\S+)\s+(\S+)\s+\((.*)\)\s*$")
+
+
+def contexts() -> list[dict]:
+    """Parse `config list` (text only); lines that do not look like a context are skipped."""
+    found = (_CONTEXT_LINE.match(line) for line in run("config", "list").splitlines())
+    return [
+        {"name": m[2], "host": m[3], "auth": m[4], "current": bool(m[1])}
+        for m in found
+        if m
+    ]
+
+
+def use_context(name: str) -> None:
+    """Changes cv4pve-cli's GLOBAL current context."""
+    run("config", "use", name)

tests/test_app.py

@@ -0,0 +1,729 @@
+"""Textual pilot tests for cv4pve_tui.app; spec: docs/plans/guest-tui-mvp.md ("UI contract").
+
+All cli functions are monkeypatched: cv4pve-cli is never executed.
+"""
+
+import subprocess
+import threading
+
+import pytest
+from textual.screen import ModalScreen
+from textual.widgets import Button, DataTable, Input, OptionList, Select, Static
+
+from cv4pve_tui import cli
+from cv4pve_tui.app import ConfirmScreen, ContextScreen, EditScreen, GuestApp
+
+VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu", "status": "running",
+      "maxcpu": 4, "maxmem": 8192 * 2**20, "maxdisk": 10 * 2**30}  # fmt: skip
+CT = {"vmid": 100, "name": "backrest", "node": "mainframe", "type": "lxc", "status": "stopped",
+      "maxcpu": 1, "maxmem": 512 * 2**20, "maxdisk": 8 * 2**30}  # fmt: skip
+
+
+class Fake:
+    """Recorder standing in for the cli module functions the app calls."""
+
+    def __init__(self):
+        self.guests = [VM, CT]
+        self.pending = []
+        self.disk_map = {"scsi0": "10G"}
+        self.list_calls = 0
+        self.actions, self.sets, self.resizes, self.pending_calls = [], [], [], []
+        self.action_error = None
+        self.action_hook = None
+        self.live_status, self.status_calls = {}, []
+        self.current = "mainframe"
+        self.names = ["mainframe", "mainframe-edge"]
+        self.contexts_error = None
+        self.contexts_calls = 0
+        self.use_calls = []
+        self.use_error = None
+        self.use_hook = None
+        self.guests_by_context = {}
+        self.list_hook = None
+
+    def list_guests(self):
+        self.list_calls += 1
+        snapshot = list(self.guests_by_context.get(self.current, self.guests))
+        if (
+            self.list_hook
+        ):  # runs after reading: simulates a slow reply from that cluster
+            self.list_hook()
+        return snapshot
+
+    def guest_config(self, guest):
+        return {"cores": guest["maxcpu"]}
+
+    def disks(self, config, kind):
+        return dict(self.disk_map)
+
+    def action(self, name, guest):
+        self.actions.append((name, guest["vmid"]))
+        if self.action_hook:
+            self.action_hook()
+        if self.action_error:
+            raise self.action_error
+
+    def set_resources(self, guest, cores=None, memory_mb=None):
+        given = {"cores": cores, "memory_mb": memory_mb}
+        self.sets.append({k: v for k, v in given.items() if v is not None})
+
+    def resize_disk(self, guest, disk, grow_gb):
+        self.resizes.append((guest["vmid"], disk, grow_gb))
+
+    def pending_keys(self, guest):
+        self.pending_calls.append(guest["vmid"])
+        return list(self.pending)
+
+    def status(self, guest):
+        self.status_calls.append(guest["vmid"])
+        return self.live_status.get(guest["vmid"], guest["status"])
+
+    def contexts(self):
+        self.contexts_calls += 1
+        if self.contexts_error:
+            raise self.contexts_error
+        return [
+            {"name": n, "host": f"{n}.home.arpa:8006", "auth": "api-token",
+             "current": n == self.current}
+            for n in self.names
+        ]  # fmt: skip
+
+    def use_context(self, name):
+        self.use_calls.append(name)
+        if self.use_hook:
+            self.use_hook()
+        if self.use_error:
+            raise self.use_error
+        self.current = name
+
+
+@pytest.fixture
+def fake(monkeypatch):
+    f = Fake()
+    for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
+                 "resize_disk", "pending_keys", "status", "contexts", "use_context"):  # fmt: skip
+        monkeypatch.setattr(cli, name, getattr(f, name))
+    return f
+
+
+def make_app():
+    app = GuestApp()
+    app.seen = []  # (message, severity) of every notify
+    original = app.notify
+
+    def spy(message, **kwargs):
+        app.seen.append((str(message), kwargs.get("severity", "information")))
+        return original(message, **kwargs)
+
+    app.notify = spy
+    return app
+
+
+async def settle(app, pilot):
+    await pilot.pause()
+    await app.workers.wait_for_complete()
+    await pilot.pause()
+
+
+def text_of(screen):
+    return " ".join(str(w.render()) for w in screen.query(Static))
+
+
+def errors(app):
+    return [m for m, sev in app.seen if sev == "error"]
+
+
+async def press_button(app, pilot, selector):
+    app.screen.query_one(selector, Button).press()
+    await settle(app, pilot)
+
+
+# --- table ------------------------------------------------------------------
+
+
+async def test_table_loads_one_row_per_guest_in_list_order(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        table = app.query_one("#guests", DataTable)
+        assert table.row_count == 2
+        assert [k.value for k in table.rows] == ["108", "100"]
+        cells = [str(c) for c in table.get_row("108")]
+        assert "docker" in cells and "mainframe" in cells and "running" in cells
+        assert "8192" in cells  # memory in MiB
+
+
+async def test_refresh_key_reloads_guests(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        before = fake.list_calls
+        fake.guests = [CT]
+        await pilot.press("g")
+        await settle(app, pilot)
+        assert fake.list_calls == before + 1
+        assert app.query_one("#guests", DataTable).row_count == 1
+
+
+# --- lifecycle actions ------------------------------------------------------
+
+
+async def test_start_runs_without_confirm_and_refreshes_table(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        before = fake.list_calls
+        await pilot.press("s")
+        await settle(app, pilot)
+        assert fake.actions == [("start", 108)]
+        assert not isinstance(app.screen, ModalScreen)
+        assert fake.list_calls > before
+
+
+async def test_action_targets_the_cursor_row(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press("down", "s")
+        await settle(app, pilot)
+        assert fake.actions == [("start", 100)]
+
+
+@pytest.mark.parametrize(
+    "key,name", [("S", "shutdown"), ("x", "stop"), ("r", "reboot")]
+)
+async def test_confirmed_action_calls_cli_only_after_yes(fake, key, name):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press(key)
+        await pilot.pause()
+        assert isinstance(app.screen, ConfirmScreen)
+        assert "docker" in text_of(app.screen) and "108" in text_of(app.screen)
+        assert fake.actions == []
+        await press_button(app, pilot, "#yes")
+        assert fake.actions == [(name, 108)]
+
+
+@pytest.mark.parametrize("key", ["S", "x", "r"])
+async def test_declined_action_does_not_call_cli(fake, key):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press(key)
+        await pilot.pause()
+        await press_button(app, pilot, "#no")
+        assert fake.actions == []
+        assert not isinstance(app.screen, ModalScreen)
+
+
+async def test_stop_confirm_says_hard_power_off(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press("x")
+        await pilot.pause()
+        assert "hard power-off" in text_of(app.screen).lower()
+
+
+@pytest.mark.parametrize("key", ["s", "S", "x", "r", "e"])
+async def test_keys_on_empty_table_do_nothing(fake, key):
+    fake.guests = []
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press(key)
+        await settle(app, pilot)
+        assert fake.actions == [] and fake.sets == [] and fake.resizes == []
+        assert not isinstance(app.screen, ModalScreen)
+
+
+async def test_cli_error_is_notified_with_summary(fake, monkeypatch):
+    with monkeypatch.context() as m:  # build a real CliError through cli.run
+        m.setattr(cli.subprocess, "run", lambda argv, **kw: subprocess.CompletedProcess(
+            argv, 5, "UPID:mainframe:0:0:0:qmstart:108:root@pam:\nstart failed: no quorum\n", ""))  # fmt: skip
+        with pytest.raises(cli.CliError) as exc:
+            cli.run("do", "start")
+    fake.action_error = exc.value
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press("s")
+        await settle(app, pilot)
+        assert any(exc.value.summary in m for m in errors(app))
+        assert not any("UPID" in m for m in errors(app))
+
+
+async def test_busy_guest_refuses_second_action(fake):
+    started, release = threading.Event(), threading.Event()
+    fake.action_hook = lambda: (started.set(), release.wait(10))
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        try:
+            await pilot.press("s")
+            for _ in range(200):
+                if started.is_set():
+                    break
+                await pilot.pause(0.01)
+            assert started.is_set()
+            assert app.seen, "a working notice is expected while the action runs"
+            await pilot.press("s")
+            await pilot.pause()
+            assert fake.actions == [("start", 108)]
+        finally:
+            release.set()
+        await settle(app, pilot)
+        assert fake.actions == [("start", 108)]
+
+
+# --- edit flow --------------------------------------------------------------
+
+
+async def open_edit(app, pilot):
+    await settle(app, pilot)
+    await pilot.press("e")
+    await settle(app, pilot)
+    assert isinstance(app.screen, EditScreen)
+
+
+def fill(app, **values):
+    for key, value in values.items():
+        w = app.screen.query_one(f"#{key}", Select if key == "disk" else Input)
+        w.value = value
+
+
+async def test_edit_applies_only_given_fields_as_ints(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, cores="8")
+        await press_button(app, pilot, "#apply")
+        assert fake.sets == [{"cores": 8}]
+        await pilot.press("e")
+        await settle(app, pilot)
+        fill(app, cores="2", memory="4096")
+        await press_button(app, pilot, "#apply")
+        assert fake.sets[-1] == {"cores": 2, "memory_mb": 4096}
+        assert all(isinstance(v, int) for v in fake.sets[-1].values())
+
+
+@pytest.mark.parametrize("field", ["cores", "memory"])
+@pytest.mark.parametrize("bad", ["abc", "0", "-4", "2.5"])
+async def test_edit_invalid_resource_input_errors_without_cli_call(fake, field, bad):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, **{field: bad})
+        await press_button(app, pilot, "#apply")
+        assert errors(app)
+        assert fake.sets == [] and fake.resizes == []
+
+
+@pytest.mark.parametrize("bad", ["abc", "0", "-1"])
+async def test_edit_invalid_grow_errors_without_cli_call(fake, bad):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, disk="scsi0", grow=bad)
+        await press_button(app, pilot, "#apply")
+        assert errors(app)
+        assert fake.resizes == []
+
+
+async def test_edit_with_empty_fields_changes_nothing(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        await press_button(app, pilot, "#apply")
+        assert fake.sets == [] and fake.resizes == []
+
+
+async def test_edit_cancel_changes_nothing(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, cores="8")
+        await press_button(app, pilot, "#cancel")
+        assert fake.sets == [] and fake.resizes == []
+        assert not isinstance(app.screen, ModalScreen)
+
+
+async def test_grow_needs_confirm_then_resizes(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, disk="scsi0", grow="5")
+        await press_button(app, pilot, "#apply")
+        assert isinstance(app.screen, ConfirmScreen)
+        assert fake.resizes == []
+        await press_button(app, pilot, "#yes")
+        assert fake.resizes == [(108, "scsi0", 5)]
+
+
+async def test_grow_declined_does_not_resize(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, disk="scsi0", grow="5")
+        await press_button(app, pilot, "#apply")
+        await press_button(app, pilot, "#no")
+        assert fake.resizes == []
+
+
+@pytest.mark.parametrize(
+    "pending,expect_notice", [(["memory"], True), (["cores"], True), ([], False)]
+)
+async def test_pending_cores_or_memory_notifies_reboot(fake, pending, expect_notice):
+    fake.pending = pending
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, memory="4096")
+        await press_button(app, pilot, "#apply")
+        assert fake.pending_calls == [108]
+        assert any("reboot" in m.lower() for m, _ in app.seen) is expect_notice
+
+
+# --- review findings: robustness --------------------------------------------
+
+
+async def test_guest_without_size_fields_still_listed(fake):
+    # e.g. a guest on an unreachable node: the API omits maxcpu/maxmem/maxdisk
+    fake.guests = [
+        {
+            "vmid": 999,
+            "name": "ghost",
+            "node": "down",
+            "type": "qemu",
+            "status": "unknown",
+        }
+    ]
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        assert app.query_one("#guests", DataTable).row_count == 1
+
+
+async def test_unexpected_error_in_action_is_notified_not_fatal(fake):
+    fake.action_error = KeyError("key")
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press("s")
+        await settle(app, pilot)
+        assert errors(app)
+        assert app.is_running and app.busy == set()
+
+
+async def test_unexpected_error_in_listing_is_notified_not_fatal(fake, monkeypatch):
+    def broken():
+        raise ValueError("bad json")
+
+    monkeypatch.setattr(cli, "list_guests", broken)
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        assert errors(app) and app.is_running
+
+
+async def test_pending_check_failure_does_not_skip_confirmed_grow(fake, monkeypatch):
+    def broken(guest):
+        raise cli.CliError("Error: boom")
+
+    monkeypatch.setattr(cli, "pending_keys", broken)
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, memory="4096", disk="scsi0", grow="5")
+        await press_button(app, pilot, "#apply")
+        await press_button(app, pilot, "#yes")
+        assert fake.sets == [{"memory_mb": 4096}]
+        assert fake.resizes == [(108, "scsi0", 5)]
+
+
+async def test_declining_grow_cancels_whole_edit(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_edit(app, pilot)
+        fill(app, cores="8", disk="scsi0", grow="5")
+        await press_button(app, pilot, "#apply")
+        await press_button(app, pilot, "#no")
+        assert fake.sets == [] and fake.resizes == []
+
+
+@pytest.mark.parametrize("key", ["S", "e"])
+async def test_escape_closes_modal_without_cli_call(fake, key):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        await pilot.press(key)
+        await settle(app, pilot)
+        assert isinstance(app.screen, ModalScreen)
+        await pilot.press("escape")
+        await settle(app, pilot)
+        assert not isinstance(app.screen, ModalScreen)
+        assert fake.actions == [] and fake.sets == []
+
+
+async def test_table_shows_live_status_after_action_not_lagging_list(fake):
+    # /cluster/resources lags a few seconds behind; the acted-on guest must show live status
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        fake.live_status = {108: "stopped"}  # list_guests still says "running"
+        await pilot.press("S")
+        await settle(app, pilot)
+        await press_button(app, pilot, "#yes")
+        assert fake.status_calls == [108]
+        assert "stopped" in [
+            str(c) for c in app.query_one("#guests", DataTable).get_row("108")
+        ]
+
+
+# --- context switching ------------------------------------------------------
+# spec: docs/plans/guest-tui-mvp.md ("Context switching" > "app.py (UI contract)")
+
+
+async def open_contexts(app, pilot):
+    await settle(app, pilot)
+    await pilot.press("c")
+    await settle(app, pilot)
+    assert isinstance(app.screen, ContextScreen)
+    return app.screen.query_one("#contexts", OptionList)
+
+
+async def choose(app, pilot, index):
+    ol = app.screen.query_one("#contexts", OptionList)
+    ol.focus()
+    ol.highlighted = index
+    await pilot.pause()
+    await pilot.press("enter")
+    await settle(app, pilot)
+
+
+async def test_sub_title_shows_active_context_after_mount(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        assert app.sub_title == "context: mainframe"
+
+
+async def test_contexts_failure_notifies_and_sub_title_is_unknown(fake):
+    fake.contexts_error = cli.CliError("Error: no config")
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        assert errors(app)
+        assert app.sub_title == "context: ?"
+
+
+async def test_c_opens_context_screen_listing_contexts_with_current_marked(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        ol = await open_contexts(app, pilot)
+        assert ol.option_count == 2
+        options = [ol.get_option_at_index(i) for i in range(2)]
+        assert [o.id for o in options] == ["mainframe", "mainframe-edge"]
+        assert "*" in str(options[0].prompt)
+        assert "*" not in str(options[1].prompt)
+        assert "mainframe-edge" in str(options[1].prompt)
+
+
+async def test_selecting_other_context_switches_and_reloads(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_contexts(app, pilot)
+        before = fake.list_calls
+        await choose(app, pilot, 1)
+        assert fake.use_calls == ["mainframe-edge"]
+        assert app.sub_title == "context: mainframe-edge"
+        assert fake.list_calls > before
+        assert not isinstance(app.screen, ModalScreen)
+        assert app.query_one("#guests", DataTable).row_count == 2
+
+
+async def test_table_is_cleared_while_switch_is_in_progress(fake):
+    started, release = threading.Event(), threading.Event()
+    fake.use_hook = lambda: (started.set(), release.wait(10))
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_contexts(app, pilot)
+        assert app.query_one("#guests", DataTable).row_count == 2
+        try:
+            ol = app.screen.query_one("#contexts", OptionList)
+            ol.focus()
+            ol.highlighted = 1
+            await pilot.pause()
+            await pilot.press("enter")
+            for _ in range(200):
+                if started.is_set():
+                    break
+                await pilot.pause(0.01)
+            assert started.is_set()
+            await pilot.pause()
+            assert app.query_one("#guests", DataTable).row_count == 0
+        finally:
+            release.set()
+        await settle(app, pilot)
+        assert app.query_one("#guests", DataTable).row_count == 2
+
+
+async def test_selecting_current_context_makes_no_cli_call(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_contexts(app, pilot)
+        before = fake.list_calls
+        await choose(app, pilot, 0)
+        assert fake.use_calls == []
+        assert not isinstance(app.screen, ModalScreen)
+        assert app.sub_title == "context: mainframe"
+        assert fake.list_calls == before
+
+
+async def test_escape_closes_context_screen_without_cli_call(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_contexts(app, pilot)
+        await pilot.press("escape")
+        await settle(app, pilot)
+        assert not isinstance(app.screen, ModalScreen)
+        assert fake.use_calls == []
+        assert app.sub_title == "context: mainframe"
+
+
+async def test_c_while_guest_busy_warns_and_opens_nothing(fake):
+    started, release = threading.Event(), threading.Event()
+    fake.action_hook = lambda: (started.set(), release.wait(10))
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        try:
+            await pilot.press("s")
+            for _ in range(200):
+                if started.is_set():
+                    break
+                await pilot.pause(0.01)
+            assert started.is_set()
+            await pilot.press("c")
+            await pilot.pause()
+            assert not isinstance(app.screen, ContextScreen)
+            assert any(sev == "warning" for _, sev in app.seen)
+        finally:
+            release.set()
+        await settle(app, pilot)
+        assert fake.use_calls == []
+
+
+async def test_use_context_failure_notifies_rereads_context_and_reloads(fake):
+    err = cli.CliError("Error: Context 'mainframe-edge' not found.")
+
+    def hook():  # the CLI state differs from what the app assumed
+        fake.current = "mainframe-edge"
+
+    fake.use_hook, fake.use_error = hook, err
+    app = make_app()
+    async with app.run_test() as pilot:
+        await open_contexts(app, pilot)
+        contexts_before, list_before = fake.contexts_calls, fake.list_calls
+        await choose(app, pilot, 1)
+        assert any(err.summary in m for m in errors(app))
+        assert fake.contexts_calls > contexts_before
+        assert app.sub_title == "context: mainframe-edge"  # whatever contexts() says
+        assert fake.list_calls > list_before
+        assert app.query_one("#guests", DataTable).row_count == 2
+
+
+# --- review finding: no old-cluster rows after a switch ----------------------
+
+EDGE = {"vmid": 300, "name": "edge-vm", "node": "edge", "type": "qemu", "status": "running",
+        "maxcpu": 2, "maxmem": 2**30, "maxdisk": 2**31}  # fmt: skip
+
+
+async def until(pilot, cond):
+    for _ in range(300):
+        if cond():
+            return
+        await pilot.pause(0.01)
+    raise AssertionError("condition not reached")
+
+
+def blocking_once():
+    started, release = threading.Event(), threading.Event()
+
+    def hook():
+        if not started.is_set():
+            started.set()
+            release.wait(10)
+
+    return hook, started, release
+
+
+def rows(app):
+    return [k.value for k in app.query_one("#guests", DataTable).rows]
+
+
+async def switch_to_edge(app, pilot):
+    await pilot.press("c")
+    await until(pilot, lambda: isinstance(app.screen, ContextScreen))
+    ol = app.screen.query_one("#contexts", OptionList)
+    ol.focus()
+    ol.highlighted = 1
+    await pilot.pause()
+    await pilot.press("enter")
+
+
+async def test_reload_in_flight_before_switch_cannot_restore_old_rows(fake):
+    fake.guests_by_context = {"mainframe-edge": [EDGE]}
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        hook, started, release = blocking_once()
+        fake.list_hook = hook
+        try:
+            await pilot.press("g")  # reads mainframe, then stalls
+            await until(pilot, started.is_set)
+            await switch_to_edge(app, pilot)
+            await until(pilot, lambda: rows(app) == ["300"])
+        finally:
+            release.set()
+        await settle(app, pilot)
+        assert rows(app) == ["300"]
+        assert app.sub_title == "context: mainframe-edge"
+
+
+async def test_refresh_refused_while_switching(fake):
+    started, release = threading.Event(), threading.Event()
+    fake.use_hook = lambda: (started.set(), release.wait(10))
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        try:
+            await switch_to_edge(app, pilot)
+            await until(pilot, started.is_set)
+            before = fake.list_calls
+            await pilot.press("g")
+            await pilot.pause()
+            assert fake.list_calls == before
+            assert any(sev == "warning" for _, sev in app.seen)
+        finally:
+            release.set()
+        await settle(app, pilot)
+
+
+async def test_context_refused_while_post_action_reload_runs(fake):
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        hook, started, release = blocking_once()
+        fake.list_hook = hook
+        try:
+            await pilot.press("s")  # start is instant; the reload after it stalls
+            await until(pilot, started.is_set)
+            await pilot.press("c")
+            await pilot.pause(0.1)
+            assert not isinstance(app.screen, ContextScreen)
+            assert any(sev == "warning" for _, sev in app.seen)
+        finally:
+            release.set()
+        await settle(app, pilot)

tests/test_cli.py

@@ -0,0 +1,421 @@
+"""Tests for cv4pve_tui.cli; spec: docs/plans/guest-tui-mvp.md ("Module cv4pve_tui/cli.py").
+
+subprocess.run is always mocked: cv4pve-cli is never executed.
+"""
+
+import json
+import subprocess
+
+import pytest
+
+from cv4pve_tui import cli
+
+VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu"}
+CT = {"vmid": 100, "name": "backrest", "node": "mainframe", "type": "lxc"}
+
+
+class FakeRun:
+    def __init__(self, monkeypatch, stdout="", stderr="", returncode=0):
+        self.calls = []
+        self.result = (stdout, stderr, returncode)
+        monkeypatch.setattr(cli.subprocess, "run", self)
+
+    def __call__(self, argv, **kwargs):
+        self.calls.append((list(argv), kwargs))
+        out, err, rc = self.result
+        return subprocess.CompletedProcess(argv, rc, out, err)
+
+    @property
+    def argv(self):
+        assert len(self.calls) == 1
+        return self.calls[0][0]
+
+
+@pytest.fixture
+def fake(monkeypatch):
+    return FakeRun(monkeypatch)
+
+
+def error_for(monkeypatch, stdout, stderr="", returncode=1):
+    FakeRun(monkeypatch, stdout, stderr, returncode)
+    with pytest.raises(cli.CliError) as exc:
+        cli.run("do", "start")
+    return exc.value
+
+
+# --- run / CliError ---------------------------------------------------------
+
+
+def test_run_passes_exact_subprocess_arguments(fake):
+    fake.result = ("ok\n", "", 0)
+    assert cli.run("get", "guests").strip() == "ok"
+    argv, kwargs = fake.calls[0]
+    assert argv == ["cv4pve-cli", "get", "guests"]
+    assert kwargs["capture_output"] is True
+    assert kwargs["text"] is True
+    assert kwargs["stdin"] == subprocess.DEVNULL
+    assert kwargs["timeout"] == 330
+
+
+def test_run_nonzero_exit_raises_clierror_with_stripped_output(monkeypatch):
+    err = error_for(monkeypatch, "  Error: boom\n\n", returncode=5)
+    assert str(err) == "Error: boom"
+
+
+def test_clierror_includes_stderr_when_stdout_is_empty(monkeypatch):
+    err = error_for(monkeypatch, "", stderr="Error: only on stderr\n")
+    assert "only on stderr" in str(err)
+
+
+def test_summary_multiline_sync_failure(monkeypatch):
+    out = (
+        "Error: Proxmox VE refused the call (HTTP 500):\n"
+        "  Configuration file 'nodes/mainframe/qemu-server/999.conf' does not exist\n"
+        "\n"
+        "Call: POST /nodes/mainframe/qemu/999/status/start\n"
+        "Alias: prod\n"
+    )
+    summary = error_for(monkeypatch, out).summary
+    assert "\n" not in summary
+    assert summary.startswith("Proxmox VE refused the call (HTTP 500):")
+    assert "Error:" not in summary
+    assert "Configuration file" in summary and "does not exist" in summary
+    assert "Call:" not in summary and "Alias:" not in summary
+
+
+def test_summary_skips_upid_line_exit_5(monkeypatch):
+    out = "UPID:mainframe:0001:0002:0003:qmstart:999:root@pam:\nConfiguration file 'x' does not exist\n"
+    summary = error_for(monkeypatch, out, returncode=5).summary
+    assert summary == "Configuration file 'x' does not exist"
+
+
+def test_summary_single_line_strips_error_prefix(monkeypatch):
+    assert error_for(monkeypatch, "Error: no such guest\n").summary == "no such guest"
+
+
+# --- kind / list_guests / guest_config --------------------------------------
+
+
+def test_kind_maps_type():
+    assert cli.kind({"type": "qemu"}) == "vm"
+    assert cli.kind({"type": "lxc"}) == "ct"
+
+
+def test_list_guests_argv_and_sorted_by_vmid(fake):
+    fake.result = (json.dumps([CT | {"vmid": 200}, VM, CT]), "", 0)
+    guests = cli.list_guests()
+    assert fake.argv == ["cv4pve-cli", "get", "guests", "-o", "json"]
+    assert [g["vmid"] for g in guests] == [100, 108, 200]
+
+
+def test_guest_config_flattens_key_value_list(fake):
+    fake.result = (
+        json.dumps([{"key": "cores", "value": 4}, {"key": "memory", "value": 8192}]),
+        "",
+        0,
+    )
+    assert cli.guest_config(VM) == {"cores": 4, "memory": 8192}
+    assert fake.argv == ["cv4pve-cli", "show", "vm", "mainframe", "108", "-o", "json"]
+
+
+def test_guest_config_uses_ct_for_lxc(fake):
+    fake.result = ("[]", "", 0)
+    cli.guest_config(CT)
+    assert fake.argv == ["cv4pve-cli", "show", "ct", "mainframe", "100", "-o", "json"]
+
+
+# --- disks ------------------------------------------------------------------
+
+
+def test_disks_vm_keeps_sized_disks_only():
+    config = {
+        "cores": 4,
+        "memory": 8192,
+        "scsi0": "local-lvm:vm-108-disk-0,discard=on,size=10G,ssd=1",
+        "virtio1": "local-lvm:vm-108-disk-2,size=32G",
+        "ide2": "local-lvm:vm-108-cloudinit,media=cdrom",
+        "sata0": "local-lvm:vm-108-cloudinit,size=4M",  # cloud-init drive without media=cdrom
+        "ide3": "none,media=cdrom",
+        "efidisk0": "local-lvm:vm-108-disk-1,efitype=4m,size=4M",
+        "net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0",
+    }
+    assert cli.disks(config, "vm") == {"scsi0": "10G", "virtio1": "32G"}
+
+
+def test_disks_ct_rootfs_and_mount_points():
+    config = {
+        "cores": 1,
+        "memory": 512,
+        "rootfs": "local-lvm:vm-100-disk-0,size=8G",
+        "mp0": "local-lvm:vm-100-disk-1,mp=/data,size=2G",
+        "net0": "name=eth0,bridge=vmbr0",
+    }
+    assert cli.disks(config, "ct") == {"rootfs": "8G", "mp0": "2G"}
+
+
+def test_disks_without_disks_is_empty():
+    assert cli.disks({"cores": 1}, "vm") == {}
+    assert cli.disks({"cores": 1}, "ct") == {}
+
+
+# --- action -----------------------------------------------------------------
+
+
+def test_actions_constant():
+    assert cli.ACTIONS == ("start", "shutdown", "stop", "reboot")
+
+
+@pytest.mark.parametrize("name", cli.ACTIONS)
+def test_action_argv_vm(fake, name):
+    cli.action(name, VM)
+    assert fake.argv == [
+        "cv4pve-cli",
+        "do",
+        name,
+        "vm",
+        "mainframe",
+        "108",
+        "-y",
+        "--wait",
+    ]
+
+
+def test_action_argv_ct(fake):
+    cli.action("stop", CT)
+    assert fake.argv == [
+        "cv4pve-cli",
+        "do",
+        "stop",
+        "ct",
+        "mainframe",
+        "100",
+        "-y",
+        "--wait",
+    ]
+
+
+def test_action_unknown_name_raises_and_does_not_call(fake):
+    with pytest.raises(ValueError):
+        cli.action("destroy", VM)
+    assert fake.calls == []
+
+
+def test_action_failure_raises_clierror(monkeypatch):
+    FakeRun(monkeypatch, "UPID:x:\nstart failed\n", returncode=5)
+    with pytest.raises(cli.CliError):
+        cli.action("start", VM)
+
+
+# --- set_resources ----------------------------------------------------------
+
+
+def test_set_resources_cores_and_memory(fake):
+    cli.set_resources(VM, cores=4, memory_mb=8192)
+    assert fake.argv == [
+        "cv4pve-cli", "set", "vm", "config", "mainframe", "108",
+        "--cores", "4", "--memory", "8192", "--wait",
+    ]  # fmt: skip
+
+
+def test_set_resources_only_given_fields(fake):
+    cli.set_resources(CT, memory_mb=1024)
+    assert fake.argv == [
+        "cv4pve-cli",
+        "set",
+        "ct",
+        "config",
+        "mainframe",
+        "100",
+        "--memory",
+        "1024",
+        "--wait",
+    ]
+    fake.calls.clear()
+    cli.set_resources(CT, cores=2)
+    assert fake.argv == [
+        "cv4pve-cli",
+        "set",
+        "ct",
+        "config",
+        "mainframe",
+        "100",
+        "--cores",
+        "2",
+        "--wait",
+    ]
+
+
+def test_set_resources_nothing_to_set(fake):
+    with pytest.raises(ValueError):
+        cli.set_resources(VM)
+    assert fake.calls == []
+
+
+@pytest.mark.parametrize(
+    "kwargs", [{"cores": 0}, {"cores": -1}, {"memory_mb": 0}, {"memory_mb": -512}]
+)
+def test_set_resources_rejects_non_positive(fake, kwargs):
+    with pytest.raises(ValueError):
+        cli.set_resources(VM, **kwargs)
+    assert fake.calls == []
+
+
+# --- resize_disk ------------------------------------------------------------
+
+
+def test_resize_disk_argv(fake):
+    cli.resize_disk(VM, "scsi0", 5)
+    assert fake.argv == [
+        "cv4pve-cli",
+        "do",
+        "resize",
+        "vm",
+        "disk",
+        "mainframe",
+        "108",
+        "scsi0",
+        "+5G",
+        "--wait",
+    ]
+
+
+def test_resize_disk_ct_argv(fake):
+    cli.resize_disk(CT, "rootfs", 2)
+    assert fake.argv == [
+        "cv4pve-cli",
+        "do",
+        "resize",
+        "ct",
+        "disk",
+        "mainframe",
+        "100",
+        "rootfs",
+        "+2G",
+        "--wait",
+    ]
+
+
+@pytest.mark.parametrize("grow", [0, -1, -10])
+def test_resize_disk_growth_must_be_positive(fake, grow):
+    with pytest.raises(ValueError):
+        cli.resize_disk(VM, "scsi0", grow)
+    assert fake.calls == []
+
+
+# --- pending_keys -----------------------------------------------------------
+
+
+def test_pending_keys_only_non_null_pending(fake):
+    fake.result = (
+        json.dumps([
+            {"key": "cores", "value": 4, "pending": None, "delete": None},
+            {"key": "memory", "value": 8192, "pending": 16384},
+            {"key": "balloon", "value": 0, "pending": 0},
+        ]),
+        "",
+        0,
+    )  # fmt: skip
+    assert sorted(cli.pending_keys(VM)) == ["balloon", "memory"]
+    assert fake.argv == [
+        "cv4pve-cli",
+        "api",
+        "get",
+        "/nodes/mainframe/qemu/108/pending",
+        "-o",
+        "json",
+    ]
+
+
+def test_pending_keys_ct_path_and_none_pending(fake):
+    fake.result = (json.dumps([{"key": "cores", "value": 1, "pending": None}]), "", 0)
+    assert cli.pending_keys(CT) == []
+    assert fake.argv[3] == "/nodes/mainframe/lxc/100/pending"
+
+
+def test_missing_binary_or_timeout_becomes_cli_error(monkeypatch):
+    for exc in (
+        FileNotFoundError("cv4pve-cli"),
+        subprocess.TimeoutExpired("cv4pve-cli", 330),
+    ):
+
+        def boom(*a, _exc=exc, **k):
+            raise _exc
+
+        monkeypatch.setattr(subprocess, "run", boom)
+        with pytest.raises(cli.CliError):
+            cli.run("get", "version")
+
+
+@pytest.mark.parametrize("raw", [{}, {"name": None}, {"name": ""}])
+def test_list_guests_fills_missing_name(fake, raw):
+    guest = {k: v for k, v in VM.items() if k != "name"} | raw
+    fake.result = (json.dumps([guest]), "", 0)
+    assert cli.list_guests()[0]["name"] == "VM 108"
+
+
+def test_status_reads_live_status_current(fake):
+    fake.result = (
+        json.dumps(
+            [{"key": "cpus", "value": 1}, {"key": "status", "value": "stopped"}]
+        ),
+        "",
+        0,
+    )
+    assert cli.status(CT) == "stopped"
+    assert fake.argv == [
+        "cv4pve-cli",
+        "api",
+        "get",
+        "/nodes/mainframe/lxc/100/status/current",
+        "-o",
+        "json",
+    ]
+
+
+# --- contexts / use_context (spec: "Context switching" > cli.py) -------------
+
+CONTEXT_LIST = (
+    "* mainframe            mainframe.iot.home.arpa:8006   (api-token)\n"
+    "  mainframe-edge       mainframe-edge.home.arpa:8006   (api-token)\n"
+)
+
+
+def test_contexts_parses_current_name_host_auth(fake):
+    fake.result = (CONTEXT_LIST, "", 0)
+    assert cli.contexts() == [
+        {"name": "mainframe", "host": "mainframe.iot.home.arpa:8006",
+         "auth": "api-token", "current": True},
+        {"name": "mainframe-edge", "host": "mainframe-edge.home.arpa:8006",
+         "auth": "api-token", "current": False},
+    ]  # fmt: skip
+
+
+def test_contexts_runs_config_list(fake):
+    fake.result = (CONTEXT_LIST, "", 0)
+    cli.contexts()
+    assert fake.argv == ["cv4pve-cli", "config", "list"]
+
+
+def test_contexts_ignores_junk_and_empty_lines(fake):
+    fake.result = (
+        "\nContexts:\n" + CONTEXT_LIST + "\n   \nnot a context line\n",
+        "",
+        0,
+    )
+    assert [c["name"] for c in cli.contexts()] == ["mainframe", "mainframe-edge"]
+
+
+def test_contexts_empty_output_gives_empty_list(fake):
+    assert cli.contexts() == []
+
+
+def test_use_context_runs_config_use_with_name(fake):
+    cli.use_context("mainframe-edge")
+    assert fake.argv == ["cv4pve-cli", "config", "use", "mainframe-edge"]
+
+
+def test_use_context_unknown_name_raises_clierror(monkeypatch):
+    FakeRun(monkeypatch, "Error: Context 'x' not found.\n", returncode=3)
+    with pytest.raises(cli.CliError) as exc:
+        cli.use_context("x")
+    assert "Context 'x' not found" in str(exc.value)

uv.lock

@@ -0,0 +1,195 @@
+version = 1
+revision = 5
+requires-python = ">=3.12"
+
+[[package]]
+name = "colorama"
+version = "0.4.6"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
+]
+
+[[package]]
+name = "cv4pve-tui"
+version = "0.1.0"
+source = { editable = "." }
+dependencies = [
+    { name = "textual" },
+]
+
+[package.dev-dependencies]
+dev = [
+    { name = "pytest" },
+    { name = "pytest-asyncio" },
+]
+
+[package.metadata]
+requires-dist = [{ name = "textual", specifier = ">=8.2" }]
+
+[package.metadata.requires-dev]
+dev = [
+    { name = "pytest", specifier = ">=8" },
+    { name = "pytest-asyncio", specifier = ">=1.0" },
+]
+
+[[package]]
+name = "iniconfig"
+version = "2.3.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/01/e1/2069291243c926a2ff1cd706c7f3eeb9b62144bf60f77c9fb9ff2fb26bd3/iniconfig-2.3.1.tar.gz", hash = "sha256:67f4b9c50da0dedf52af349e7749a80a9057a5031199791b906c3bb3ae878960", size = 21209, upload-time = "2026-10-06T22:48:38.076Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/56/43/4ca9e49d27a1fcf6bece6f6aec0ea46bb9112489b93d4b688fb415457bdb/iniconfig-2.3.1-py3-none-any.whl", hash = "sha256:9121e2c1fdb355232495be3194c8dfe87ccc2d5dee45947b78e68f499790d7a7", size = 7552, upload-time = "2026-10-06T22:48:36.959Z" },
+]
+
+[[package]]
+name = "linkify-it-py"
+version = "2.2.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/45/98/7a1a5f31fd5c7ba93e963b168e244b8e3dd705b3d2a718e3c3307583bf57/linkify_it_py-2.2.0.tar.gz", hash = "sha256:907acd2d17ac1fbb9ddb62c8957ccbd6158cac602231a15c3b0cd1e215f03cee", size = 32939, upload-time = "2026-08-29T07:07:08.305Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/13/d4/1152d1c7ab42d8b908be64fd200ddc870dc9d4925e951198702084aa1a7d/linkify_it_py-2.2.0-py3-none-any.whl", hash = "sha256:3adc40eb5af300b2605fcfdb968c24e1d780a90f1f2221af7c15e5111e94d443", size = 21971, upload-time = "2026-08-29T07:07:07.164Z" },
+]
+
+[[package]]
+name = "markdown-it-py"
+version = "4.2.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "mdurl" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" },
+]
+
+[package.optional-dependencies]
+linkify = [
+    { name = "linkify-it-py" },
+]
+
+[[package]]
+name = "mdit-py-plugins"
+version = "0.6.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "markdown-it-py" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/59/fc/f8d0863f8862f25602c0404d75568e89fb6b4109804645e5cdfb1be5cf56/mdit_py_plugins-0.6.1.tar.gz", hash = "sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0", size = 56114, upload-time = "2026-05-13T09:03:38.91Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/a5/69/6da5581c6a7fede7dc261bf4e67d6adca4196f176b43288b55b3db395b6e/mdit_py_plugins-0.6.1-py3-none-any.whl", hash = "sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d", size = 66663, upload-time = "2026-05-13T09:03:37.76Z" },
+]
+
+[[package]]
+name = "mdurl"
+version = "0.1.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" },
+]
+
+[[package]]
+name = "packaging"
+version = "26.3"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412, upload-time = "2026-08-04T18:15:28.737Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" },
+]
+
+[[package]]
+name = "platformdirs"
+version = "4.14.3"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/9a/6e/886dc7c9d21e2d95aceb696f1e31271742033aeef3661f6e9c8921493592/platformdirs-4.14.3.tar.gz", hash = "sha256:021add1ab113483edbcd4694f5fbb2d688b893ea30d35e1155bb371cf4be0be8", size = 71124, upload-time = "2026-11-01T16:27:32.928Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/93/41/f0aa003909ee77a9e7e55f8659abb635ca23ab0d022a449bbb18da87a1ef/platformdirs-4.14.3-py3-none-any.whl", hash = "sha256:cba2d05fe5e18b26b1fa6f376b8e6234c012abe1f8696c486fface7d0054e931", size = 35276, upload-time = "2026-11-01T16:27:31.498Z" },
+]
+
+[[package]]
+name = "pluggy"
+version = "1.7.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/bf/db/7fc19e6f2dc92a966727031389fc2e08b558f0f25eb7403c1119ad4713cd/pluggy-1.7.0.tar.gz", hash = "sha256:d1eaa46ebb595891b860ab086b4d09c8588af65ebd4361b8e8f4bb8920b90ba8", size = 123304, upload-time = "2026-10-15T09:50:58.343Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/40/9e/2b38731e0fc536806f16490e1a12d7f0dc2a1235aa8cc07bcc75416a7daa/pluggy-1.7.0-py3-none-any.whl", hash = "sha256:7dd7b0d8832ba3cb632c306926ded123429211b83641b35dc5c41ad2d34f9bec", size = 27082, upload-time = "2026-10-15T09:50:56.808Z" },
+]
+
+[[package]]
+name = "pygments"
+version = "2.21.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" },
+]
+
+[[package]]
+name = "pytest"
+version = "9.2.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "colorama", marker = "sys_platform == 'win32'" },
+    { name = "iniconfig" },
+    { name = "packaging" },
+    { name = "pluggy" },
+    { name = "pygments" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/79/42/bb4641652e58dedeac2ec51abc1f8cae4fee4ec63899a0999016229a0f2a/pytest-9.2.0.tar.gz", hash = "sha256:97c4f3b1f3c7414aa36465e09b8220f1d03627993bd51b1282e193753c80557b", size = 1797555, upload-time = "2026-10-27T21:30:20.728Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/51/39/1f5d8c8877a0fa327db959c19341d4be61660de1ab6b7391f9eb805bcd67/pytest-9.2.0-py3-none-any.whl", hash = "sha256:07acdcf88ec174f907dc01af70bf52d49b014bbab03faeeade31c96b18c2e588", size = 411776, upload-time = "2026-10-27T21:30:19.131Z" },
+]
+
+[[package]]
+name = "pytest-asyncio"
+version = "1.4.1.post1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pytest" },
+    { name = "typing-extensions", marker = "python_full_version < '3.13'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/73/7f/c60a4338420b33453d4f814ac0c930abd399f162b9bdd21af4b9e363037f/pytest_asyncio-1.4.1.post1.tar.gz", hash = "sha256:f582d9ea4a1c082195abe8bd29a1fe106aeb1cd435e3dad2d3bc2749935c36a2", size = 145360, upload-time = "2026-11-01T18:26:03.732Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e3/de/adbc7578959cfab07e5e7c05cf6d54224a51e7667ea2c07458b6bc571e75/pytest_asyncio-1.4.1.post1-py3-none-any.whl", hash = "sha256:0755e488cc0a2b9bd2eab7292cc6390d4cbb8390f1020745647bae5a708e0165", size = 17140, upload-time = "2026-11-01T18:26:02.168Z" },
+]
+
+[[package]]
+name = "rich"
+version = "15.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "markdown-it-py" },
+    { name = "pygments" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" },
+]
+
+[[package]]
+name = "textual"
+version = "8.2.8"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "markdown-it-py", extra = ["linkify"] },
+    { name = "mdit-py-plugins" },
+    { name = "platformdirs" },
+    { name = "pygments" },
+    { name = "rich" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/00/21/39a76b01bd5eea82a04baaca7580e105d8c59450df03998345bb2cfb307b/textual-8.2.8.tar.gz", hash = "sha256:3f106a9fbc73e39dd266c9712432087de78a6d644084c7c241d6a25c3169115b", size = 1860502, upload-time = "2026-06-30T06:51:24.495Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/fb/be/35261223d9416a0751cdff1c7b4a6f881387218a12d439fe22fefebc8c04/textual-8.2.8-py3-none-any.whl", hash = "sha256:267375fd402dc8d981457212efa71f0e3365fd17bba144ba9bb3ed7563cb374a", size = 731418, upload-time = "2026-06-30T06:51:26.364Z" },
+]
+
+[[package]]
+name = "typing-extensions"
+version = "4.16.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f6/cc/6253133b5bb138fc3306cebfbda2c520f545d36b5be2c7255cc528bb45d6/typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5", size = 113555, upload-time = "2026-07-02T08:40:05.92Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/49/d3/b8441a820a491ddfc024b0b0cf0393375b75ea13866d9c66727e54c2fc80/typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", size = 45571, upload-time = "2026-07-02T08:40:04.659Z" },
+]