Merge develop into PR #337 and resolve shared app cache conflict

AppImage/scripts/audit_checks.py

@@ -0,0 +1,342 @@
+"""Check registry and evaluation engine for Audit & Report.
+
+A check declares an identifier, an area and the severity its failure
+carries, and returns the outcome of one evaluation. Checks never modify
+the host: an assessment reads, it does not act.
+
+Identifiers are ``<area>.<slug>`` and are frozen once published. Rewording
+a title never changes the identifier, because the accepted-risk register
+and the per-check history are keyed by it. A check whose meaning changes
+materially gets a new identifier and the old one is retired rather than
+reused, so a decision recorded months earlier still resolves.
+
+Checks read from ``AuditContext``, which collects each source once per run
+and hands the same result to every check that needs it. A full assessment
+runs against a production hypervisor, so repeating collection per check is
+not acceptable.
+"""
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import time
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+import audit_store
+
+# Report areas. These group the categories `health_monitor` already emits
+# so the two surfaces share one vocabulary instead of maintaining a
+# parallel taxonomy.
+AREA_SYSTEM = "system"
+AREA_STORAGE = "storage"
+AREA_NETWORK = "network"
+AREA_SECURITY = "security"
+AREA_BACKUP = "backup"
+AREA_GUESTS = "guests"
+AREA_HARDWARE = "hardware"
+
+AREAS = (
+    AREA_SYSTEM, AREA_STORAGE, AREA_NETWORK, AREA_SECURITY,
+    AREA_BACKUP, AREA_GUESTS, AREA_HARDWARE,
+)
+
+SEVERITIES = ("OK", "INFO", "WARNING", "CRITICAL")
+
+# Per-check wall-clock budget. A check that cannot answer within it is
+# recorded as not applicable rather than stalling the whole assessment.
+CHECK_TIMEOUT = 20
+
+
+class Check:
+    """One registered assessment.
+
+    ``evaluate`` receives the context and returns a dict with ``state``
+    and, optionally, ``summary``, ``affected``, ``evidence`` and
+    ``remediable_by``. Returning ``None`` marks the check as not
+    applicable on this host.
+    """
+
+    def __init__(self, check_id: str, area: str, severity: str,
+                 evaluate: Callable[["AuditContext"], Optional[dict]]):
+        if area not in AREAS:
+            raise ValueError(f"unknown area for {check_id}: {area}")
+        if severity not in SEVERITIES:
+            raise ValueError(f"unknown severity for {check_id}: {severity}")
+        if not check_id.startswith(f"{area}."):
+            raise ValueError(f"{check_id} must be prefixed with its area")
+        self.check_id = check_id
+        self.area = area
+        self.severity = severity
+        self.evaluate = evaluate
+
+
+_REGISTRY: dict[str, Check] = {}
+
+
+def register(check_id: str, area: str, severity: str):
+    """Decorator registering a check under a stable identifier."""
+    def wrap(fn):
+        if check_id in _REGISTRY:
+            raise ValueError(f"duplicate check identifier: {check_id}")
+        _REGISTRY[check_id] = Check(check_id, area, severity, fn)
+        return fn
+    return wrap
+
+
+def registered_checks() -> list[Check]:
+    return sorted(_REGISTRY.values(), key=lambda c: (c.area, c.check_id))
+
+
+# ---------------------------------------------------------------------------
+# Collection context
+# ---------------------------------------------------------------------------
+
+class AuditContext:
+    """Lazily collects each source once and shares it across checks."""
+
+    def __init__(self):
+        self._cache: dict[str, Any] = {}
+
+    def _once(self, key: str, producer: Callable[[], Any]) -> Any:
+        if key not in self._cache:
+            try:
+                self._cache[key] = producer()
+            except Exception:
+                self._cache[key] = None
+        return self._cache[key]
+
+    def run(self, cmd: list[str], timeout: int = 10) -> tuple[int, str]:
+        """Run a read-only command, returning exit code and output."""
+        key = f"cmd:{' '.join(cmd)}"
+        if key in self._cache:
+            return self._cache[key]
+        try:
+            proc = subprocess.run(cmd, capture_output=True, text=True,
+                                  timeout=timeout)
+            result = (proc.returncode, (proc.stdout or "") + (proc.stderr or ""))
+        except Exception as exc:
+            result = (-1, str(exc))
+        self._cache[key] = result
+        return result
+
+    @property
+    def lxc_configs(self) -> dict[int, str]:
+        """Raw text of every local container configuration."""
+        def load():
+            out: dict[int, str] = {}
+            base = Path("/etc/pve/lxc")
+            if not base.is_dir():
+                return out
+            for path in base.glob("*.conf"):
+                try:
+                    out[int(path.stem)] = path.read_text(errors="replace")
+                except (OSError, ValueError):
+                    continue
+            return out
+        return self._once("lxc_configs", load) or {}
+
+    @property
+    def qemu_configs(self) -> dict[int, str]:
+        def load():
+            out: dict[int, str] = {}
+            base = Path("/etc/pve/qemu-server")
+            if not base.is_dir():
+                return out
+            for path in base.glob("*.conf"):
+                try:
+                    out[int(path.stem)] = path.read_text(errors="replace")
+                except (OSError, ValueError):
+                    continue
+            return out
+        return self._once("qemu_configs", load) or {}
+
+    @property
+    def apt_sources(self) -> dict[str, str]:
+        """Contents of the apt source files that define PVE repositories."""
+        def load():
+            out: dict[str, str] = {}
+            candidates = [Path("/etc/apt/sources.list")]
+            d = Path("/etc/apt/sources.list.d")
+            if d.is_dir():
+                candidates.extend(sorted(d.glob("*.list")))
+                candidates.extend(sorted(d.glob("*.sources")))
+            for path in candidates:
+                try:
+                    out[str(path)] = path.read_text(errors="replace")
+                except OSError:
+                    continue
+            return out
+        return self._once("apt_sources", load) or {}
+
+    @property
+    def vzdump_jobs(self) -> str:
+        """Raw backup job definitions from the cluster configuration."""
+        def load():
+            text = ""
+            for path in (Path("/etc/pve/jobs.cfg"), Path("/etc/vzdump.cron")):
+                try:
+                    text += path.read_text(errors="replace") + "\n"
+                except OSError:
+                    continue
+            return text
+        return self._once("vzdump_jobs", load) or ""
+
+    @property
+    def storages(self) -> list[dict]:
+        """Storage definitions from ``storage.cfg``.
+
+        Each entry keeps its type, identifier and settings. ``shared``
+        matters to anything that reasons about ownership: on shared
+        storage a volume may belong to a guest running on another node,
+        which is invisible from here.
+        """
+        def load():
+            out: list[dict] = []
+            try:
+                text = Path("/etc/pve/storage.cfg").read_text(errors="replace")
+            except OSError:
+                return out
+            current: Optional[dict] = None
+            for line in text.splitlines():
+                if not line.strip():
+                    continue
+                header = re.match(r"^(\w+):\s*(\S+)", line)
+                if header:
+                    current = {"type": header.group(1), "id": header.group(2)}
+                    out.append(current)
+                    continue
+                if current is None or not line[:1].isspace():
+                    continue
+                parts = line.strip().split(None, 1)
+                if parts:
+                    current[parts[0]] = parts[1] if len(parts) > 1 else ""
+            return out
+        return self._once("storages", load) or []
+
+    @property
+    def pve_user_cfg(self) -> str:
+        """Raw access-control configuration, which also defines pools."""
+        def load():
+            try:
+                return Path("/etc/pve/user.cfg").read_text(errors="replace")
+            except OSError:
+                return ""
+        return self._once("pve_user_cfg", load) or ""
+
+
+# ---------------------------------------------------------------------------
+# Evaluation
+# ---------------------------------------------------------------------------
+
+def run_assessment(profile: str = "full",
+                   only_areas: Optional[set[str]] = None) -> str:
+    """Evaluate every registered check and persist the result.
+
+    A check that raises is recorded as not applicable with the error kept
+    as evidence. One faulty check must never abort an assessment: a
+    partial report that says which check failed is more useful than no
+    report at all.
+    """
+    ctx = AuditContext()
+    exceptions = audit_store.active_exceptions()
+    run_id = audit_store.start_run(profile)
+    findings: list[dict[str, Any]] = []
+    error: Optional[str] = None
+
+    try:
+        for check in registered_checks():
+            if only_areas and check.area not in only_areas:
+                continue
+            started = time.monotonic()
+            try:
+                result = check.evaluate(ctx)
+            except Exception as exc:
+                result = {
+                    "state": audit_store.STATE_NOT_APPLICABLE,
+                    "summary_key": "evaluationFailed",
+                    "evidence": f"{type(exc).__name__}: {exc}",
+                }
+            elapsed = time.monotonic() - started
+
+            if result is None:
+                result = {"state": audit_store.STATE_NOT_APPLICABLE}
+
+            state = result.get("state", audit_store.STATE_NOT_APPLICABLE)
+            # An accepted risk keeps its evidence and its declared
+            # severity; only the state changes, so the report can still
+            # show what was accepted and why it mattered.
+            if state in (audit_store.STATE_FAIL, audit_store.STATE_WARN) \
+                    and check.check_id in exceptions:
+                state = audit_store.STATE_ACCEPTED
+
+            evidence = result.get("evidence")
+            if elapsed > CHECK_TIMEOUT:
+                evidence = (evidence or "") + \
+                    f"\n[check exceeded its time budget: {elapsed:.1f}s]"
+
+            findings.append({
+                "check_id": check.check_id,
+                "area": check.area,
+                "severity": check.severity,
+                "state": state,
+                "summary_key": result.get("summary_key"),
+                "summary_params": result.get("summary_params") or {},
+                "affected": result.get("affected") or [],
+                "evidence": evidence,
+                "remediable_by": result.get("remediable_by"),
+            })
+    except Exception as exc:
+        error = f"{type(exc).__name__}: {exc}"
+
+    audit_store.record_findings(run_id, findings)
+    audit_store.finish_run(run_id, checks_total=len(findings), error=error)
+    return run_id
+
+
+def compare_runs(base_run: str, other_run: str) -> dict[str, list[dict]]:
+    """Classify how findings moved between two runs.
+
+    A finding that stopped failing because someone accepted it is reported
+    separately from one that stopped failing because the host changed.
+    Both leave the active set, but only the second is a fix, and a report
+    that merges them would tell its reader the problem went away when the
+    decision was to live with it.
+
+    ``unchanged`` is kept so a report can state that the rest of the
+    surface held steady rather than leaving it unaccounted for.
+    """
+    failing = {audit_store.STATE_FAIL, audit_store.STATE_WARN}
+    base = {f["check_id"]: f for f in audit_store.get_findings(base_run)}
+    other = {f["check_id"]: f for f in audit_store.get_findings(other_run)}
+
+    new, resolved, accepted, unchanged = [], [], [], []
+    for check_id, current in other.items():
+        previous = base.get(check_id)
+        was = previous["state"] in failing if previous else False
+        now = current["state"] in failing
+        if now and not was:
+            new.append(current)
+        elif was and not now:
+            if current["state"] == audit_store.STATE_ACCEPTED:
+                accepted.append(current)
+            else:
+                resolved.append(current)
+        elif previous and previous["state"] == current["state"]:
+            unchanged.append(current)
+    # A check present in the base run but absent from the later one was
+    # retired between the two. It is reported as no longer assessed rather
+    # than as resolved, since nothing verified that it stopped failing.
+    retired = [
+        previous for check_id, previous in base.items()
+        if check_id not in other and previous["state"] in failing
+    ]
+
+    return {
+        "new": new,
+        "resolved": resolved,
+        "accepted": accepted,
+        "unchanged": unchanged,
+        "retired": retired,
+    }

AppImage/scripts/audit_checks_pve.py

@@ -0,0 +1,430 @@
+"""Proxmox-specific checks for Audit & Report.
+
+Importing this module registers its checks. Everything here reads the
+host and reports; nothing modifies it.
+
+The checks are deliberately about configuration and posture rather than
+transient load. A condition that resolves on its own as usage drops
+belongs to the health monitor, which keeps its own catalogue and remains
+the source of notifications.
+"""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import audit_store
+from audit_checks import (
+    AREA_BACKUP, AREA_GUESTS, AREA_SECURITY, AREA_STORAGE, AREA_SYSTEM,
+    register,
+)
+
+FAIL = audit_store.STATE_FAIL
+WARN = audit_store.STATE_WARN
+PASS = audit_store.STATE_PASS
+NA = audit_store.STATE_NOT_APPLICABLE
+
+
+# ---------------------------------------------------------------------------
+# Backup
+# ---------------------------------------------------------------------------
+
+def _parse_vzdump_jobs(text: str) -> list[dict]:
+    """Split ``jobs.cfg`` into one entry per backup job.
+
+    A job opens with ``vzdump: <id>`` and its settings follow as indented
+    ``key value`` lines. Values are kept verbatim; interpretation belongs
+    to the caller.
+    """
+    jobs: list[dict] = []
+    current: dict | None = None
+    for line in text.splitlines():
+        if not line.strip():
+            continue
+        header = re.match(r"^vzdump:\s*(\S+)", line)
+        if header:
+            current = {"id": header.group(1)}
+            jobs.append(current)
+            continue
+        if current is None or not line[:1].isspace():
+            continue
+        parts = line.strip().split(None, 1)
+        if parts:
+            current[parts[0]] = parts[1] if len(parts) > 1 else ""
+    return jobs
+
+
+def _pool_members(text: str) -> dict[str, set[int]]:
+    """Map pool name to member guest identifiers from ``user.cfg``.
+
+    Pool entries are colon-separated: ``pool:<name>:<comment>:<vmids>:``.
+    """
+    pools: dict[str, set[int]] = {}
+    for line in (text or "").splitlines():
+        if not line.startswith("pool:"):
+            continue
+        fields = line.split(":")
+        if len(fields) < 4:
+            continue
+        pools[fields[1]] = {int(x) for x in re.findall(r"\d+", fields[3])}
+    return pools
+
+
+@register("backup.guest_coverage", AREA_BACKUP, "CRITICAL")
+def _guest_coverage(ctx):
+    """Guests that no enabled backup job includes.
+
+    A job selects guests by enumerating them (``vmid``), by taking every
+    guest (``all 1``), or by pool, and may subtract an ``exclude`` list.
+    A job carrying ``enabled 0`` selects nothing: it is defined but never
+    runs, which is precisely the situation this check exists to surface,
+    since a disabled job looks like coverage in the interface.
+    """
+    guests = {}
+    for vmid in ctx.lxc_configs:
+        guests[vmid] = "lxc"
+    for vmid in ctx.qemu_configs:
+        guests[vmid] = "qemu"
+    if not guests:
+        return None
+
+    jobs = _parse_vzdump_jobs(ctx.vzdump_jobs)
+    if not jobs:
+        return {
+            "state": FAIL,
+            "summary_key": "noJobs",
+            "affected": [{"vmid": v, "type": t} for v, t in sorted(guests.items())],
+            "evidence": "no job definitions found in /etc/pve/jobs.cfg "
+                        "or /etc/vzdump.cron",
+        }
+
+    pools = _pool_members(ctx.pve_user_cfg)
+    covered: set[int] = set()
+    considered: list[str] = []
+    skipped: list[str] = []
+
+    for job in jobs:
+        if job.get("enabled", "1").strip() == "0":
+            skipped.append(f"{job['id']} (disabled)")
+            continue
+        excluded = {int(x) for x in re.findall(r"\d+", job.get("exclude", ""))}
+        selected: set[int] = set()
+        if job.get("all", "0").strip() == "1":
+            selected = set(guests)
+        else:
+            selected |= {int(x) for x in re.findall(r"\d+", job.get("vmid", ""))}
+            for pool in re.split(r"[,\s]+", job.get("pool", "").strip()):
+                if pool:
+                    selected |= pools.get(pool, set())
+        covered |= selected - excluded
+        considered.append(f"{job['id']} -> {sorted(selected - excluded) or 'nothing'}")
+
+    evidence = "enabled jobs:\n  " + ("\n  ".join(considered) or "(none)")
+    if skipped:
+        evidence += "\nignored jobs:\n  " + "\n  ".join(skipped)
+
+    uncovered = sorted(set(guests) - covered)
+    if not uncovered:
+        return {
+            "state": PASS,
+            "summary_key": "covered",
+            "summary_params": {"total": len(guests)},
+            "evidence": evidence,
+        }
+    return {
+        "state": FAIL,
+        "summary_key": "uncovered",
+        "summary_params": {"count": len(uncovered), "total": len(guests)},
+        "affected": [{"vmid": v, "type": guests[v]} for v in uncovered],
+        "evidence": evidence + f"\n\nuncovered: {uncovered}",
+    }
+
+
+# ---------------------------------------------------------------------------
+# System
+# ---------------------------------------------------------------------------
+
+@register("system.pending_reboot", AREA_SYSTEM, "WARNING")
+def _pending_reboot(ctx):
+    """Kernel or packages installed but not yet in effect."""
+    marker = Path("/var/run/reboot-required")
+    packages = ""
+    pkg_file = Path("/var/run/reboot-required.pkgs")
+    if pkg_file.exists():
+        try:
+            packages = pkg_file.read_text(errors="replace").strip()
+        except OSError:
+            packages = ""
+
+    rc, running = ctx.run(["uname", "-r"])
+    running = running.strip()
+
+    if not marker.exists():
+        return {
+            "state": PASS,
+            "summary_key": "none",
+            "evidence": f"running kernel: {running}",
+        }
+    return {
+        "state": WARN,
+        "summary_key": "pending",
+        "affected": [{"package": p} for p in packages.splitlines() if p],
+        "evidence": f"running kernel: {running}\n"
+                    f"packages requesting a restart:\n{packages or '(not reported)'}",
+    }
+
+
+@register("system.enterprise_repo_without_subscription", AREA_SYSTEM, "WARNING")
+def _enterprise_repo(ctx):
+    """Enterprise repository enabled on a host without a subscription.
+
+    The combination leaves ``apt update`` failing on every run, which
+    tends to be misread as a broken host rather than a licensing state.
+    """
+    enabled = []
+    for path, text in ctx.apt_sources.items():
+        for line in text.splitlines():
+            stripped = line.strip()
+            if stripped.startswith("#") or not stripped:
+                continue
+            if "enterprise.proxmox.com" in stripped:
+                enabled.append((path, stripped))
+    if not enabled:
+        return {
+            "state": PASS,
+            "summary_key": "notEnabled",
+        }
+
+    rc, out = ctx.run(["pvesubscription", "get"])
+    status = ""
+    for line in (out or "").splitlines():
+        if line.lower().startswith("status:"):
+            status = line.split(":", 1)[1].strip().lower()
+            break
+
+    evidence = "\n".join(f"{p}: {l}" for p, l in enabled)
+    evidence += f"\n\npvesubscription status: {status or '(unavailable)'}"
+
+    if status in ("active", "new"):
+        return {
+            "state": PASS,
+            "summary_key": "subscribed",
+            "evidence": evidence,
+        }
+    return {
+        "state": WARN,
+        "summary_key": "unsubscribed",
+        "affected": [{"file": p, "line": l} for p, l in enabled],
+        "evidence": evidence,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Guests
+# ---------------------------------------------------------------------------
+
+@register("guests.privileged_containers", AREA_GUESTS, "WARNING")
+def _privileged_containers(ctx):
+    """Containers running privileged.
+
+    A privileged container shares the host's user namespace, so a process
+    that escapes it is already root on the hypervisor. Proxmox creates
+    containers unprivileged by default; a container is privileged when
+    ``unprivileged: 1`` is absent from its configuration.
+    """
+    configs = ctx.lxc_configs
+    if not configs:
+        return None
+
+    privileged = []
+    for vmid, text in sorted(configs.items()):
+        if not re.search(r"^unprivileged:\s*1\s*$", text, re.M):
+            name = ""
+            m = re.search(r"^hostname:\s*(\S+)", text, re.M)
+            if m:
+                name = m.group(1)
+            privileged.append({"vmid": vmid, "name": name})
+
+    if not privileged:
+        return {
+            "state": PASS,
+            "summary_key": "allUnprivileged",
+            "summary_params": {"total": len(configs)},
+        }
+    listed = ", ".join(
+        f"{c['vmid']}{' (' + c['name'] + ')' if c['name'] else ''}"
+        for c in privileged
+    )
+    return {
+        "state": WARN,
+        "summary_key": "privileged",
+        "summary_params": {"count": len(privileged), "total": len(configs)},
+        "affected": privileged,
+        "evidence": f"privileged containers: {listed}",
+    }
+
+
+@register("guests.qemu_without_agent", AREA_GUESTS, "INFO")
+def _qemu_without_agent(ctx):
+    """Virtual machines with no guest agent declared.
+
+    Without it the host cannot request a clean shutdown, quiesce the
+    filesystem for a snapshot, or report real disk usage.
+    """
+    configs = ctx.qemu_configs
+    if not configs:
+        return None
+
+    missing = []
+    for vmid, text in sorted(configs.items()):
+        if not re.search(r"^agent:\s*(1|enabled=1)", text, re.M):
+            name = ""
+            m = re.search(r"^name:\s*(\S+)", text, re.M)
+            if m:
+                name = m.group(1)
+            missing.append({"vmid": vmid, "name": name})
+
+    if not missing:
+        return {
+            "state": PASS,
+            "summary_key": "allHaveAgent",
+            "summary_params": {"total": len(configs)},
+        }
+    listed = ", ".join(
+        f"{v['vmid']}{' (' + v['name'] + ')' if v['name'] else ''}"
+        for v in missing
+    )
+    return {
+        "state": WARN,
+        "summary_key": "missingAgent",
+        "summary_params": {"count": len(missing), "total": len(configs)},
+        "affected": missing,
+        "evidence": f"without agent: {listed}",
+    }
+
+
+# ---------------------------------------------------------------------------
+# Security
+# ---------------------------------------------------------------------------
+
+@register("security.host_firewall_enabled", AREA_SECURITY, "WARNING")
+def _host_firewall(ctx):
+    """Proxmox firewall enabled at datacenter and node level.
+
+    Both levels matter: the node rules are not applied while the
+    datacenter switch is off, so a node that looks configured can still
+    be filtering nothing.
+    """
+    def enabled_in(path: Path) -> tuple[bool, str]:
+        try:
+            text = path.read_text(errors="replace")
+        except OSError:
+            return False, f"{path}: not present"
+        for line in text.splitlines():
+            if re.match(r"^\s*enable:\s*1\s*$", line):
+                return True, f"{path}: enable: 1"
+        return False, f"{path}: enable not set to 1"
+
+    dc_on, dc_note = enabled_in(Path("/etc/pve/firewall/cluster.fw"))
+    try:
+        node = Path("/etc/hostname").read_text().strip()
+    except OSError:
+        node = ""
+    node_path = Path(f"/etc/pve/nodes/{node}/host.fw") if node else None
+    node_on, node_note = (False, "node firewall file not resolved")
+    if node_path:
+        node_on, node_note = enabled_in(node_path)
+
+    evidence = f"{dc_note}\n{node_note}"
+    if dc_on and node_on:
+        return {
+            "state": PASS,
+            "summary_key": "bothEnabled",
+            "evidence": evidence,
+        }
+    if not dc_on:
+        return {
+            "state": WARN,
+            "summary_key": "datacenterOff",
+            "evidence": evidence,
+        }
+    return {
+        "state": WARN,
+        "summary_key": "nodeOff",
+        "evidence": evidence,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Storage
+# ---------------------------------------------------------------------------
+
+@register("storage.orphaned_volumes", AREA_STORAGE, "WARNING")
+def _orphaned_volumes(ctx):
+    """Disk images that no guest configuration references.
+
+    A volume survives when a guest is removed without its disks, or when
+    a restore leaves the previous copy behind. Nothing reports it and it
+    keeps occupying the pool.
+
+    Only storage that is not shared is examined. On shared storage a
+    volume may belong to a guest running on another node, which this node
+    cannot see, so flagging it would be wrong rather than merely noisy.
+    """
+    known = set(ctx.lxc_configs) | set(ctx.qemu_configs)
+    if not known:
+        return None
+
+    candidates = [
+        s for s in ctx.storages
+        if str(s.get("shared", "0")).strip() != "1"
+        and any(c in (s.get("content") or "") for c in ("images", "rootdir"))
+    ]
+    if not candidates:
+        return None
+
+    orphans: list[dict] = []
+    inspected: list[str] = []
+    for storage in candidates:
+        sid = storage["id"]
+        rc, out = ctx.run(["pvesm", "list", sid], timeout=15)
+        if rc != 0:
+            inspected.append(f"{sid}: not readable")
+            continue
+        count = 0
+        for line in (out or "").splitlines()[1:]:
+            fields = line.split()
+            if len(fields) < 5:
+                continue
+            volid, vmid_raw = fields[0], fields[-1]
+            if not vmid_raw.isdigit():
+                continue
+            count += 1
+            vmid = int(vmid_raw)
+            if vmid not in known:
+                orphans.append({"volume": volid, "vmid": vmid})
+        inspected.append(f"{sid}: {count} volume(s)")
+
+    shared_skipped = [
+        s["id"] for s in ctx.storages
+        if str(s.get("shared", "0")).strip() == "1"
+    ]
+    evidence = "inspected:\n  " + "\n  ".join(inspected)
+    if shared_skipped:
+        evidence += ("\nskipped as shared (ownership not resolvable from this "
+                     "node):\n  " + ", ".join(shared_skipped))
+
+    if not orphans:
+        return {
+            "state": PASS,
+            "summary_key": "none",
+            "evidence": evidence,
+        }
+    return {
+        "state": WARN,
+        "summary_key": "found",
+        "summary_params": {"count": len(orphans)},
+        "affected": orphans,
+        "evidence": evidence + "\n\norphans:\n  " + "\n  ".join(
+            f"{o['volume']} (no config for {o['vmid']})" for o in orphans),
+    }

AppImage/scripts/audit_store.py

@@ -0,0 +1,426 @@
+"""Persistence layer for Audit & Report.
+
+Holds assessment runs, their findings, the accepted-risk register and the
+designated baseline.
+
+The store lives in its own database rather than alongside health and
+notification state. An assessment writes every finding of a run in one
+burst and its retention pass deletes whole runs; sharing a file with the
+notification dispatcher — which opens ``BEGIN IMMEDIATE`` transactions on
+every delivered event — would make those two paths contend for the same
+write lock.
+
+Findings persist i18n keys, never rendered text. A report exported today
+may be read in a different language than the one active when the
+assessment ran, and the printed document renders from the key at
+presentation time. Evidence is the exception: it is raw command output
+and is stored verbatim.
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+import threading
+import time
+import uuid
+from pathlib import Path
+from typing import Any, Optional
+
+DB_PATH = Path("/usr/local/share/proxmenux/audit.db")
+
+# Result of a check within one run. Severity is what the check declares
+# for a failure; state is what actually happened this time.
+STATE_FAIL = "fail"
+STATE_WARN = "warn"
+STATE_PASS = "pass"
+STATE_NOT_APPLICABLE = "not_applicable"
+STATE_ACCEPTED = "accepted"
+
+RUN_RUNNING = "running"
+RUN_COMPLETE = "complete"
+RUN_FAILED = "failed"
+
+_schema_lock = threading.Lock()
+_schema_ready = False
+
+
+def _connect() -> sqlite3.Connection:
+    conn = sqlite3.connect(str(DB_PATH), timeout=10)
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA busy_timeout=5000")
+    conn.execute("PRAGMA foreign_keys=ON")
+    return conn
+
+
+def init_db() -> None:
+    """Create the schema. Safe to call repeatedly."""
+    global _schema_ready
+    with _schema_lock:
+        if _schema_ready:
+            return
+        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+        conn = _connect()
+        try:
+            conn.executescript("""
+                CREATE TABLE IF NOT EXISTS audit_runs (
+                    run_id        TEXT PRIMARY KEY,
+                    profile       TEXT NOT NULL,
+                    started_at    INTEGER NOT NULL,
+                    finished_at   INTEGER,
+                    status        TEXT NOT NULL,
+                    error         TEXT,
+                    is_baseline   INTEGER NOT NULL DEFAULT 0,
+                    checks_total  INTEGER NOT NULL DEFAULT 0,
+                    schema_version INTEGER NOT NULL DEFAULT 1
+                );
+
+                -- summary_key names a translation entry and summary_params
+                -- carries its placeholders. Storing a rendered sentence
+                -- instead would freeze a finding in whichever language was
+                -- active when the assessment ran, and a report exported
+                -- today may well be read in another one.
+                CREATE TABLE IF NOT EXISTS audit_findings (
+                    id             INTEGER PRIMARY KEY AUTOINCREMENT,
+                    run_id         TEXT NOT NULL,
+                    check_id       TEXT NOT NULL,
+                    area           TEXT NOT NULL,
+                    severity       TEXT NOT NULL,
+                    state          TEXT NOT NULL,
+                    summary_key    TEXT,
+                    summary_params TEXT,
+                    affected       TEXT,
+                    evidence       TEXT,
+                    remediable_by  TEXT,
+                    FOREIGN KEY (run_id) REFERENCES audit_runs(run_id)
+                        ON DELETE CASCADE
+                );
+
+                -- Accepted risks outlive the run that surfaced them, so they
+                -- are keyed by check rather than by finding. expires_at NULL
+                -- means the acceptance does not lapse on its own.
+                CREATE TABLE IF NOT EXISTS audit_exceptions (
+                    check_id      TEXT PRIMARY KEY,
+                    reason        TEXT NOT NULL,
+                    accepted_by   TEXT NOT NULL,
+                    accepted_at   INTEGER NOT NULL,
+                    expires_at    INTEGER
+                );
+
+                CREATE INDEX IF NOT EXISTS idx_audit_findings_run
+                    ON audit_findings(run_id);
+                CREATE INDEX IF NOT EXISTS idx_audit_findings_check
+                    ON audit_findings(check_id);
+                CREATE INDEX IF NOT EXISTS idx_audit_runs_started
+                    ON audit_runs(started_at);
+            """)
+            conn.commit()
+            _schema_ready = True
+        finally:
+            conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Runs
+# ---------------------------------------------------------------------------
+
+def start_run(profile: str) -> str:
+    """Open a run and return its identifier."""
+    init_db()
+    run_id = uuid.uuid4().hex[:16]
+    conn = _connect()
+    try:
+        conn.execute(
+            "INSERT INTO audit_runs (run_id, profile, started_at, status) "
+            "VALUES (?, ?, ?, ?)",
+            (run_id, profile, int(time.time()), RUN_RUNNING),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+    return run_id
+
+
+def finish_run(run_id: str, *, checks_total: int,
+               error: Optional[str] = None) -> None:
+    """Close a run, marking it failed when an error is supplied."""
+    init_db()
+    conn = _connect()
+    try:
+        conn.execute(
+            "UPDATE audit_runs SET finished_at = ?, status = ?, error = ?, "
+            "checks_total = ? WHERE run_id = ?",
+            (int(time.time()), RUN_FAILED if error else RUN_COMPLETE,
+             error, checks_total, run_id),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def get_run(run_id: str) -> Optional[dict[str, Any]]:
+    init_db()
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        row = conn.execute(
+            "SELECT * FROM audit_runs WHERE run_id = ?", (run_id,)
+        ).fetchone()
+        return dict(row) if row else None
+    finally:
+        conn.close()
+
+
+def list_runs(limit: int = 20) -> list[dict[str, Any]]:
+    init_db()
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            "SELECT * FROM audit_runs ORDER BY started_at DESC LIMIT ?",
+            (limit,),
+        ).fetchall()
+        return [dict(r) for r in rows]
+    finally:
+        conn.close()
+
+
+def latest_run(status: str = RUN_COMPLETE) -> Optional[dict[str, Any]]:
+    init_db()
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        row = conn.execute(
+            "SELECT * FROM audit_runs WHERE status = ? "
+            "ORDER BY started_at DESC LIMIT 1",
+            (status,),
+        ).fetchone()
+        return dict(row) if row else None
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Findings
+# ---------------------------------------------------------------------------
+
+def record_findings(run_id: str, findings: list[dict[str, Any]]) -> int:
+    """Write a run's findings in a single transaction.
+
+    ``affected`` is stored as JSON so a check that covers several objects
+    keeps the per-object detail without emitting one finding per object.
+    """
+    init_db()
+    if not findings:
+        return 0
+    rows = [
+        (
+            run_id,
+            f["check_id"],
+            f["area"],
+            f["severity"],
+            f["state"],
+            f.get("summary_key"),
+            json.dumps(f.get("summary_params") or {}, ensure_ascii=False),
+            json.dumps(f.get("affected") or [], ensure_ascii=False),
+            f.get("evidence"),
+            f.get("remediable_by"),
+        )
+        for f in findings
+    ]
+    conn = _connect()
+    try:
+        conn.execute("BEGIN IMMEDIATE")
+        conn.executemany(
+            "INSERT INTO audit_findings (run_id, check_id, area, severity, "
+            "state, summary_key, summary_params, affected, evidence, "
+            "remediable_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            rows,
+        )
+        conn.commit()
+    finally:
+        conn.close()
+    return len(rows)
+
+
+def get_findings(run_id: str) -> list[dict[str, Any]]:
+    init_db()
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            "SELECT * FROM audit_findings WHERE run_id = ? ORDER BY id",
+            (run_id,),
+        ).fetchall()
+        out = []
+        for r in rows:
+            item = dict(r)
+            try:
+                item["affected"] = json.loads(item.get("affected") or "[]")
+            except (TypeError, ValueError):
+                item["affected"] = []
+            try:
+                item["summary_params"] = json.loads(
+                    item.get("summary_params") or "{}")
+            except (TypeError, ValueError):
+                item["summary_params"] = {}
+            out.append(item)
+        return out
+    finally:
+        conn.close()
+
+
+def check_history(check_id: str, limit: int = 30) -> list[dict[str, Any]]:
+    """Return how one check resolved across recent runs."""
+    init_db()
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            "SELECT f.state, f.severity, r.run_id, r.started_at "
+            "FROM audit_findings f JOIN audit_runs r ON r.run_id = f.run_id "
+            "WHERE f.check_id = ? ORDER BY r.started_at DESC LIMIT ?",
+            (check_id, limit),
+        ).fetchall()
+        return [dict(r) for r in rows]
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Accepted risks
+# ---------------------------------------------------------------------------
+
+def accept_risk(check_id: str, reason: str, accepted_by: str,
+                expires_at: Optional[int] = None) -> None:
+    """Record a deliberate decision to leave a finding unresolved.
+
+    A reason is mandatory: an acceptance without one is indistinguishable
+    from having silenced the check, which is what this register exists to
+    prevent.
+    """
+    if not (reason or "").strip():
+        raise ValueError("an accepted risk requires a reason")
+    init_db()
+    conn = _connect()
+    try:
+        conn.execute(
+            "INSERT OR REPLACE INTO audit_exceptions "
+            "(check_id, reason, accepted_by, accepted_at, expires_at) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (check_id, reason.strip(), accepted_by, int(time.time()),
+             expires_at),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def revoke_risk(check_id: str) -> bool:
+    init_db()
+    conn = _connect()
+    try:
+        cur = conn.execute(
+            "DELETE FROM audit_exceptions WHERE check_id = ?", (check_id,)
+        )
+        conn.commit()
+        return cur.rowcount > 0
+    finally:
+        conn.close()
+
+
+def active_exceptions() -> dict[str, dict[str, Any]]:
+    """Return accepted risks that have not lapsed, keyed by check.
+
+    Lapsed entries are left on disk so the decision remains auditable;
+    they simply stop suppressing the finding.
+    """
+    init_db()
+    now = int(time.time())
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            "SELECT * FROM audit_exceptions "
+            "WHERE expires_at IS NULL OR expires_at > ?",
+            (now,),
+        ).fetchall()
+        return {r["check_id"]: dict(r) for r in rows}
+    finally:
+        conn.close()
+
+
+def all_exceptions() -> list[dict[str, Any]]:
+    init_db()
+    now = int(time.time())
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            "SELECT * FROM audit_exceptions ORDER BY accepted_at DESC"
+        ).fetchall()
+        out = []
+        for r in rows:
+            item = dict(r)
+            item["lapsed"] = bool(
+                item["expires_at"] is not None and item["expires_at"] <= now
+            )
+            out.append(item)
+        return out
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Baseline and retention
+# ---------------------------------------------------------------------------
+
+def set_baseline(run_id: str) -> None:
+    """Designate a run as the reference to compare later runs against."""
+    init_db()
+    conn = _connect()
+    try:
+        conn.execute("BEGIN IMMEDIATE")
+        conn.execute("UPDATE audit_runs SET is_baseline = 0")
+        conn.execute(
+            "UPDATE audit_runs SET is_baseline = 1 WHERE run_id = ?", (run_id,)
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def get_baseline() -> Optional[dict[str, Any]]:
+    init_db()
+    conn = _connect()
+    try:
+        conn.row_factory = sqlite3.Row
+        row = conn.execute(
+            "SELECT * FROM audit_runs WHERE is_baseline = 1 LIMIT 1"
+        ).fetchone()
+        return dict(row) if row else None
+    finally:
+        conn.close()
+
+
+def prune_runs(keep: int = 30) -> int:
+    """Drop the oldest runs beyond ``keep``.
+
+    The baseline is never pruned: it is the reference every comparison is
+    measured against and losing it silently would break that comparison
+    long after the run that produced it was forgotten.
+    """
+    init_db()
+    conn = _connect()
+    try:
+        conn.execute("BEGIN IMMEDIATE")
+        cur = conn.execute(
+            "DELETE FROM audit_runs WHERE is_baseline = 0 AND run_id NOT IN ("
+            "  SELECT run_id FROM audit_runs "
+            "  WHERE is_baseline = 0 ORDER BY started_at DESC LIMIT ?"
+            ")",
+            (keep,),
+        )
+        conn.commit()
+        return cur.rowcount
+    finally:
+        conn.close()

AppImage/scripts/build_appimage.sh

@@ -165,6 +165,10 @@ cp "$SCRIPT_DIR/startup_grace.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠
 cp "$SCRIPT_DIR/flask_notification_routes.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  flask_notification_routes.py not found"
 cp "$SCRIPT_DIR/oci_manager.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  oci_manager.py not found"
 cp "$SCRIPT_DIR/flask_oci_routes.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  flask_oci_routes.py not found"
+cp "$SCRIPT_DIR/flask_audit_routes.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  flask_audit_routes.py not found"
+cp "$SCRIPT_DIR/audit_store.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  audit_store.py not found"
+cp "$SCRIPT_DIR/audit_checks.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  audit_checks.py not found"
+cp "$SCRIPT_DIR/audit_checks_pve.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  audit_checks_pve.py not found"
 cp "$SCRIPT_DIR/oci/description_templates.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  description_templates.py not found"
 
 # Copy AI providers module for notification enhancement

AppImage/scripts/flask_audit_routes.py

@@ -0,0 +1,281 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+ProxMenux Audit Routes
+Flask blueprint for the Audit & Report assessment engine.
+
+An assessment reads the host and records findings; it never modifies
+anything. The run endpoint is therefore the only POST that does real
+work, and it is deliberately serialised: two concurrent assessments would
+compete for the same collectors without producing a better answer.
+"""
+
+import threading
+import time
+
+from flask import Blueprint, jsonify, request
+from jwt_middleware import require_auth
+
+audit_bp = Blueprint('audit', __name__)
+
+try:
+    import audit_store
+    import audit_checks
+    import audit_checks_pve  # noqa: F401 — importing registers the checks
+except ImportError:
+    audit_store = None
+    audit_checks = None
+
+# One assessment at a time. The flag is also what the interface polls to
+# know a run is still in progress.
+_run_lock = threading.Lock()
+_running: dict = {'active': False, 'run_id': None, 'started_at': 0}
+
+
+def _unavailable():
+    return jsonify({
+        "success": False,
+        "message": "Audit engine not available",
+    }), 500
+
+
+@audit_bp.route('/api/audit/checks', methods=['GET'])
+@require_auth
+def list_checks():
+    """Catalogue of registered checks, independent of any run."""
+    if not audit_checks:
+        return _unavailable()
+    try:
+        return jsonify({
+            "success": True,
+            "areas": list(audit_checks.AREAS),
+            "checks": [
+                {
+                    "check_id": c.check_id,
+                    "area": c.area,
+                    "severity": c.severity,
+                }
+                for c in audit_checks.registered_checks()
+            ],
+        })
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/status', methods=['GET'])
+@require_auth
+def status():
+    """Latest run, whether an assessment is in progress, and the baseline."""
+    if not audit_store:
+        return _unavailable()
+    try:
+        latest = audit_store.latest_run()
+        summary = {}
+        if latest:
+            for f in audit_store.get_findings(latest['run_id']):
+                summary[f['state']] = summary.get(f['state'], 0) + 1
+        return jsonify({
+            "success": True,
+            "running": _running['active'],
+            "latest": latest,
+            "summary": summary,
+            "baseline": audit_store.get_baseline(),
+            "exceptions": len(audit_store.active_exceptions()),
+        })
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/run', methods=['POST'])
+@require_auth
+def run():
+    """Start an assessment in the background.
+
+    The response returns immediately with the run identifier; the
+    interface polls ``/api/audit/status``. A full assessment is short but
+    runs against a production host, so it must not hold an HTTP worker.
+    """
+    if not audit_checks:
+        return _unavailable()
+
+    data = request.get_json(silent=True) or {}
+    profile = str(data.get('profile') or 'full')
+    areas = data.get('areas')
+    only = set(areas) if isinstance(areas, list) and areas else None
+
+    with _run_lock:
+        if _running['active']:
+            return jsonify({
+                "success": False,
+                "message": "An assessment is already running",
+                "run_id": _running['run_id'],
+            }), 409
+        _running.update({'active': True, 'run_id': None,
+                         'started_at': time.time()})
+
+    def worker():
+        try:
+            run_id = audit_checks.run_assessment(profile, only_areas=only)
+            _running['run_id'] = run_id
+            audit_store.prune_runs()
+        except Exception as e:
+            print(f"[audit] assessment failed: {e}")
+        finally:
+            _running['active'] = False
+
+    threading.Thread(target=worker, daemon=True, name='audit-run').start()
+    return jsonify({"success": True, "started": True})
+
+
+@audit_bp.route('/api/audit/runs', methods=['GET'])
+@require_auth
+def runs():
+    if not audit_store:
+        return _unavailable()
+    try:
+        limit = min(int(request.args.get('limit', 20)), 100)
+        return jsonify({"success": True, "runs": audit_store.list_runs(limit)})
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/runs/<run_id>', methods=['GET'])
+@require_auth
+def run_detail(run_id):
+    """Findings of one run, with the accepted-risk record attached.
+
+    Accepted findings are returned like any other so the interface can
+    show them muted rather than dropping them: hiding an accepted risk
+    turns the register into a way of forgetting decisions.
+    """
+    if not audit_store:
+        return _unavailable()
+    try:
+        run = audit_store.get_run(run_id)
+        if not run:
+            return jsonify({"success": False, "message": "Run not found"}), 404
+        exceptions = audit_store.active_exceptions()
+        findings = audit_store.get_findings(run_id)
+        for f in findings:
+            f['exception'] = exceptions.get(f['check_id'])
+        return jsonify({"success": True, "run": run, "findings": findings})
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/compare', methods=['GET'])
+@require_auth
+def compare():
+    """Difference between two runs, defaulting the base to the baseline."""
+    if not audit_store:
+        return _unavailable()
+    try:
+        other = request.args.get('to')
+        base = request.args.get('from')
+        if not base:
+            baseline = audit_store.get_baseline()
+            base = baseline['run_id'] if baseline else None
+        if not other:
+            latest = audit_store.latest_run()
+            other = latest['run_id'] if latest else None
+        if not base or not other:
+            return jsonify({
+                "success": False,
+                "message": "Two runs are required to compare",
+            }), 400
+        return jsonify({
+            "success": True,
+            "from": base,
+            "to": other,
+            **audit_checks.compare_runs(base, other),
+        })
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/baseline', methods=['POST'])
+@require_auth
+def set_baseline():
+    if not audit_store:
+        return _unavailable()
+    try:
+        data = request.get_json(silent=True) or {}
+        run_id = data.get('run_id')
+        if not run_id or not audit_store.get_run(run_id):
+            return jsonify({"success": False, "message": "Run not found"}), 404
+        audit_store.set_baseline(run_id)
+        return jsonify({"success": True})
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/exceptions', methods=['GET'])
+@require_auth
+def list_exceptions():
+    if not audit_store:
+        return _unavailable()
+    try:
+        return jsonify({
+            "success": True,
+            "exceptions": audit_store.all_exceptions(),
+        })
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/exceptions', methods=['POST'])
+@require_auth
+def accept_exception():
+    """Record a finding as a deliberate decision.
+
+    The reason is mandatory. An acceptance without one cannot be
+    distinguished later from having silenced the check, which is the
+    outcome this register exists to prevent.
+    """
+    if not audit_store:
+        return _unavailable()
+    try:
+        data = request.get_json(silent=True) or {}
+        check_id = (data.get('check_id') or '').strip()
+        reason = (data.get('reason') or '').strip()
+        if not check_id:
+            return jsonify({"success": False,
+                            "message": "check_id is required"}), 400
+        if not reason:
+            return jsonify({"success": False,
+                            "message": "A reason is required"}), 400
+
+        expires_at = None
+        days = data.get('expires_in_days')
+        if days:
+            try:
+                expires_at = int(time.time()) + int(days) * 86400
+            except (TypeError, ValueError):
+                return jsonify({"success": False,
+                                "message": "Invalid expiry"}), 400
+
+        audit_store.accept_risk(
+            check_id, reason,
+            accepted_by=str(data.get('accepted_by') or 'admin'),
+            expires_at=expires_at,
+        )
+        return jsonify({"success": True})
+    except ValueError as e:
+        return jsonify({"success": False, "message": str(e)}), 400
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500
+
+
+@audit_bp.route('/api/audit/exceptions/<path:check_id>', methods=['DELETE'])
+@require_auth
+def revoke_exception(check_id):
+    if not audit_store:
+        return _unavailable()
+    try:
+        removed = audit_store.revoke_risk(check_id)
+        if not removed:
+            return jsonify({"success": False,
+                            "message": "Exception not found"}), 404
+        return jsonify({"success": True})
+    except Exception as e:
+        return jsonify({"success": False, "message": str(e)}), 500

AppImage/scripts/flask_server.py

@@ -92,6 +92,7 @@ from flask_proxmenux_routes import proxmenux_bp  # noqa: E402
 from flask_security_routes import security_bp  # noqa: E402
 from flask_notification_routes import notification_bp  # noqa: E402
 from flask_oci_routes import oci_bp  # noqa: E402
+from flask_audit_routes import audit_bp  # noqa: E402
 from notification_manager import notification_manager  # noqa: E402
 import post_install_versions  # noqa: E402  — Sprint 12A: detect post-install function updates
 from jwt_middleware import require_auth, require_auth_or_ticket, require_admin_scope  # noqa: E402
@@ -227,6 +228,7 @@ app.register_blueprint(proxmenux_bp)
 app.register_blueprint(security_bp)
 app.register_blueprint(notification_bp)
 app.register_blueprint(oci_bp)
+app.register_blueprint(audit_bp)
 
 # Initialize terminal / WebSocket routes
 init_terminal_routes(app)
@@ -1697,7 +1699,7 @@ _VM_DISK_REFRESH_WORKERS = 6  # parallelism cap for the fsinfo pass
 # state must be probed rather than inferred from an API response.
 _vm_details_cache: dict = {}      # vmid -> (ts, payload)
 _vm_backups_cache: dict = {}      # vmid -> (ts, payload)
-_vm_apps_cache: dict = {}         # vmid -> (ts, payload)
+# Registered apps use lxc_apps' shared, file-versioned snapshot cache.
 _vm_app_suggestions_cache: dict = {}  # vmid -> (ts, payload)
 _vm_schedule_cache: dict = {}     # vmid -> (ts, payload)
 _vm_mounts_cache: dict = {}       # vmid -> (ts, payload) — LXC only
@@ -1735,7 +1737,6 @@ _lxc_ip_cache: dict = {}
 _VM_CACHE_INDEFINITE = 315_360_000   # 10 years — effectively infinite
 _VM_DETAILS_TTL   = _VM_CACHE_INDEFINITE
 _VM_BACKUPS_TTL   = _VM_CACHE_INDEFINITE
-_VM_APPS_TTL      = _VM_CACHE_INDEFINITE
 _VM_APP_SUGGESTIONS_TTL = _VM_CACHE_INDEFINITE
 _VM_SCHEDULE_TTL  = _VM_CACHE_INDEFINITE
 _VM_MOUNTS_TTL    = _VM_CACHE_INDEFINITE
@@ -1817,7 +1818,7 @@ def _vm_cache_invalidate(vmid: int, *caches) -> None:
     affect any of them (e.g. control start/stop flips status, which
     lives in the details payload)."""
     targets = caches or (
-        _vm_details_cache, _vm_backups_cache, _vm_apps_cache,
+        _vm_details_cache, _vm_backups_cache,
         _vm_app_suggestions_cache, _vm_schedule_cache, _vm_mounts_cache,
     )
     with _vm_modal_cache_lock:
@@ -1906,7 +1907,6 @@ def _refresh_started_guest(vmid: int, vm_type: str) -> None:
 
             sidecar = lxc_apps.check_all(vmid, force=True)
             sidecar = sidecar or {'vmid': vmid, 'apps': []}
-            _vm_cache_put(_vm_apps_cache, vmid, sidecar)
 
             docker_registered = any(
                 isinstance(item, dict) and item.get('helper_slug') == 'docker'
@@ -11891,51 +11891,69 @@ def api_vm_metrics(vmid):
 
         return jsonify({'error': str(e)}), 500
 
-# Per-process cache for the RRD payload of /api/node/metrics. Two unrelated
-# dashboard components (`network-traffic-chart` for the network panel and
-# `node-metrics-charts` for the CPU/memory panel) mount in parallel on the
-# Overview page and each fires this endpoint independently with the same
-# `?timeframe=` argument. The underlying `pvesh get rrddata` call takes
-# ~1 second; without a cache, the second fetch blocks behind the first
-# (especially under gevent), occasionally surfacing as a transient 502
-# while gevent is single-threaded for blocking calls. RRD data is updated
-# on a per-minute cadence by PVE, so a 10-second cache is safe and the
-# UI experience is materially better.
+# Shared RRD snapshots and single-flight locks, one per supported timeframe.
+_NODE_METRICS_TTL = 120.0
+_NODE_METRICS_TIMEOUT = 30.0
+_NODE_METRICS_RETRY_DELAY = 30.0
 _NODE_METRICS_CACHE = {}
-# TTL is 120 s because the prewarmer only refreshes the `hour`
-# timeframe (the Overview's default) every 90 s. The other four
-# timeframes get their first fetch lazily when the user picks them —
-# they then live in cache for 120 s, which covers back-and-forth
-# switching without paying pvesh cost. Pre-warming every timeframe
-# on a fast cadence (as the first version did) burned ~30 % of a
-# core continuously scanning data nobody was looking at.
-_NODE_METRICS_TTL = 120.0  # seconds
+_NODE_METRICS_FAILURES = {}
+_NODE_METRICS_LOCKS = {
+    timeframe: threading.Lock()
+    for timeframe in ('hour', 'day', 'week', 'month', 'year')
+}
 
 
-def _node_metrics_cache_get(timeframe):
-    entry = _NODE_METRICS_CACHE.get(timeframe)
-    if not entry:
-        return None
-    if time.monotonic() - entry['ts'] > _NODE_METRICS_TTL:
-        return None
-    return entry['payload']
+class _NodeMetricsError(Exception):
+    def __init__(self, payload):
+        super().__init__(payload['error'])
+        self.payload = payload
 
 
-def _node_metrics_cache_set(timeframe, payload):
-    _NODE_METRICS_CACHE[timeframe] = {'payload': payload, 'ts': time.monotonic()}
+def _node_metrics_fallback(timeframe, error):
+    cached = _NODE_METRICS_CACHE.get(timeframe)
+    if cached is not None:
+        return {**cached['payload'], 'cache_status': 'stale', 'refresh_error': error}
+    raise _NodeMetricsError(error)
 
 
-def _compute_node_metrics_payload(timeframe: str) -> dict | None:
-    """Do the actual pvesh-backed RRD fetch + massaging that
-    `api_node_metrics` used to do inline. Returns the payload dict on
-    success (and populates the cache), None on any failure.
-    Extracted so the background prewarmer can call it without going
-    through HTTP + `@require_auth` — same code path as the handler,
-    zero duplication of the massaging logic."""
+def _get_node_metrics_payload(timeframe, max_age=_NODE_METRICS_TTL):
+    lock = _NODE_METRICS_LOCKS[timeframe]
+    if not lock.acquire(timeout=_NODE_METRICS_TIMEOUT + 5):
+        return _node_metrics_fallback(timeframe, {
+            'error': 'A metrics query is still in progress', 'code': 'metrics_busy',
+        })
+    try:
+        now = time.monotonic()
+        cached = _NODE_METRICS_CACHE.get(timeframe)
+        failure = _NODE_METRICS_FAILURES.get(timeframe)
+        if failure is not None and now - failure['ts'] < _NODE_METRICS_RETRY_DELAY:
+            return _node_metrics_fallback(timeframe, failure['error'])
+        if cached is not None and now - cached['ts'] < max_age:
+            return cached['payload']
+        try:
+            payload = _compute_node_metrics_payload(timeframe)
+        except Exception as exc:
+            error = exc.payload if isinstance(exc, _NodeMetricsError) else {
+                'error': 'Unable to load Proxmox metrics', 'code': 'metrics_unavailable',
+            }
+            _NODE_METRICS_FAILURES[timeframe] = {'error': error, 'ts': time.monotonic()}
+            print(f"[ProxMenux] node metrics ({timeframe}): {exc}", file=sys.stderr, flush=True)
+            return _node_metrics_fallback(timeframe, error)
+        payload = {**payload, 'last_checked': int(time.time()), 'cache_status': 'fresh'}
+        _NODE_METRICS_CACHE[timeframe] = {'payload': payload, 'ts': time.monotonic()}
+        _NODE_METRICS_FAILURES.pop(timeframe, None)
+        return payload
+    finally:
+        lock.release()
+
+
+def _compute_node_metrics_payload(timeframe: str) -> dict:
+    """Fetch and shape one node RRD snapshot within a bounded query budget."""
     valid_timeframes = ('hour', 'day', 'week', 'month', 'year')
     if timeframe not in valid_timeframes:
-        return None
+        raise ValueError('Invalid timeframe')
 
+    deadline = time.monotonic() + _NODE_METRICS_TIMEOUT
     local_node = get_proxmox_node_name()
 
     zfs_arc_size = 0
@@ -11954,16 +11972,70 @@ def _compute_node_metrics_payload(timeframe: str) -> dict | None:
         rrd_result = subprocess.run(
             ['pvesh', 'get', f'/nodes/{local_node}/rrddata',
              '--timeframe', timeframe, '--output-format', 'json'],
-            capture_output=True, text=True, timeout=10,
+            capture_output=True, text=True, timeout=_NODE_METRICS_TIMEOUT,
         )
-    except (subprocess.SubprocessError, OSError):
-        return None
+    except subprocess.TimeoutExpired:
+        raise _NodeMetricsError({
+            'error': 'Proxmox metrics query timed out',
+            'code': 'metrics_timeout',
+            'details': 'The metrics query exceeded the 30-second limit. It can be retried without restarting any services.',
+        })
+    except (subprocess.SubprocessError, OSError) as exc:
+        raise _NodeMetricsError({'error': 'Proxmox metrics command failed', 'raw': str(exc)[:500]})
     if rrd_result.returncode != 0:
-        return None
+        stderr_str = (rrd_result.stderr or '') + (rrd_result.stdout or '')
+        stderr_lower = stderr_str.lower()
+        if 'mmaping file' in stderr_lower and 'invalid argument' in stderr_lower:
+            raise _NodeMetricsError({
+                'error': 'Proxmox RRD database is corrupt',
+                'details': (
+                    'The host metrics file Proxmox keeps under '
+                    '/var/lib/rrdcached/db/pve-node-9.0/ failed to '
+                    'memory-map (Invalid argument). This is a Proxmox-side '
+                    'data-store issue, not a Monitor bug.'
+                ),
+                'suggestion': (
+                    'Stop pvestatd + pve-cluster + rrdcached, move the '
+                    'broken RRD aside, restart the services. Proxmox will '
+                    'rebuild the RRD from scratch (history is lost).'
+                ),
+                'raw': stderr_str.strip()[:500],
+            })
+        if 'no such file' in stderr_lower or 'no such node' in stderr_lower or 'does not exist' in stderr_lower:
+            raise _NodeMetricsError({
+                'error': 'Proxmox node name mismatch',
+                'details': (
+                    f"pvesh could not find node '{local_node}'. The "
+                    'usual cause is that the host was renamed after '
+                    'Proxmox was installed, so /etc/pve/nodes/ still '
+                    'carries the old name. This is a Proxmox-side '
+                    'config issue, not a Monitor bug.'
+                ),
+                'suggestion': 'Compare `hostname` with `ls /etc/pve/nodes/` — they must match.',
+                'raw': stderr_str.strip()[:500],
+            })
+        if 'rrd' in stderr_lower or 'empty' in stderr_lower:
+            raise _NodeMetricsError({
+                'error': 'Proxmox RRD data not available',
+                'details': 'The RRD database appears empty. Proxmox may not have collected metrics yet (fresh install) or rrdcached was down at boot.',
+                'suggestion': 'systemctl restart rrdcached pvestatd ; wait ~5 min and reload this page.',
+                'raw': stderr_str.strip()[:500],
+            })
+        raise _NodeMetricsError({
+            'error': 'Proxmox metrics command failed',
+            'details': 'pvesh exited non-zero. Check Proxmox host status.',
+            'raw': stderr_str.strip()[:500],
+        })
+
     try:
         rrd_data = json.loads(rrd_result.stdout)
+        if not isinstance(rrd_data, list) or any(not isinstance(item, dict) for item in rrd_data):
+            raise ValueError('Expected an array of RRD points')
     except (json.JSONDecodeError, ValueError):
-        return None
+        raise _NodeMetricsError({
+            'error': 'Proxmox returned invalid metrics data',
+            'code': 'metrics_invalid_data',
+        })
 
     for item in rrd_data:
         if 'arcsize' in item:
@@ -11988,34 +12060,35 @@ def _compute_node_metrics_payload(timeframe: str) -> dict | None:
         }
 
     def _pvesh_rrd(cf):
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return None
         try:
             extra = subprocess.run(
                 ['pvesh', 'get', f'/nodes/{local_node}/rrddata',
                  '--timeframe', timeframe, '--cf', cf,
                  '--output-format', 'json'],
-                capture_output=True, text=True, timeout=10,
+                capture_output=True, text=True, timeout=remaining,
             )
             if extra.returncode == 0 and extra.stdout:
-                return json.loads(extra.stdout)
+                points = json.loads(extra.stdout)
+                if isinstance(points, list) and all(isinstance(item, dict) for item in points):
+                    return points
         except (subprocess.SubprocessError, json.JSONDecodeError, OSError):
             pass
         return None
 
+    # PVE supports AVERAGE and MAX, not MIN. Share MAX across both charts.
+    cf_max = _pvesh_rrd('MAX') if timeframe in ('week', 'month') else None
+
     def _build_stats(field_key, scale=1.0):
         native = _stats_native(field_key, scale)
         if native is None:
             return None
-        if timeframe in ('week', 'month'):
-            cf_max = _pvesh_rrd('MAX')
-            if cf_max:
-                vals = _values_from(cf_max, field_key, scale)
-                if vals:
-                    native['max'] = max(vals)
-            cf_min = _pvesh_rrd('MIN')
-            if cf_min:
-                vals = _values_from(cf_min, field_key, scale)
-                if vals:
-                    native['min'] = min(vals)
+        if cf_max:
+            vals = _values_from(cf_max, field_key, scale)
+            if vals:
+                native['max'] = max(vals)
         return native
 
     period_stats = {
@@ -12053,28 +12126,17 @@ def _compute_node_metrics_payload(timeframe: str) -> dict | None:
         'data': rrd_data,
         'period_stats': period_stats,
     }
-    _node_metrics_cache_set(timeframe, payload)
     return payload
 
 
 def _node_metrics_prewarmer_loop():
-    """Keep `_NODE_METRICS_CACHE['hour']` hot so the Overview page's
-    default view (CPU + Memory charts, 1-hour range) never waits on
-    `pvesh get rrddata`. Only `hour` is prewarmed — the other
-    timeframes (day/week/month/year) are lazy-cached on first click
-    and stick around for the 120 s TTL. Prewarming every timeframe
-    burned ~30 % of a core continuously against pvesh for data
-    nobody was looking at, and week/month each cost 3 pvesh calls
-    (base + MAX + MIN)."""
-    time.sleep(3)  # let the app finish importing before the first pass
+    """Prewarm the Overview's default range using the same single-flight cache."""
+    time.sleep(3)
     while True:
         try:
-            _compute_node_metrics_payload('hour')
-        except Exception as e:
-            print(f"[ProxMenux] node-metrics prewarmer error: {e}",
-                  file=sys.stderr, flush=True)
-        # Refresh well before the 120 s TTL expires so the user never
-        # hits a cold cache during a natural page open.
+            _get_node_metrics_payload('day', max_age=90.0)
+        except _NodeMetricsError:
+            pass  # The shared fetch path already records the failure.
         time.sleep(90)
 
 
@@ -12105,9 +12167,9 @@ def _vm_modal_prewarmer_pass():
              f'/api/vms/{vmid}/backups', 'backups'),
         ]
         if vm_type == 'lxc':
+            import lxc_apps
+            lxc_apps.load_sidecar(vmid)
             endpoints.extend([
-                (_vm_apps_cache, _VM_APPS_TTL, api_vm_apps_get,
-                 f'/api/vms/{vmid}/apps', 'apps'),
                 (_vm_schedule_cache, _VM_SCHEDULE_TTL, api_vm_apps_schedule,
                  f'/api/vms/{vmid}/schedule', 'schedule'),
                 (_vm_mounts_cache, _VM_MOUNTS_TTL, api_lxc_mount_points,
@@ -12176,249 +12238,17 @@ def _vm_modal_prewarmer_loop():
 @app.route('/api/node/metrics', methods=['GET'])
 @require_auth
 def api_node_metrics():
-    """Get historical metrics (RRD data) for the node.
-
-    Per-timeframe cached for ~10 s so the two dashboard panels that mount
-    together don't hit `pvesh` twice; see `_NODE_METRICS_CACHE` comment.
-    """
+    """Share one cached RRD snapshot across the dashboard's charts."""
+    timeframe = request.args.get('timeframe', 'week')
+    if timeframe not in _NODE_METRICS_LOCKS:
+        return jsonify({'error': 'Invalid timeframe. Must be one of: hour, day, week, month, year'}), 400
     try:
-        timeframe = request.args.get('timeframe', 'week')  # hour, day, week, month, year
-
-        # Validate timeframe
-        valid_timeframes = ['hour', 'day', 'week', 'month', 'year']
-        if timeframe not in valid_timeframes:
-            return jsonify({'error': f'Invalid timeframe. Must be one of: {", ".join(valid_timeframes)}'}), 400
-
-        # Serve from cache when fresh — completely skips the pvesh call.
-        cached = _node_metrics_cache_get(timeframe)
-        if cached is not None:
-            return jsonify(cached)
-        
-        # Get local node name
-        # local_node = socket.gethostname()
-        local_node = get_proxmox_node_name()
-
-        # print(f"[v0] Local node: {local_node}")
-        pass
-        
-
-        zfs_arc_size = 0
-        try:
-            with open('/proc/spl/kstat/zfs/arcstats', 'r') as f:
-                for line in f:
-                    if line.startswith('size'):
-                        parts = line.split()
-                        if len(parts) >= 3:
-                            zfs_arc_size = int(parts[2])
-                            break
-        except (FileNotFoundError, PermissionError, ValueError):
-            # ZFS not available or no access
-            pass
-
-        # Get RRD data for the node
-
-        rrd_result = subprocess.run(['pvesh', 'get', f'/nodes/{local_node}/rrddata', 
-                                    '--timeframe', timeframe, '--output-format', 'json'],
-                                   capture_output=True, text=True, timeout=10)
-        
-        # Detect well-known Proxmox-side failures BEFORE trying to parse
-        # the JSON. These are PVE host problems (rrdcached down, RRD file
-        # corrupt, node-name mismatch). None of them are caused by the
-        # Monitor itself — surface a specific message so the operator
-        # doesn't blame ProxMenux for a Proxmox-host data-store issue.
-        if rrd_result.returncode != 0:
-            stderr_str = (rrd_result.stderr or '') + (rrd_result.stdout or '')
-            stderr_lower = stderr_str.lower()
-            if 'mmaping file' in stderr_lower and 'invalid argument' in stderr_lower:
-                # Corrupt RRD file on disk. Operator must recreate it.
-                return jsonify({
-                    'error': 'Proxmox RRD database is corrupt',
-                    'details': (
-                        'The host metrics file Proxmox keeps under '
-                        '/var/lib/rrdcached/db/pve-node-9.0/ failed to '
-                        'memory-map (Invalid argument). This is a Proxmox-side '
-                        'data-store issue, not a Monitor bug.'
-                    ),
-                    'suggestion': (
-                        'Stop pvestatd + pve-cluster + rrdcached, move the '
-                        'broken RRD aside, restart the services. Proxmox will '
-                        'rebuild the RRD from scratch (history is lost).'
-                    ),
-                    'raw': stderr_str.strip()[:500],
-                }), 503
-            if 'no such file' in stderr_lower or 'no such node' in stderr_lower or 'does not exist' in stderr_lower:
-                return jsonify({
-                    'error': 'Proxmox node name mismatch',
-                    'details': (
-                        f"pvesh could not find node '{local_node}'. The "
-                        'usual cause is that the host was renamed after '
-                        'Proxmox was installed, so /etc/pve/nodes/ still '
-                        'carries the old name. This is a Proxmox-side '
-                        'config issue, not a Monitor bug.'
-                    ),
-                    'suggestion': 'Compare `hostname` with `ls /etc/pve/nodes/` — they must match.',
-                    'raw': stderr_str.strip()[:500],
-                }), 503
-            if 'rrd' in stderr_lower or 'empty' in stderr_lower:
-                return jsonify({
-                    'error': 'Proxmox RRD data not available',
-                    'details': 'The RRD database appears empty. Proxmox may not have collected metrics yet (fresh install) or rrdcached was down at boot.',
-                    'suggestion': 'systemctl restart rrdcached pvestatd ; wait ~5 min and reload this page.',
-                    'raw': stderr_str.strip()[:500],
-                }), 503
-            return jsonify({
-                'error': 'Proxmox metrics command failed',
-                'details': 'pvesh exited non-zero. Check Proxmox host status.',
-                'raw': stderr_str.strip()[:500],
-            }), 503
-
-        if rrd_result.returncode == 0:
-            rrd_data = json.loads(rrd_result.stdout)
-
-            # PVE 9.x exposes the actual ARC history as `arcsize` in RRD;
-            # the previous code ignored it and stamped every point with
-            # the live ARC size, producing a flat band at the current
-            # value (issue: ZFS ARC line painted full-bar). Use the real
-            # series when present so the chart matches Proxmox's own
-            # Summary view. On older PVE that doesn't expose `arcsize`,
-            # fall back to the live value as a constant placeholder.
-            for item in rrd_data:
-                if 'arcsize' in item:
-                    item['zfsarc'] = item['arcsize']
-                elif zfs_arc_size > 0 and ('zfsarc' not in item or item.get('zfsarc', 0) == 0):
-                    item['zfsarc'] = zfs_arc_size
-
-            # Period stats — computed BEFORE downsampling so the
-            # AVG/MAX/MIN header in the chart reflects real per-minute
-            # extremes instead of averages.
-            #
-            # Three sources depending on the timeframe:
-            #
-            #   - hour/day  → PVE returns 1-min raw points. AVG/MAX/MIN
-            #     of the in-memory list IS the truth.
-            #
-            #   - week/month → PVE already downsamples to 30-min /
-            #     ~1-hour points using consolidation function AVG, so
-            #     the in-memory points are already averages. Taking
-            #     max() of them gives "max of averages", NOT the real
-            #     peak. We issue two extra pvesh calls per request
-            #     (`--cf MAX` and `--cf MIN`) to recover the real
-            #     extremes from PVE's own RRD consolidation. The
-            #     extra calls add ~150 ms — only on week/month and
-            #     only when the chart loads, so the overhead is small.
-            def _values_from(items, field_key, scale=1.0):
-                return [item[field_key] * scale for item in items
-                        if isinstance(item.get(field_key), (int, float))
-                        and not isinstance(item[field_key], bool)
-                        and item[field_key] is not None]
-
-            def _stats_native(field_key, scale=1.0):
-                values = _values_from(rrd_data, field_key, scale)
-                if not values:
-                    return None
-                return {
-                    'avg': sum(values) / len(values),
-                    'max': max(values),
-                    'min': min(values),
-                }
-
-            def _pvesh_rrd(cf):
-                """One extra pvesh call with a non-default CF.
-                Returns the parsed list or None on any failure — caller
-                falls back to the AVG-based numbers."""
-                try:
-                    extra = subprocess.run(
-                        ['pvesh', 'get', f'/nodes/{local_node}/rrddata',
-                         '--timeframe', timeframe, '--cf', cf,
-                         '--output-format', 'json'],
-                        capture_output=True, text=True, timeout=10,
-                    )
-                    if extra.returncode == 0 and extra.stdout:
-                        return json.loads(extra.stdout)
-                except (subprocess.SubprocessError, json.JSONDecodeError, OSError):
-                    pass
-                return None
-
-            def _build_stats(field_key, scale=1.0):
-                native = _stats_native(field_key, scale)
-                if native is None:
-                    return None
-                # On week/month, the points we already have are AVG.
-                # Try to upgrade max/min to the real RRD extremes.
-                if timeframe in ('week', 'month'):
-                    cf_max = _pvesh_rrd('MAX')
-                    if cf_max:
-                        vals = _values_from(cf_max, field_key, scale)
-                        if vals:
-                            native['max'] = max(vals)
-                    cf_min = _pvesh_rrd('MIN')
-                    if cf_min:
-                        vals = _values_from(cf_min, field_key, scale)
-                        if vals:
-                            native['min'] = min(vals)
-                return native
-
-            period_stats = {
-                # cpu: RRD stores fraction 0-1, surface as %.
-                'cpu': _build_stats('cpu', scale=100.0),
-                # memory_used: bytes → GB so units match the chart.
-                'memory_used': _build_stats('memused', scale=1 / (1024 ** 3)),
-            }
-
-            # 24h downsampling: RRD returns ~1440 minute-level points which
-            # plots as a dense thicket of vertical spikes. Group into 5-min
-            # buckets and average each numeric field — same shape that
-            # `get_temperature_history` uses for its 24h view so the look
-            # is consistent across the dashboard's 24h charts.
-            if timeframe == 'day' and rrd_data:
-                bucket_seconds = 300  # 5-min
-                buckets = {}
-                for item in rrd_data:
-                    t = item.get('time')
-                    if t is None:
-                        continue
-                    bk = (int(t) // bucket_seconds) * bucket_seconds
-                    if bk not in buckets:
-                        buckets[bk] = {'_count': 0, '_sums': {}}
-                    b = buckets[bk]
-                    b['_count'] += 1
-                    for k, v in item.items():
-                        if k == 'time' or not isinstance(v, (int, float)) or isinstance(v, bool):
-                            continue
-                        b['_sums'][k] = b['_sums'].get(k, 0) + v
-                rrd_data = []
-                for bk in sorted(buckets.keys()):
-                    b = buckets[bk]
-                    point = {'time': bk}
-                    for k, total in b['_sums'].items():
-                        point[k] = total / b['_count']
-                    rrd_data.append(point)
-
-            payload = {
-                'node': local_node,
-                'timeframe': timeframe,
-                'data': rrd_data,
-                # AVG/MAX/MIN computed over the raw (pre-downsampling)
-                # points so the chart header captures real per-minute
-                # extremes even on multi-day timeframes.
-                'period_stats': period_stats,
-            }
-            _node_metrics_cache_set(timeframe, payload)
-            return jsonify(payload)
-        # Note: the old `else` branch that handled rrd_result.returncode != 0
-        # was removed — the early-return block above now catches every
-        # non-zero exit BEFORE we ever attempt json.loads(), so reaching
-        # this point with returncode != 0 is impossible.
-
-    except json.JSONDecodeError:
-        # pvesh returned invalid JSON - likely empty RRD
-        return jsonify({
-            'error': 'Proxmox RRD data not available',
-            'details': 'pvesh returned non-JSON output. The RRD database is likely empty (fresh install where pvestatd has not run yet) or the rrdcached daemon is down.',
-            'suggestion': 'systemctl restart rrdcached pvestatd ; wait ~5 min and reload.',
-        }), 503
-    except Exception as e:
-        return jsonify({'error': str(e)}), 500
+        return jsonify(_get_node_metrics_payload(timeframe))
+    except _NodeMetricsError as exc:
+        response = jsonify(exc.payload)
+        response.status_code = 503
+        response.headers['Retry-After'] = str(int(_NODE_METRICS_RETRY_DELAY))
+        return response
 
 @app.route('/api/logs/counts', methods=['GET'])
 @require_auth
@@ -13572,18 +13402,10 @@ def api_lxc_updates_detection_set():
 def api_vm_apps_get(vmid):
     try:
         import lxc_apps
-        cached = _vm_cache_get(_vm_apps_cache, vmid, _VM_APPS_TTL)
-        if cached is not None:
-            # Annotated on the way out, never on the way in: this cache is
-            # invalidated by events, not by time, and the Docker inventory it
-            # reads is built asynchronously. Annotating before storing would
-            # freeze whatever was known at first read — for an app registered
-            # before the first scan, permanently no available version.
-            lxc_apps.annotate_delegated_apps(cached.get('apps') or [], _get_lxc_docker_inventory_map().get(str(vmid)))
-            return jsonify(cached)
         sidecar = lxc_apps.load_sidecar(vmid)
         payload = sidecar if sidecar else {'vmid': vmid, 'apps': []}
-        _vm_cache_put(_vm_apps_cache, vmid, payload)
+        # Read the shared, file-versioned snapshot; annotate only on the way
+        # out so asynchronously collected Docker metadata remains current.
         lxc_apps.annotate_delegated_apps(payload.get('apps') or [], _get_lxc_docker_inventory_map().get(str(vmid)))
         return jsonify(payload)
     except Exception as e:
@@ -13599,7 +13421,6 @@ def api_vm_apps_add(vmid):
         ok, result = lxc_apps.add_app(vmid, payload)
         if not ok:
             return jsonify({'error': result}), 400
-        _vm_cache_put(_vm_apps_cache, vmid, result)
         return jsonify(result)
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -13635,7 +13456,6 @@ def api_vm_apps_update(vmid, app_id):
         if not ok:
             code = 404 if 'not found' in str(result).lower() else 400
             return jsonify({'error': result}), code
-        _vm_cache_put(_vm_apps_cache, vmid, result)
         return jsonify(result)
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -13648,7 +13468,6 @@ def api_vm_apps_delete_one(vmid, app_id):
         import lxc_apps
         ok = lxc_apps.delete_app(vmid, app_id)
         sidecar = lxc_apps.load_sidecar(vmid) or {'vmid': vmid, 'apps': []}
-        _vm_cache_put(_vm_apps_cache, vmid, sidecar)
         return jsonify({**sidecar, 'success': ok, 'app_id': app_id}), 200
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -13661,7 +13480,6 @@ def api_vm_apps_delete_all(vmid):
         import lxc_apps
         ok = lxc_apps.delete_all(vmid)
         sidecar = {'vmid': vmid, 'apps': []}
-        _vm_cache_put(_vm_apps_cache, vmid, sidecar)
         return jsonify({**sidecar, 'success': ok}), 200
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -13675,7 +13493,6 @@ def api_vm_apps_check_one(vmid, app_id):
         sidecar = lxc_apps.check_app(vmid, app_id, force=True)
         if not sidecar:
             return jsonify({'error': 'app not found'}), 404
-        _vm_cache_put(_vm_apps_cache, vmid, sidecar)
         return jsonify(sidecar)
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -13689,7 +13506,6 @@ def api_vm_apps_check_all(vmid):
         sidecar = lxc_apps.check_all(vmid, force=True)
         if not sidecar:
             sidecar = {'vmid': vmid, 'apps': []}
-        _vm_cache_put(_vm_apps_cache, vmid, sidecar)
         return jsonify(sidecar)
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -14077,7 +13893,6 @@ def api_vm_apps_dismiss(vmid):
         ok, result = lxc_apps.set_dismissed_slug(vmid, slug, dismissed)
         if not ok:
             return jsonify({'error': result}), 400
-        _vm_cache_put(_vm_apps_cache, vmid, result)
         return jsonify(result)
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -14398,11 +14213,8 @@ def _finalize_lxc_update(
                     vmid,
                     [item[4:] for item in requested if item.startswith('app:')],
                 )
-            refreshed_sidecar = lxc_apps.check_all(vmid, force=True)
-            refreshed_sidecar = refreshed_sidecar or {'vmid': vmid, 'apps': []}
-            _vm_cache_put(_vm_apps_cache, vmid, refreshed_sidecar)
+            lxc_apps.check_all(vmid, force=True)
         except Exception as exc:
-            _vm_cache_invalidate(vmid, _vm_apps_cache)
             verification_errors.append(f'application refresh failed: {exc}')
         docker_attempted = refresh_docker_inventory or any(
             target_id.startswith('docker-') for target_id in requested
@@ -15109,18 +14921,19 @@ def api_vms_modal_cache_all():
     frontend replace the current 84-request warm-up (4 endpoints ×
     ~21 guests) with a single fetch on page load.
 
-    Reads **exclusively** from the in-memory caches populated by
+    Reads modal caches populated by
     the backend prewarmer (`_vm_modal_prewarmer_loop`). Never falls
     through to a live handler call — that would let a single cold
-    guest block the whole bulk response for 10-20s. If a guest is
-    not yet cached the corresponding field is `null` and the client
-    fetches that one endpoint dirigido on demand.
+    guest block the whole bulk response for 10-20s. Registered apps use
+    the shared sidecar snapshot, checking only local file metadata for changes.
+    If a guest's other modal data is not yet cached, that field is `null`
+    and the client fetches the corresponding endpoint on demand.
 
     Trade-off: for the ~20-70s window right after `systemctl
     restart proxmenux-monitor` some fields come back `null`; the
     client transparently falls back to per-endpoint fetches for
-    those. Once the initial warm-up finishes the entire response
-    is served from dict reads (<20ms even with 30+ guests).
+    those. Once the initial warm-up finishes, registered apps reuse their
+    parsed snapshots and the other fields are served from dict reads.
 
     Response shape:
         {
@@ -15150,7 +14963,8 @@ def api_vms_modal_cache_all():
                 'backups': _vm_cache_get(_vm_backups_cache, vmid, _VM_BACKUPS_TTL),
             }
             if vm_type == 'lxc':
-                entry['apps'] = _vm_cache_get(_vm_apps_cache, vmid, _VM_APPS_TTL)
+                import lxc_apps
+                entry['apps'] = lxc_apps.load_sidecar(vmid) or {'vmid': vmid, 'apps': []}
                 entry['suggestions'] = _vm_cache_get(
                     _vm_app_suggestions_cache, vmid,
                     _VM_APP_SUGGESTIONS_TTL,
@@ -21887,7 +21701,8 @@ def _compose_scheduled_update_command(vmid: int, target: str, targets: list[str]
             and cmd == _DOCKER_ENGINE_INTEGRATED_COMMAND
         )
         if cmd and not is_integrated_docker_command:
-            parts.append(cmd)
+            # Protect each legacy launcher before joining the multi-app plan.
+            parts.append(lxc_apps.protect_download_update_command(cmd))
     selected_projects = {
         item.split(":", 1)[1]
         for item in targets
@@ -21961,7 +21776,7 @@ def _scheduled_helper_enabled(vmid: int, target: str, targets: list[str]) -> boo
     # conservatively instead of running both methods for the same app.
     if any((app.get("update_command") or "").strip() for app in matching_apps):
         return False
-    return True
+    return lxc_apps.helper_update_selected(vmid, helper_slug, targets)
 
 
 def _resolve_bulk_update_plan(vmid: int, targets: list[str]) -> dict:
@@ -22004,6 +21819,7 @@ def _resolve_bulk_update_plan(vmid: int, targets: list[str]) -> dict:
 
     def add_command(command: str) -> None:
         command = str(command or '').strip()
+        command = lxc_apps.protect_download_update_command(command)
         if command and command not in commands:
             commands.append(command)
 
@@ -22041,7 +21857,8 @@ def _resolve_bulk_update_plan(vmid: int, targets: list[str]) -> dict:
         command = str(app.get('update_command') or '').strip()
         if command:
             add_command(command)
-        elif helper_enabled and app.get('helper_slug'):
+        elif (helper_enabled and app.get('update_method') == 'helper'
+              and _scheduled_helper_enabled(vmid, 'app', [target_id])):
             run_helper = True
         else:
             unavailable.append({'target': target_id, 'reason': 'no executable update method is available'})
@@ -22249,6 +22066,36 @@ def _run_scheduled_update(vmid: int, sched: dict) -> dict:
         registered_apps = (lxc_apps._read_sidecar(vmid) or {}).get("apps") or []
     except Exception:
         registered_apps = []
+    # A saved schedule can outlive an application's updater choice. Keep
+    # its configuration, but explicitly report unavailable targets instead
+    # of silently treating an OS-only/no-op run as a complete app update.
+    helper_ready = _scheduled_helper_enabled(vmid, 'app', targets)
+    available_app_ids = {
+        str(app.get('id')) for app in registered_apps
+        if not app.get('managed_oci_app_id') and (
+            (app.get('update_command') or '').strip()
+            or app.get('helper_slug') == 'docker'
+            or (helper_ready and app.get('update_method') == 'helper'
+                and _scheduled_helper_enabled(vmid, 'app', [f"app:{app.get('id')}"]))
+        )
+    }
+    filtered_targets = []
+    for value in targets:
+        if value == 'apps':
+            eligible = [f"app:{app.get('id')}" for app in registered_apps
+                        if str(app.get('id')) in available_app_ids
+                        and app.get('helper_slug') != 'docker']
+            if eligible:
+                filtered_targets.extend(eligible)
+            else:
+                deferred_targets.append(value)
+                reasons.append('no application update method has been selected or is available')
+        elif value.startswith('app:') and value.split(':', 1)[1] not in available_app_ids:
+            deferred_targets.append(value)
+            reasons.append(f'{value}: update method is not selected or no longer available')
+        else:
+            filtered_targets.append(value)
+    targets = list(dict.fromkeys(filtered_targets))
     if any(value.startswith("docker-") for value in targets):
         docker_registered = any(app.get("helper_slug") == "docker" for app in registered_apps)
         if not docker_registered:
@@ -22721,22 +22568,16 @@ if __name__ == '__main__':
     except Exception as e:
         print(f"[ProxMenux] app-updates startup emitter failed to arm: {e}")
 
-    # ── Node-metrics Prewarmer ──
-    # Keeps `_NODE_METRICS_CACHE` hot for every timeframe (hour / day /
-    # week / month / year) on a 30 s cadence, so the Overview page's
-    # CPU + memory charts never wait on `pvesh get rrddata` when the
-    # user opens the dashboard. Cache TTL is 60 s; the loop refreshes
-    # every 30 s, giving 30 s of headroom against transient pvesh
-    # latency.
+    # Prewarm only the Overview's default day range; other ranges are lazy.
     try:
         metrics_thread = threading.Thread(target=_node_metrics_prewarmer_loop, daemon=True, name='node-metrics-prewarmer')
         metrics_thread.start()
-        print("[ProxMenux] Node-metrics prewarmer started (30s interval)")
+        print("[ProxMenux] Node-metrics prewarmer started (day range, 90s interval)")
     except Exception as e:
         print(f"[ProxMenux] Node-metrics prewarmer failed to start: {e}")
 
     # ── VM/CT modal-cache prewarmer ──
-    # Keeps _vm_details_cache / _vm_backups_cache / _vm_apps_cache /
+    # Keeps _vm_details_cache / _vm_backups_cache / app snapshots /
     # _vm_schedule_cache warm from the backend so the "Loading
     # configuration..." message never appears on modal open, even
     # after the tab has been closed for minutes. The React-side

AppImage/scripts/lxc_apps.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 import datetime
+import copy
 import concurrent.futures
 import hashlib
 import json
@@ -418,15 +419,43 @@ def _now_iso() -> str:
     return datetime.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
 
 
+_sidecar_cache: dict = {}
+_sidecar_cache_lock = threading.RLock()
+_sidecar_revision = 0
+
+
+def _sidecar_signature(stat) -> tuple:
+    return (stat.st_dev, stat.st_ino, stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size)
+
+
+def _publish_sidecar_snapshot(path: str, data: dict, signature: tuple) -> dict:
+    """Publish under _sidecar_cache_lock; revisions exist only in memory."""
+    global _sidecar_revision
+    _sidecar_revision = max(_sidecar_revision + 1, int(time.time() * 1000))
+    snapshot = _migrate_legacy(copy.deepcopy(data))
+    _migrate_update_methods(snapshot)
+    snapshot['_revision'] = _sidecar_revision
+    _sidecar_cache[path] = (signature, snapshot)
+    return snapshot
+
+
 def _read_sidecar(vmid) -> Optional[dict]:
     path = _sidecar_path(vmid)
-    try:
-        with open(path) as f:
-            data = json.load(f)
-        if isinstance(data, dict):
-            return _migrate_legacy(data)
-    except (FileNotFoundError, json.JSONDecodeError, OSError):
-        pass
+    with _sidecar_cache_lock:
+        try:
+            signature = _sidecar_signature(os.stat(path))
+            cached = _sidecar_cache.get(path)
+            if cached is not None and cached[0] == signature:
+                return copy.deepcopy(cached[1])
+            with open(path) as f:
+                data = json.load(f)
+                signature = _sidecar_signature(os.fstat(f.fileno()))
+            if isinstance(data, dict):
+                snapshot = _publish_sidecar_snapshot(path, data, signature)
+                return copy.deepcopy(snapshot)
+        except (FileNotFoundError, json.JSONDecodeError, OSError):
+            pass
+        _sidecar_cache.pop(path, None)
     return None
 
 
@@ -466,15 +495,107 @@ def _migrate_legacy(data: dict) -> dict:
             "updated_at": data.get("updated_at") or _now_iso()}
 
 
+def _migrate_update_methods(data: dict) -> None:
+    """Preserve saved choices, never turn detection into updater consent.
+
+    Old commands and explicitly saved bulk/enabled schedule selections keep
+    working. New registrations always carry update_method, so an old `apps`
+    wildcard cannot opt newly registered applications into Helper-Scripts.
+    Projection is read-only; the next normal sidecar write persists it.
+    """
+    schedule = data.get("schedule") or {}
+    selected = set((data.get("bulk_update") or {}).get("targets") or [])
+    if schedule.get("enabled"):
+        targets = schedule.get("targets")
+        if not targets:
+            targets = ["apps"] if schedule.get("target", "both") in ("app", "both") else []
+        selected.update(targets)
+    for app in data.get("apps") or []:
+        if "update_method" in app:
+            continue
+        if (app.get("update_command") or "").strip():
+            app["update_method"] = "custom"
+        elif (app.get("helper_slug") and app.get("helper_slug") not in ("docker", "adguard")
+              and ("apps" in selected or f"app:{app.get('id')}" in selected)):
+            app["update_method"] = "helper"
+        else:
+            app["update_method"] = "none"
+
+
+def protect_download_update_command(command: str) -> str:
+    """Guard the historical downloaded-shell launcher at execution time.
+
+    Only a literal, standalone wget/curl + shell -c launcher is recognised.
+    Other custom commands are returned byte-for-byte, never evaluated here.
+    Saved configuration is not rewritten. Grouping preserves && composition.
+    """
+    launcher = re.fullmatch(
+        r'''\s*(?P<prefix>PHS_SILENT=[01][ \t]+)?(?P<shell>(?:/bin/|/usr/bin/)?(?:bash|sh))[ \t]+-c[ \t]+"\$\((?P<fetch>[^\n]+)\)"\s*''',
+        command,
+    )
+    if not launcher:
+        return command
+    fetch = re.fullmatch(
+        r'''(?P<tool>wget|curl)[ \t]+(?P<flags>-qLO[ \t]+-|-qO[ \t]+-|-qO-|-fsSL|-fSL)[ \t]+(?P<quote>['"]?)(?P<url>https?://[A-Za-z0-9_./:%?=&+#@,~!;-]+)(?P=quote)''',
+        launcher['fetch'],
+    )
+    if not fetch:
+        return command
+    # Require the original shell token to be literal too. Unquoted shell
+    # operators or glob patterns are not this known launcher format.
+    if not fetch['quote'] and any(c in fetch['url'] for c in '&;?'):
+        return command
+    flags = fetch['flags'].split()
+    if ((fetch['tool'] == 'wget' and flags not in (['-qLO', '-'], ['-qO', '-'], ['-qO-']))
+            or (fetch['tool'] == 'curl' and flags not in (['-fsSL'], ['-fSL']))):
+        return command
+    fetch_command = shlex.join([fetch['tool'], *flags, fetch['url']])
+    invocation = (launcher['prefix'] or '') + launcher['shell']
+    return (
+        '(\n'
+        f'_proxmenux_updater=$({fetch_command}) || {{\n'
+        '  echo "ERROR: updater download failed; nothing was executed." >&2\n'
+        '  exit 1\n'
+        '}\n'
+        '[ -n "$_proxmenux_updater" ] || {\n'
+        '  echo "ERROR: downloaded updater is empty; nothing was executed." >&2\n'
+        '  exit 1\n'
+        '}\n'
+        f'{invocation} -c "$_proxmenux_updater"\n'
+        ')'
+    )
+
+
+def helper_update_selected(vmid, slug: str, targets=None) -> bool:
+    """Execution-time consent check, shared with the shell runner.
+
+    Wrapper provenance is independently verified by the caller. Duplicate
+    registrations with different choices must not run a CT-wide helper.
+    """
+    apps = (_read_sidecar(vmid) or {}).get("apps") or []
+    matching = [app for app in apps if app.get("helper_slug") == slug
+                and not app.get("managed_oci_app_id")]
+    if not matching or any(app.get("update_method") != "helper"
+                           or (app.get("update_command") or "").strip() for app in matching):
+        return False
+    if targets is None or "apps" in targets:
+        return True
+    return any(f"app:{app.get('id')}" in targets for app in matching)
+
+
 def _write_sidecar(vmid, data: dict) -> bool:
+    _migrate_update_methods(data)
     _ensure_dir()
     path = _sidecar_path(vmid)
     tmp = f"{path}.tmp.{os.getpid()}"
     try:
         with open(tmp, "w") as f:
-            json.dump(data, f, indent=2, sort_keys=True)
+            json.dump({k: v for k, v in data.items() if k != '_revision'}, f, indent=2, sort_keys=True)
         os.chmod(tmp, 0o600)
-        os.replace(tmp, path)
+        with _sidecar_cache_lock:
+            os.replace(tmp, path)
+            snapshot = _publish_sidecar_snapshot(path, data, _sidecar_signature(os.stat(path)))
+            data['_revision'] = snapshot['_revision']
         return True
     except OSError as e:
         print(f"[ProxMenux] lxc_apps: could not write sidecar {path}: {e}")
@@ -1092,6 +1213,17 @@ def validate_config(payload: dict) -> tuple[bool, Any]:
             # two-step strategy payloads and normalize them on write.
             conf["update_strategy"] = "custom_override"
 
+    method = payload.get("update_method", "custom" if conf.get("update_command") else "none")
+    if method not in ("none", "helper", "custom"):
+        return _err("update_method must be none, helper or custom")
+    if method == "custom" and not conf.get("update_command"):
+        return _err("update_command is required for update_method=custom")
+    if method != "custom" and conf.get("update_command"):
+        return _err("update_command is only allowed for update_method=custom")
+    if method == "helper" and (not hs or hs in ("docker", "adguard")):
+        return _err("a supported helper_slug is required for update_method=helper")
+    conf["update_method"] = method
+
     # Optional per-app dismiss flag for the "no update method defined"
     # notice shown in the Updates tab. Only affects the notice card;
     # the App tab keeps its purple update signal regardless.
@@ -4057,6 +4189,7 @@ def _summarise_app(app: dict) -> dict:
         "health_path": app.get("health_path"),
         "installed_version": state.get("installed_version"),
         "latest_version": state.get("latest_version"),
+        "latest_published_at": state.get("latest_published_at"),
         "update_available": state.get("update_available"),
         "error": state.get("error"),
         "checked_at": state.get("checked_at"),
@@ -4066,6 +4199,7 @@ def _summarise_app(app: dict) -> dict:
         # it) and whether the "no method" notice is suppressed for
         # this app.
         "update_command": app.get("update_command") or "",
+        "update_method": app.get("update_method", "custom" if app.get("update_command") else "none"),
         # Compatibility field for older clients. The only supported
         # strategy is now replacement; legacy sidecars are normalized
         # in the API even before their next write.
@@ -4273,7 +4407,10 @@ def get_active_apps() -> dict:
         apps = sidecar.get("apps") or []
         if not apps:
             continue
-        out[str(vmid)] = [_summarise_app(a) for a in apps]
+        out[str(vmid)] = [
+            {**_summarise_app(a), 'state_revision': sidecar.get('_revision')}
+            for a in apps
+        ]
     return out
 
 

AppImage/scripts/lxc_mount_points.py

@@ -123,6 +123,20 @@ def _read_lxc_config(vmid: str) -> list[dict[str, Any]]:
     return out
 
 
+def _mount_target_key(target: str) -> str:
+    """Return a stable comparison key for a CT-side mount target.
+
+    Proxmox config accepts a trailing slash in ``mp=/path/`` while
+    ``/proc/<pid>/mounts`` reports the realised target as ``/path``.
+    They name the same mount point, so comparisons must not treat the
+    spelling difference as a runtime divergence. Keep the root path
+    intact: stripping its only slash would turn it into an empty key.
+    """
+    if target == "/":
+        return target
+    return target.rstrip("/")
+
+
 # ---------------------------------------------------------------------------
 # Type classification + source resolution
 # ---------------------------------------------------------------------------
@@ -605,14 +619,14 @@ def get_lxc_mount_points_static(vmid: str) -> dict[str, Any]:
     if running and host_pid:
         try:
             config_targets = {
-                entry.get("target", "")
+                _mount_target_key(entry.get("target", ""))
                 for entry in config_entries
                 if entry.get("target")
             }
             for rt in _read_ct_proc_mounts(host_pid):
                 if not _REMOTE_FS_RE.match(rt.get("rt_fstype", "")):
                     continue
-                if rt.get("rt_target") in config_targets:
+                if _mount_target_key(rt.get("rt_target", "")) in config_targets:
                     continue
                 ad_hoc_hint_count += 1
         except Exception:
@@ -658,7 +672,9 @@ def get_lxc_mount_points_runtime(vmid: str) -> dict[str, Any]:
     # mount point are I/O-bound. Serialised, a CT with 5+ binds
     # tripped Caddy's 3s reverse-proxy timeout.
     from concurrent.futures import ThreadPoolExecutor
-    rt_by_target: dict[str, dict[str, Any]] = {m["rt_target"]: m for m in rt_mounts}
+    rt_by_target: dict[str, dict[str, Any]] = {
+        _mount_target_key(m["rt_target"]): m for m in rt_mounts
+    }
 
     runtime_by_target: dict[str, dict[str, Any]] = {}
     matched_targets: set[str] = set()
@@ -673,9 +689,10 @@ def get_lxc_mount_points_runtime(vmid: str) -> dict[str, Any]:
             host_pid=host_pid if running else "",
             target=tgt,
         )
-        live_target = bool(running and tgt and tgt in rt_by_target)
+        target_key = _mount_target_key(tgt)
+        live_target = bool(running and tgt and target_key in rt_by_target)
         health = _stat_via_host(host_pid, tgt) if live_target else None
-        return entry, capacity, live_target, health
+        return entry, capacity, target_key, live_target, health
 
     if config_entries:
         max_workers = max(2, min(8, len(config_entries)))
@@ -684,11 +701,11 @@ def get_lxc_mount_points_runtime(vmid: str) -> dict[str, Any]:
     else:
         gathered = []
 
-    for entry, cap, live_target, health in gathered:
+    for entry, cap, target_key, live_target, health in gathered:
         target = entry.get("target", "")
         rt_item: dict[str, Any] = {**cap}
         if live_target:
-            rt = rt_by_target[target]
+            rt = rt_by_target[target_key]
             rt_item.update({
                 "runtime_mounted": True,
                 "runtime_source": rt["rt_source"],
@@ -698,7 +715,7 @@ def get_lxc_mount_points_runtime(vmid: str) -> dict[str, Any]:
                 "runtime_reachable": health["reachable"],
                 "runtime_error": health["error"],
             })
-            matched_targets.add(target)
+            matched_targets.add(target_key)
         elif running:
             rt_item["runtime_mounted"] = False
             rt_item["runtime_error"] = "configured but not mounted"
@@ -712,7 +729,7 @@ def get_lxc_mount_points_runtime(vmid: str) -> dict[str, Any]:
     if running:
         ad_hoc_candidates = [
             rt for rt in rt_mounts
-            if rt["rt_target"] not in matched_targets
+            if _mount_target_key(rt["rt_target"]) not in matched_targets
             and _REMOTE_FS_RE.match(rt["rt_fstype"])
         ]
         if ad_hoc_candidates:
@@ -776,7 +793,9 @@ def get_lxc_mount_points(vmid: str) -> dict[str, Any]:
 
     # Index runtime mounts by their CT-side target path so we can
     # match a config entry to its current realised state in O(1).
-    rt_by_target: dict[str, dict[str, Any]] = {m["rt_target"]: m for m in rt_mounts}
+    rt_by_target: dict[str, dict[str, Any]] = {
+        _mount_target_key(m["rt_target"]): m for m in rt_mounts
+    }
 
     out: list[dict[str, Any]] = []
     matched_targets: set[str] = set()
@@ -801,15 +820,16 @@ def get_lxc_mount_points(vmid: str) -> dict[str, Any]:
             target=tgt,
         )
         host_src = _host_source_state(src)
-        live_target = bool(running and tgt and tgt in rt_by_target)
+        target_key = _mount_target_key(tgt)
+        live_target = bool(running and tgt and target_key in rt_by_target)
         health = _stat_via_host(host_pid, tgt) if live_target else None
-        return entry, classification, capacity, host_src, live_target, health
+        return entry, classification, capacity, host_src, target_key, live_target, health
 
     max_workers = max(2, min(8, len(config_entries) or 1))
     with ThreadPoolExecutor(max_workers=max_workers) as pool:
         gathered = list(pool.map(_gather_one, config_entries))
 
-    for entry, cls, cap, host_src, live_target, health in gathered:
+    for entry, cls, cap, host_src, target_key, live_target, health in gathered:
         source = entry.get("source", "")
         target = entry.get("target", "")
 
@@ -830,7 +850,7 @@ def get_lxc_mount_points(vmid: str) -> dict[str, Any]:
 
         # Runtime enrichment when CT is up.
         if live_target:
-            rt = rt_by_target[target]
+            rt = rt_by_target[target_key]
             item.update({
                 "runtime_mounted": True,
                 "runtime_source": rt["rt_source"],
@@ -840,7 +860,7 @@ def get_lxc_mount_points(vmid: str) -> dict[str, Any]:
                 "runtime_reachable": health["reachable"],
                 "runtime_error": health["error"],
             })
-            matched_targets.add(target)
+            matched_targets.add(target_key)
         elif running:
             # CT is running but the configured mount isn't in
             # /proc/<pid>/mounts — divergence. Could be a startup
@@ -860,7 +880,7 @@ def get_lxc_mount_points(vmid: str) -> dict[str, Any]:
     if running:
         ad_hoc_candidates = [
             rt for rt in rt_mounts
-            if rt["rt_target"] not in matched_targets
+            if _mount_target_key(rt["rt_target"]) not in matched_targets
             and _REMOTE_FS_RE.match(rt["rt_fstype"])
         ]
         # Same parallelisation as the configured-mp loop: stat'ing

AppImage/scripts/notification_events.py

@@ -2586,6 +2586,11 @@ class PollingCollector:
         # once for each genuinely new available version. The persistent
         # history is stored in updates_available.json beside the scan.
         self._last_post_install_check = 0
+        # Re-open that announcement once per service start, so a host
+        # carrying optimizations it never applied surfaces them again
+        # after a ProxMenux update or a node reboot instead of staying
+        # silent forever. Consumed by the first check cycle.
+        self._post_install_startup_reset_pending = True
         # Sprint 14.7: fingerprint (item_id → latest_version) of the
         # last managed-installs update notification, across all types
         # in the registry. A new notification fires when the
@@ -3511,6 +3516,22 @@ class PollingCollector:
 
     # ── Post-install function updates check (Sprint 12D) ────────────
 
+    def _reset_post_install_announcements(self):
+        """Re-open the optimization announcement after a service start.
+
+        ``post_install_update`` sits behind a second gate the other
+        update events don't have: a per-version history that keeps a
+        pending optimization from being announced twice. The manager's
+        ``_EVENT_TYPES_RESET_ON_START`` already clears the delivery
+        cooldown, so only that history has to be dropped here for the
+        first cycle to report whatever is still pending.
+        """
+        try:
+            import post_install_versions
+            post_install_versions.reset_notified_versions()
+        except Exception as e:
+            print(f"[PollingCollector] post-install history reset failed: {e}")
+
     def _check_post_install_updates(self):
         """Notify the operator when post-install functions have new versions.
 
@@ -3522,6 +3543,9 @@ class PollingCollector:
         shrinks the pending set and must not produce a second notification.
         """
         now = time.time()
+        if self._post_install_startup_reset_pending:
+            self._post_install_startup_reset_pending = False
+            self._reset_post_install_announcements()
         if now - self._last_post_install_check < self.UPDATE_CHECK_INTERVAL:
             return
         self._last_post_install_check = now

AppImage/scripts/notification_manager.py

@@ -1608,7 +1608,7 @@ class NotificationManager:
         showed zero digest entries even when the schedule was firing
         (issue #233).
         """
-        host = _hostname(self._config)
+        host = _resolve_display_hostname(self._config)
         summary_title = (
             f"{host}: 24h summary ({now.strftime('%Y-%m-%d %H:%M')})"
         )
@@ -1847,23 +1847,44 @@ class NotificationManager:
         if not rows:
             return
 
-        host = _hostname(self._config)
+        host = _resolve_display_hostname(self._config)
         summary_title = (
             f"{host}: {len(rows)} events buffered during Quiet Hours"
         )
         summary_body = self._compose_digest_body(rows)
 
+        result: dict = {'success': False, 'error': ''}
         try:
-            channel.send(summary_title, summary_body, severity='INFO',
-                         data={'_quiet_hours_summary': True, '_count': len(rows)})
+            result = channel.send(
+                summary_title, summary_body, severity='INFO',
+                data={'_quiet_hours_summary': True, '_count': len(rows)},
+            ) or result
         except Exception as e:
             print(f"[NotificationManager] quiet send failed for "
                   f"{ch_name}: {e}")
-            return
+            result = {'success': False, 'error': str(e)}
+
+        if result.get('success'):
+            self._stats['total_sent'] += 1
+            self._stats['last_sent_at'] = datetime.now().isoformat()
+        else:
+            self._stats['total_errors'] += 1
+        # Mirrors the digest path: the release is a real delivery, so it
+        # belongs in the history and the counters the operator reads.
+        self._record_history(
+            'quiet_hours', ch_name, summary_title, summary_body, 'INFO',
+            result.get('success', False), result.get('error', '') or '',
+            'quiet_scheduler',
+        )
 
         # Only drop the rows after a successful send so a transient
         # transport failure (Telegram timeout, SMTP outage) doesn't
-        # lose the user's overnight context.
+        # lose the user's overnight context. A channel reporting failure
+        # without raising counts as a failure here too — otherwise the
+        # buffer is wiped for a summary the operator never received.
+        if not result.get('success'):
+            return
+
         try:
             ids = [r[0] for r in rows]
             conn = sqlite3.connect(str(DB_PATH), timeout=10)
@@ -2082,6 +2103,7 @@ class NotificationManager:
         'secure_gateway_update_available',
         'app_update_available',
         'docker_stack_update_available',
+        'post_install_update',
         # Security events that must not be silenced by stale cooldowns
         # following a Monitor reinstall (Pedro Rico, 19/05).
         'auth_fail',

AppImage/scripts/post_install_versions.py

@@ -434,6 +434,29 @@ def scan(persist: bool = True) -> dict[str, Any]:
     return snapshot
 
 
+def reset_notified_versions() -> None:
+    """Forget which optimization versions have already been announced.
+
+    Each version is announced once, so a host that never applies a
+    pending optimization would otherwise stay silent about it forever.
+    Clearing the history reopens that single announcement.
+
+    The caller owns the timing: the notification collector runs this on
+    its first cycle after a service start, together with clearing the
+    matching delivery cooldown, because forgetting the history while the
+    cooldown still suppresses delivery would consume the announcement
+    without ever sending it.
+    """
+    try:
+        with _cache_lock:
+            scanned_at = float(_cache.get("scanned_at", 0.0) or 0.0)
+            updates = list(_cache.get("updates", []))
+        _write_persisted_snapshot(scanned_at, updates, {})
+    except OSError as e:
+        # Read-only host: de-duplication stays best-effort, as elsewhere.
+        print(f"[post_install_versions] could not reset notified versions: {e}")
+
+
 def scan_at_startup() -> dict[str, Any]:
     """Convenience wrapper called from flask_server startup.