diff --git a/README.md b/README.md
index 1a65006..38da74f 100644
--- a/README.md
+++ b/README.md
@@ -29,6 +29,7 @@ uv run cv4pve-tui
 | `Enter` then `p` / `R` / `D` | in info: take a snapshot (name, description, VM: include RAM) / roll back to the snapshot under the cursor (confirms; starts the guest again if it is running) / delete the snapshot under the cursor (confirms) |
 | `t` | shell in a running LXC container (`ssh -t root@<node> pct enter <vmid>`) |
 | `l` | tasks: recent cluster tasks (start, duration, node, type, VMID, user, status; `!` marks failures); `Enter` shows the log of the task under the cursor, `g` reloads. In info: `l` = only this guest's tasks |
+| `U` | package updates: pending updates per online node (node, package, current, new, origin, priority), read-only; `g` reloads. The list is as of the node's last `apt update`. The frame's bottom border shows the count |
 | `g` | refresh |
 | `c` | switch cv4pve-cli context (cluster) |
 | `q` | quit |
diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md
index f140b7b..93fa717 100644
--- a/docs/plans/guest-tui-mvp.md
+++ b/docs/plans/guest-tui-mvp.md
@@ -591,3 +591,41 @@ CLI facts (read-only probes):
   a read-only plain-text log widget `#tasklog` (no markup parsing), loaded in a worker;
   failure -> error notify. `escape` closes.
 - Main-screen footer shows `l Tasks`.
+
+---
+
+# Node package updates (issue #13)
+
+## Decision (user-approved)
+
+`U` on the main screen shows pending package updates per node; the frame's bottom
+border shows the count. Read-only: running `apt update`/`upgrade` stays out of scope.
+
+CLI facts (read-only probes):
+
+- `get nodes -o json` -> list of dicts with `node`, `status` (`online`/`offline`), ...
+- `get node packages <node> -o json` -> list of dicts with `Package`, `OldVersion`,
+  `Version`, `Title`, `Priority`, `Origin`, ... (23 on mainframe). The list is as fresh
+  as the node's last `apt update` (Proxmox runs it daily); the call does not refresh it.
+
+### `cli.py`
+
+- `nodes() -> list[str]` - names of online nodes, sorted.
+- `package_updates(node) -> list[dict]` - `get node packages <node> -o json`, sorted by
+  `Package`.
+
+### `app.py` (UI contract)
+
+- `UpdatesScreen(ModalScreen)` opened by `U`: frame `#updates_section` titled "Package
+  updates", `DataTable#updates`: Node, Package, Current (`OldVersion`), New
+  (`Version`), Origin, Priority; `Static#updates_note`: "No pending updates" when
+  empty, plus one line per node that failed ("<node>: unavailable: <summary>") and a
+  fixed hint that the list is as of the node's last apt update. Loaded in a thread
+  worker; `g` reloads; `escape` closes. A failing node does not hide the others.
+- The main frame's bottom border adds "· N package updates" when N > 0, after the
+  backup count. The count is fetched in a worker on mount, after a context switch, and
+  every time the UpdatesScreen loads - not on every guest reload (one CLI call per node).
+  It follows the same generation guard as the guest list (a count from the old cluster
+  is dropped after a switch) and is cleared when a switch starts. Failures leave the
+  count unknown (nothing shown), no error notify on mount.
+- Main footer shows `U Updates`.
diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py
index a801c60..d1d5cde 100644
--- a/src/cv4pve_tui/app.py
+++ b/src/cv4pve_tui/app.py
@@ -536,6 +536,81 @@ class TasksScreen(ModalScreen):
         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."""
 
@@ -932,6 +1007,7 @@ class GuestApp(App):
         ("g", "reload", "Refresh"),
         ("c", "context", "Context"),
         ("l", "tasks", "Tasks"),
+        ("U", "updates", "Updates"),
         ("q", "quit", "Quit"),
     ]
 
@@ -945,6 +1021,7 @@ class GuestApp(App):
         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 {
@@ -979,10 +1056,34 @@ class GuestApp(App):
         )
         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()
+
     def _show_context_frame(self, sub_title: str) -> None:
         self.context_name = sub_title.removeprefix("context: ")
         self._update_frame()
@@ -990,6 +1091,9 @@ class GuestApp(App):
     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)
@@ -1112,6 +1216,8 @@ class GuestApp(App):
         ).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)
@@ -1127,6 +1233,7 @@ class GuestApp(App):
             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)
 
diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py
index 75be38a..ca590ba 100644
--- a/src/cv4pve_tui/cli.py
+++ b/src/cv4pve_tui/cli.py
@@ -456,6 +456,18 @@ def tasks() -> list[dict]:
     return sorted(items, key=lambda t: t.get("starttime") or 0, reverse=True)
 
 
+def nodes() -> list[str]:
+    """Names of the online nodes, sorted."""
+    items = json.loads(run("get", "nodes", "-o", "json"))
+    return sorted(n["node"] for n in items if n.get("status") == "online")
+
+
+def package_updates(node: str) -> list[dict]:
+    """Pending package updates of a node (as of its last `apt update`), by package."""
+    items = json.loads(run("get", "node", "packages", node, "-o", "json"))
+    return sorted(items, key=lambda p: p.get("Package", ""))
+
+
 def task_log(upid: str) -> str:
     if not isinstance(upid, str) or not upid.startswith("UPID:"):
         raise ValueError("not a task id (UPID)")
diff --git a/tests/test_app.py b/tests/test_app.py
index 2d03797..5a9b040 100644
--- a/tests/test_app.py
+++ b/tests/test_app.py
@@ -80,6 +80,12 @@ class Fake:
         ]  # newest first, as cli.tasks returns them
         self.tasks_error, self.tasks_calls = None, 0
         self.log_text, self.log_error, self.log_calls = "TASK OK\n", None, []
+        # issue #13 (node package updates): node -> rows; one online node, nothing pending
+        self.updates = {"mainframe": []}
+        self.updates_by_context = {}  # context name -> {node: rows}, overrides self.updates
+        self.nodes_error, self.updates_errors = None, {}  # node -> exception
+        self.nodes_calls, self.updates_calls = 0, []
+        self.updates_hook = None
 
     def list_guests(self):
         self.list_calls += 1
@@ -225,6 +231,26 @@ class Fake:
             raise self.tasks_error
         return [dict(t) for t in self.task_list]
 
+    def _cluster(self):
+        return self.updates_by_context.get(self.current, self.updates)
+
+    def nodes(self):
+        self.nodes_calls += 1
+        if self.nodes_error:
+            raise self.nodes_error
+        return sorted(self._cluster())
+
+    def package_updates(self, node):
+        self.updates_calls.append(node)
+        rows = [dict(r) for r in self._cluster()[node]]
+        if (
+            self.updates_hook
+        ):  # runs after reading: simulates a slow reply from that cluster
+            self.updates_hook()
+        if node in self.updates_errors:
+            raise self.updates_errors[node]
+        return rows
+
     def task_log(self, upid):
         self.log_calls.append(upid)
         if self.log_error:
@@ -287,6 +313,8 @@ def fake(monkeypatch):
         "delete_snapshot",
         "tasks",
         "task_log",
+        "nodes",
+        "package_updates",
     ):
         monkeypatch.setattr(cli, name, getattr(f, name), raising=False)
     for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
@@ -3171,3 +3199,275 @@ async def test_finished_task_with_empty_status_is_not_shown_as_running(fake):
         row = [str(c) for c in table.get_row_at(0)]
         assert "running" not in row
         assert row[0] == "!"  # finished without OK: flagged
+
+
+# --- #13: node package updates ------------------------------------------------------------
+# spec: docs/plans/guest-tui-mvp.md ("Node package updates (issue #13)")
+
+
+def pkg(name, old="1.0", new="2.0", origin="Debian", priority="optional"):
+    return {"Package": name, "OldVersion": old, "Version": new, "Title": f"{name} title",
+            "Priority": priority, "Origin": origin}  # fmt: skip
+
+
+def many(n, prefix="pkg"):
+    return [pkg(f"{prefix}{i:02d}") for i in range(n)]
+
+
+def frame_subtitle(app):
+    return str(app.query_one("#guests_section").border_subtitle)
+
+
+async def open_updates(app, pilot):
+    await pilot.press("U")
+    await settle(app, pilot)
+    assert isinstance(app.screen, appmod.UpdatesScreen)
+    return app.screen.query_one("#updates", DataTable)
+
+
+def note_of(app):
+    return str(app.screen.query_one("#updates_note", Static).render())
+
+
+def two_nodes(fake):
+    fake.updates = {
+        "mainframe": [pkg("curl", "7.1", "7.2", "Debian", "optional"),
+                      pkg("pve-manager", "8.1", "8.2", "Proxmox", "important")],
+        "edge": [pkg("bash", "5.1", "5.2", "Debian", "required")],
+    }  # fmt: skip
+
+
+async def test_u_lists_rows_across_nodes_with_all_columns(
+    fake,
+):  # UpdatesScreen rows + columns
+    two_nodes(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        table = await open_updates(app, pilot)
+        assert [str(c.label) for c in table.columns.values()] == [
+            "Node", "Package", "Current", "New", "Origin", "Priority"]  # fmt: skip
+        assert table.row_count == 3
+        assert {tuple(cells(table, i)) for i in range(3)} == {
+            ("mainframe", "curl", "7.1", "7.2", "Debian", "optional"),
+            ("mainframe", "pve-manager", "8.1", "8.2", "Proxmox", "important"),
+            ("edge", "bash", "5.1", "5.2", "Debian", "required"),
+        }
+        assert (
+            app.screen.query_one("#updates_section").border_title == "Package updates"
+        )
+
+
+async def test_updates_screen_shows_no_pending_updates_note_when_empty(
+    fake,
+):  # "No pending updates"
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        table = await open_updates(app, pilot)
+        assert table.row_count == 0
+        assert "No pending updates" in note_of(app)
+
+
+async def test_updates_screen_has_apt_update_freshness_hint(
+    fake,
+):  # fixed hint: as of last apt update
+    two_nodes(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        await open_updates(app, pilot)
+        assert "apt update" in note_of(app).lower()
+
+
+async def test_failing_node_is_unavailable_and_other_nodes_still_listed(
+    fake,
+):  # failing node != hidden others
+    two_nodes(fake)
+    err = cli.CliError("Error: node edge timed out")
+    fake.updates_errors = {"edge": err}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        table = await open_updates(app, pilot)
+        assert table.row_count == 2
+        assert {cells(table, i)[0] for i in range(2)} == {"mainframe"}
+        note = note_of(app)
+        assert "edge: unavailable" in note and err.summary in note
+        assert "mainframe: unavailable" not in note
+
+
+async def test_g_in_updates_screen_reloads(fake):  # `g` reloads
+    two_nodes(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        table = await open_updates(app, pilot)
+        before = len(fake.updates_calls)
+        fake.updates["edge"].append(pkg("zsh"))
+        await pilot.press("g")
+        await settle(app, pilot)
+        assert len(fake.updates_calls) > before
+        assert table.row_count == 4
+
+
+async def test_escape_closes_updates_screen(fake):  # `escape` closes
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        await open_updates(app, pilot)
+        await pilot.press("escape")
+        await settle(app, pilot)
+        assert not isinstance(app.screen, appmod.UpdatesScreen)
+        assert app.is_running
+
+
+async def test_frame_subtitle_shows_package_update_count_after_mount(
+    fake,
+):  # "· N package updates"
+    fake.updates = {"mainframe": many(23)}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert "· 23 package updates" in frame_subtitle(app)
+        assert "mainframe" in frame_subtitle(app)
+
+
+async def test_frame_subtitle_count_sums_nodes(fake):  # count over all online nodes
+    two_nodes(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert "· 3 package updates" in frame_subtitle(app)
+
+
+async def test_no_package_count_when_nothing_pending(fake):  # N > 0 only
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert frame_subtitle(app) == "mainframe"
+
+
+async def test_package_count_follows_backup_count_in_subtitle(
+    fake,
+):  # "after the backup count"
+    fake.unprotected = {100}
+    fake.updates = {"mainframe": many(23)}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        sub = frame_subtitle(app)
+        assert "1 without backup job" in sub and "· 23 package updates" in sub
+        assert sub.index("without backup job") < sub.index("package updates")
+
+
+async def test_plain_guest_reload_does_not_refetch_package_count(
+    fake,
+):  # not on every guest reload
+    fake.updates = {"mainframe": many(23)}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        nodes_before, pkg_before, list_before = (
+            fake.nodes_calls,
+            len(fake.updates_calls),
+            fake.list_calls,
+        )
+        assert nodes_before >= 1 and pkg_before >= 1
+        await pilot.press("g")
+        await settle(app, pilot)
+        assert fake.list_calls > list_before
+        assert (fake.nodes_calls, len(fake.updates_calls)) == (nodes_before, pkg_before)
+        assert "· 23 package updates" in frame_subtitle(app)
+
+
+async def test_opening_updates_screen_refreshes_the_frame_count(
+    fake,
+):  # count fetched on every screen load
+    fake.updates = {"mainframe": many(23)}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        fake.updates["mainframe"] = many(5)
+        await open_updates(app, pilot)
+        assert "· 5 package updates" in frame_subtitle(app)
+        assert "23 package updates" not in frame_subtitle(app)
+
+
+async def test_context_switch_clears_old_count_then_shows_new_clusters_count(
+    fake,
+):  # cleared at switch start
+    fake.updates = {"mainframe": many(23)}
+    fake.updates_by_context = {"mainframe-edge": {"edge": many(4)}}
+    fake.guests_by_context = {"mainframe-edge": [EDGE]}
+    started, release = threading.Event(), threading.Event()
+    fake.use_hook = lambda: (started.set(), release.wait(10))
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert "23 package updates" in frame_subtitle(app)
+        try:
+            await switch_to_edge(app, pilot)
+            await until(pilot, started.is_set)
+            await pilot.pause()
+            assert "package updates" not in frame_subtitle(app)
+        finally:
+            release.set()
+        await settle(app, pilot)
+        sub = frame_subtitle(app)
+        assert "· 4 package updates" in sub and "23" not in sub
+
+
+async def test_count_fetched_before_a_switch_but_arriving_after_it_is_dropped(
+    fake,
+):  # generation guard
+    fake.updates = {"mainframe": many(23)}
+    fake.updates_by_context = {"mainframe-edge": {"edge": many(4)}}
+    fake.guests_by_context = {"mainframe-edge": [EDGE]}
+    hook, started, release = blocking_once()
+    fake.updates_hook = hook  # the mount-time count reads mainframe, then stalls
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        try:
+            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)
+        sub = frame_subtitle(app)
+        assert "23" not in sub
+        assert "· 4 package updates" in sub
+
+
+@pytest.mark.parametrize("failure", ["nodes", "node"])
+async def test_count_failure_on_mount_shows_nothing_and_does_not_notify(
+    fake, failure
+):  # unknown count
+    fake.updates = {"mainframe": many(23)}
+    if failure == "nodes":
+        fake.nodes_error = cli.CliError("Error: no route")
+    else:
+        fake.updates_errors = {"mainframe": cli.CliError("Error: no route")}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert "package updates" not in frame_subtitle(app)
+        assert errors(app) == []
+        assert app.query_one("#guests", DataTable).row_count == 2
+
+
+async def test_footer_on_main_screen_shows_u_updates(fake):  # footer `U Updates`
+    app = make_app()
+    async with app.run_test() as pilot:
+        await settle(app, pilot)
+        shown = [ab.binding for ab in app.active_bindings.values() if ab.binding.show]
+        assert any(b.key == "U" and b.description == "Updates" for b in shown)
+
+
+async def test_frame_subtitle_singular_for_one_package_update(fake):
+    fake.updates = {"mainframe": many(1)}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert frame_subtitle(app).endswith("· 1 package update")
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 6c9a0bd..e14a17f 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -1340,3 +1340,38 @@ def test_task_log_rejects_non_upid_without_calling_subprocess(
     with pytest.raises(ValueError):
         cli.task_log(bad)
     assert fake.calls == []
+
+
+# --- #13: nodes / package_updates (spec: "Node package updates (issue #13)") ---------------
+
+
+def test_nodes_argv_only_online_sorted(fake):
+    fake.result = (json.dumps([{"node": "zeta", "status": "online"},
+                               {"node": "down", "status": "offline"},
+                               {"node": "alpha", "status": "online"}]), "", 0)  # fmt: skip
+    assert cli.nodes() == ["alpha", "zeta"]
+    assert fake.argv == ["cv4pve-cli", "get", "nodes", "-o", "json"]
+
+
+def test_package_updates_argv_and_sorted_by_package(fake):
+    fake.result = (json.dumps([{"Package": "zsh", "Version": "2"},
+                               {"Package": "bash", "Version": "1"},
+                               {"Package": "curl", "Version": "3"}]), "", 0)  # fmt: skip
+    got = cli.package_updates("mainframe")
+    assert fake.argv == [
+        "cv4pve-cli",
+        "get",
+        "node",
+        "packages",
+        "mainframe",
+        "-o",
+        "json",
+    ]
+    assert [p["Package"] for p in got] == ["bash", "curl", "zsh"]
+    assert got[0] == {"Package": "bash", "Version": "1"}
+
+
+def test_package_updates_failure_raises_clierror(monkeypatch):
+    FakeRun(monkeypatch, "Error: node unreachable\n", returncode=1)
+    with pytest.raises(cli.CliError):
+        cli.package_updates("mainframe")