decky-cachyos-update/main.py

"""CachyOS Update - Decky plugin backend.

Runs as root (plugin.json declares the "root" flag) and drives pacman, yay,
flatpak and fwupdmgr non-interactively, streaming every output line to the
frontend.

All user-facing wording lives in the frontend (src/i18n.ts). This module only
ever returns stable identifiers - phase ids, hint ids, reason ids - so the UI
can render them in the user's language. Raw command output is passed through
untranslated on purpose: it is the tool's own output.

This file is also runnable standalone for development on a machine without
Decky installed:

    python3 main.py --selftest
    python3 main.py --check
    python3 main.py --update --dry-run
"""

import asyncio
import json
import os
import re
import shutil
import sys
import time
from collections import deque
from pathlib import Path

try:
    import decky  # type: ignore
except ImportError:  # standalone / development
    decky = None


# --------------------------------------------------------------------------
# decky stub for standalone runs
# --------------------------------------------------------------------------

if decky is None:

    class _StubLogger:
        def _p(self, level, msg, *a):
            print(f"[{level}] {msg % a if a else msg}", file=sys.stderr)

        def debug(self, msg, *a):
            self._p("debug", msg, *a)

        def info(self, msg, *a):
            self._p("info", msg, *a)

        def warning(self, msg, *a):
            self._p("warn", msg, *a)

        def error(self, msg, *a):
            self._p("error", msg, *a)

        def exception(self, msg, *a):
            self._p("error", msg, *a)

    class _StubDecky:
        """Mimics the subset of the decky module that this plugin uses."""

        def __init__(self):
            base = Path(os.environ.get("TMPDIR", "/tmp")) / "decky-cachyos-update-dev"
            (base / "settings").mkdir(parents=True, exist_ok=True)
            (base / "runtime").mkdir(parents=True, exist_ok=True)
            self.DECKY_PLUGIN_SETTINGS_DIR = str(base / "settings")
            self.DECKY_PLUGIN_RUNTIME_DIR = str(base / "runtime")
            self.DECKY_USER = os.environ.get("USER", "root")
            self.DECKY_USER_HOME = os.path.expanduser("~")
            self.DECKY_PLUGIN_VERSION = "dev"
            self.logger = _StubLogger()

        async def emit(self, event, *args):
            if event == "cachyos_update_log":
                print(args[0])
            else:
                print(f"<{event}> {args}", file=sys.stderr)

    decky = _StubDecky()  # type: ignore
    STANDALONE = True
else:
    STANDALONE = False


# --------------------------------------------------------------------------
# constants
# --------------------------------------------------------------------------

LOG_MAX_LINES = 500
PACMAN_LOCK = "/var/lib/pacman/db.lck"

# Weight of each phase in the progress bar, roughly proportional to how long
# each step usually takes on a handheld.
PHASE_WEIGHTS = {
    "keyring": 5,
    "pacman": 40,
    "aur": 30,
    "flatpak_system": 12,
    "flatpak_user": 5,
    "fwupd": 8,
}

DEFAULT_SETTINGS = {
    "enable_aur": True,
    "enable_flatpak": True,
    "enable_fwupd": False,
    "auto_check": True,
    "check_interval_hours": 6,
    "notify_on_updates": True,
}

# Packages whose upgrade means the machine should be restarted. Mirrors the
# trigger list of /usr/share/libalpm/hooks/cachyos-reboot-required.hook.
REBOOT_TRIGGER_PATTERNS = [
    re.compile(r"^linux(-.*)?$"),
    re.compile(r"^systemd(-.*)?$"),
    re.compile(r"^mesa$"),
    re.compile(r"^nvidia(-.*)?$"),
    re.compile(r"^wayland$"),
    re.compile(r"^xorg-server(-.*)?$"),
    re.compile(r"^mkinitcpio(-.*)?$"),
    re.compile(r"^dracut(-.*)?$"),
    re.compile(r"^(amd|intel)-ucode$"),
    re.compile(r"^linux-firmware(-.*)?$"),
    re.compile(r"^glibc$"),
]

# Output markers that mean "this needs a human", mapped to a hint id that the
# frontend turns into localized advice. Order matters: first match wins.
ERROR_HINTS = [
    ("unresolvable package conflicts", "pkg_conflict"),
    ("conflicting files", "file_conflict"),
    ("exists in filesystem", "file_conflict"),
    ("unresolvable dependencies", "unresolvable_deps"),
    ("invalid or corrupted package", "corrupt_package"),
    ("signature from", "bad_signature"),
    ("failed to synchronize", "sync_failed"),
    ("could not resolve host", "no_network"),
    ("failed to obtain auth", "fwupd_auth"),
    ("not authorized", "polkit_denied"),
    ("ac power required", "ac_power_required"),
    ("battery level is too low", "battery_too_low"),
]

PROGRESS_RE = re.compile(r"\(\s*(\d+)\s*/\s*(\d+)\s*\)")
UPGRADING_RE = re.compile(r"upgrading\s+([a-zA-Z0-9@._+-]+)")


# --------------------------------------------------------------------------
# state
# --------------------------------------------------------------------------


class State:
    def __init__(self):
        self.status = "idle"  # idle | checking | updating | done | error
        self.phase = ""
        self.progress = 0.0
        self.log = deque(maxlen=LOG_MAX_LINES)
        self.counts = {"pacman": 0, "aur": 0, "flatpak": 0, "fwupd": 0}
        self.details = {"pacman": [], "aur": [], "flatpak": [], "fwupd": []}
        self.last_check = 0.0
        self.last_update = 0.0
        self.reboot_required = False
        self.pacnew = []
        self.failed_phases = []
        self.hint_id = ""
        self.warnings = []
        self.error_text = ""  # only for unexpected exceptions

    def to_dict(self):
        return {
            "status": self.status,
            "phase": self.phase,
            "progress": self.progress,
            "log": list(self.log),
            "counts": self.counts,
            "details": self.details,
            "total": sum(self.counts.values()),
            "last_check": self.last_check,
            "last_update": self.last_update,
            "reboot_required": self.reboot_required,
            "pacnew": self.pacnew,
            "failed_phases": self.failed_phases,
            "hint_id": self.hint_id,
            "warnings": self.warnings,
            "error_text": self.error_text,
        }


# --------------------------------------------------------------------------
# helpers
# --------------------------------------------------------------------------


def _which(binary):
    return shutil.which(binary)


def _base_env():
    """Deterministic, non-interactive environment for every child process."""
    env = dict(os.environ)
    # C.UTF-8 keeps pacman's error strings parseable regardless of the user's
    # locale - the ERROR_HINTS matching below relies on the English wording.
    env["LC_ALL"] = "C.UTF-8"
    env["LANG"] = "C.UTF-8"
    env["TERM"] = "dumb"
    env["PYTHONUNBUFFERED"] = "1"
    # Suppresses flatpak's ANSI progress bars and OSC escape sequences, which
    # would otherwise end up as garbage in the log panel.
    env["FLATPAK_FANCY_OUTPUT"] = "0"
    return env


def _yay_env():
    """yay must not see SUDO_USER/DOAS_USER when running as root.

    With either set, yay picks /tmp/yay as its build directory and creates it
    root-owned, which the de-elevated (systemd DynamicUser) build then cannot
    write to. Unset, it uses /var/cache/yay, which systemd creates and chowns
    itself via CacheDirectory=yay.
    """
    env = _base_env()
    for key in ("SUDO_USER", "DOAS_USER", "SUDO_UID", "SUDO_GID"):
        env.pop(key, None)
    return env


def _flatpak_user_cmd(flatpak, user, home, args):
    """Build a `flatpak --user` invocation for the desktop user.

    `runuser` sets HOME/USER/LOGNAME but inherits everything else. If Decky's
    service environment happens to carry XDG_DATA_HOME, FLATPAK_USER_DIR or
    friends, flatpak would silently operate on the wrong tree - so the
    environment is rebuilt from scratch with `env -i` instead of scrubbed.
    """
    return [
        "runuser", "-u", user, "--",
        "env", "-i",
        f"HOME={home}",
        f"USER={user}",
        f"LOGNAME={user}",
        "PATH=/usr/local/bin:/usr/bin:/bin",
        "LC_ALL=C.UTF-8",
        "FLATPAK_FANCY_OUTPUT=0",
        flatpak,
    ] + args


def _is_root():
    return os.geteuid() == 0


def _pacman_busy():
    return os.path.exists(PACMAN_LOCK)


_installed_cache = None


def _installed(pkg):
    """True if a package is installed, without shelling out to pacman.

    The local database has thousands of entries, so the name set is built once
    and reused; it only changes across an update run, which invalidates it.
    """
    global _installed_cache
    if _installed_cache is None:
        local = Path("/var/lib/pacman/local")
        try:
            _installed_cache = {
                p.name.rsplit("-", 2)[0] for p in local.iterdir() if p.is_dir()
            }
        except OSError:
            _installed_cache = set()
    return pkg in _installed_cache


def _invalidate_installed_cache():
    global _installed_cache
    _installed_cache = None


# --------------------------------------------------------------------------
# plugin
# --------------------------------------------------------------------------


class Plugin:
    def __init__(self):
        self.state = State()
        self.settings = dict(DEFAULT_SETTINGS)
        self.lock = asyncio.Lock()
        self._auto_task = None
        self._dry_run = False

    # -- lifecycle ---------------------------------------------------------

    async def _main(self):
        self._load_settings()
        self._restore_state()
        decky.logger.info("CachyOS Update backend started (root=%s)", _is_root())
        if not _is_root():
            decky.logger.error(
                'Backend is NOT running as root - plugin.json is missing the '
                '"root" flag. Updates will fail.'
            )
        self._auto_task = asyncio.create_task(self._auto_check_loop())

    async def _unload(self):
        if self._auto_task:
            self._auto_task.cancel()
        self._save_state()
        decky.logger.info("CachyOS Update backend stopped")

    # -- persistence -------------------------------------------------------

    @property
    def _settings_file(self):
        return Path(decky.DECKY_PLUGIN_SETTINGS_DIR) / "settings.json"

    @property
    def _state_file(self):
        return Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "state.json"

    def _load_settings(self):
        try:
            if self._settings_file.exists():
                stored = json.loads(self._settings_file.read_text())
                for key in DEFAULT_SETTINGS:
                    if key in stored:
                        self.settings[key] = stored[key]
        except Exception as exc:
            decky.logger.warning("Could not read settings: %s", exc)

    def _save_settings(self):
        try:
            self._settings_file.parent.mkdir(parents=True, exist_ok=True)
            self._settings_file.write_text(json.dumps(self.settings, indent=2))
        except Exception as exc:
            decky.logger.warning("Could not save settings: %s", exc)

    def _restore_state(self):
        """Carry the last check result across plugin restarts."""
        try:
            if not self._state_file.exists():
                return
            stored = json.loads(self._state_file.read_text())
            self.state.counts = stored.get("counts", self.state.counts)
            self.state.details = stored.get("details", self.state.details)
            self.state.last_check = stored.get("last_check", 0.0)
            self.state.last_update = stored.get("last_update", 0.0)
        except Exception as exc:
            decky.logger.warning("Could not read stored state: %s", exc)

    def _save_state(self):
        try:
            self._state_file.parent.mkdir(parents=True, exist_ok=True)
            self._state_file.write_text(
                json.dumps(
                    {
                        "counts": self.state.counts,
                        "details": self.state.details,
                        "last_check": self.state.last_check,
                        "last_update": self.state.last_update,
                    },
                    indent=2,
                )
            )
        except Exception as exc:
            decky.logger.warning("Could not save state: %s", exc)

    # -- emitting ----------------------------------------------------------

    async def _log(self, line):
        line = line.rstrip()
        self.state.log.append(line)
        await decky.emit("cachyos_update_log", line)

    async def _set_phase(self, phase, done_weight, total_weight):
        self.state.phase = phase
        self.state.progress = done_weight / total_weight if total_weight else 0.0
        await decky.emit("cachyos_update_progress", self.state.progress, phase)

    async def _set_status(self, status):
        self.state.status = status
        await decky.emit("cachyos_update_state", status)

    # -- process runner ----------------------------------------------------

    async def _run(
        self,
        cmd,
        env=None,
        stream=True,
        phase_base=0.0,
        phase_span=0.0,
        merge_stderr=True,
    ):
        """Run a command, streaming stdout(+stderr) line by line.

        Returns (returncode, list_of_lines). `merge_stderr=False` is for probes
        whose output gets parsed - fwupdmgr in particular interleaves progress
        ticks on stderr that would otherwise corrupt the result.
        """
        if self._dry_run:
            await self._log("$ " + " ".join(cmd))
            return 0, []

        if stream:
            await self._log("$ " + " ".join(cmd))

        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=(
                    asyncio.subprocess.STDOUT
                    if merge_stderr
                    else asyncio.subprocess.DEVNULL
                ),
                stdin=asyncio.subprocess.DEVNULL,
                env=env or _base_env(),
            )
        except FileNotFoundError:
            if stream:
                await self._log(f"!! {cmd[0]} not found")
            return 127, []

        lines = []
        assert proc.stdout is not None
        while True:
            raw = await proc.stdout.readline()
            if not raw:
                break
            line = raw.decode("utf-8", "replace").rstrip("\n")
            lines.append(line)
            if stream:
                await self._log(line)
                # pacman/yay print "(12/34) upgrading foo" - use it to move the
                # bar inside a phase instead of only between phases.
                if phase_span:
                    match = PROGRESS_RE.search(line)
                    if match:
                        cur, total = int(match.group(1)), int(match.group(2))
                        if total:
                            frac = min(cur / total, 1.0)
                            self.state.progress = phase_base + phase_span * frac
                            await decky.emit(
                                "cachyos_update_progress",
                                self.state.progress,
                                self.state.phase,
                            )

        rc = await proc.wait()
        return rc, lines

    # -- checking ----------------------------------------------------------

    async def _check_pacman(self):
        if not _which("checkupdates"):
            return [], "missing_checkupdates"
        env = _base_env()
        # A stable database path: the default lives in TMPDIR and would be
        # re-downloaded on every call.
        db = Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "checkupdates-db"
        db.mkdir(parents=True, exist_ok=True)
        env["CHECKUPDATES_DB"] = str(db)
        rc, lines = await self._run(["checkupdates"], env=env, stream=False)
        # checkupdates is inverted: 0 = updates found, 2 = nothing to do.
        if rc == 2:
            return [], ""
        if rc != 0:
            return [], "checkupdates_failed"
        return [l for l in lines if l.strip()], ""

    async def _check_aur(self):
        if not self.settings["enable_aur"]:
            return [], ""
        yay = _which("yay")
        if not yay:
            return [], "missing_yay"
        rc, lines = await self._run(
            [yay, "-Qua", "--color", "never"], env=_yay_env(), stream=False
        )
        entries = [l for l in lines if l.strip() and "->" in l]
        if rc != 0 and not entries:
            return [], ""
        return entries, ""

    async def _check_flatpak(self):
        if not self.settings["enable_flatpak"]:
            return [], ""
        flatpak = _which("flatpak")
        if not flatpak:
            return [], ""
        entries = []
        # Refreshing appstream data is what makes --updates accurate.
        await self._run(
            [flatpak, "update", "--system", "--appstream", "--noninteractive"],
            stream=False,
        )
        rc, lines = await self._run(
            [
                flatpak, "remote-ls", "--system", "--updates",
                "--columns=application,version",
            ],
            stream=False,
        )
        if rc == 0:
            entries += [l for l in lines if l.strip()]
        user = getattr(decky, "DECKY_USER", None)
        home = getattr(decky, "DECKY_USER_HOME", "")
        if user and home and _which("runuser"):
            rc, lines = await self._run(
                _flatpak_user_cmd(
                    flatpak, user, home,
                    ["remote-ls", "--user", "--updates",
                     "--columns=application,version"],
                ),
                stream=False,
            )
            if rc == 0:
                entries += [l for l in lines if l.strip()]
        return entries, ""

    async def _check_fwupd(self):
        if not self.settings["enable_fwupd"]:
            return [], ""
        fwupdmgr = _which("fwupdmgr")
        if not fwupdmgr:
            return [], ""
        rc, lines = await self._run(
            [fwupdmgr, "get-updates", "--json"], stream=False, merge_stderr=False
        )
        # With --json, "nothing to do" is exit 0 plus an empty Devices array;
        # older fwupd returns 2 (EXIT_NOTHING_TO_DO). Anything else is unusable.
        if rc not in (0, 2):
            return [], ""
        try:
            data = json.loads("\n".join(lines))
        except ValueError:
            return [], ""
        entries = []
        for dev in data.get("Devices", []):
            name = dev.get("Name", "?")
            for rel in dev.get("Releases", []):
                entries.append(f"{name} -> {rel.get('Version', '?')}")
        return entries, ""

    async def check_updates(self):
        """Look for updates without installing anything."""
        if self.lock.locked():
            return self.state.to_dict()
        async with self.lock:
            await self._set_status("checking")
            warnings = []
            try:
                results = {}
                for key, coro in (
                    ("pacman", self._check_pacman()),
                    ("aur", self._check_aur()),
                    ("flatpak", self._check_flatpak()),
                    ("fwupd", self._check_fwupd()),
                ):
                    entries, warning = await coro
                    results[key] = entries
                    if warning:
                        warnings.append(warning)

                self.state.details = results
                self.state.counts = {k: len(v) for k, v in results.items()}
                self.state.last_check = time.time()
                self.state.warnings = warnings
                self.state.error_text = ""
                self._save_state()
            except Exception as exc:
                decky.logger.exception("Update check failed")
                self.state.error_text = str(exc)
            finally:
                await self._set_status("idle")
        return self.state.to_dict()

    # -- updating ----------------------------------------------------------

    def _enabled_phases(self):
        phases = ["keyring", "pacman"]
        if self.settings["enable_aur"] and _which("yay"):
            phases.append("aur")
        if self.settings["enable_flatpak"] and _which("flatpak"):
            phases.append("flatpak_system")
            if getattr(decky, "DECKY_USER", None) and _which("runuser"):
                phases.append("flatpak_user")
        if self.settings["enable_fwupd"] and _which("fwupdmgr"):
            phases.append("fwupd")
        return phases

    async def _phase_keyring(self, **kw):
        # Upgrading the keyring before the rest is the documented cure for the
        # classic "invalid or corrupted package (PGP signature)" failure. Only
        # valid as one atomic sequence with the -Su right after it, otherwise
        # it would be a partial upgrade.
        pkgs = [p for p in ("archlinux-keyring", "cachyos-keyring") if _installed(p)]
        cmd = ["pacman", "-Sy", "--noconfirm", "--color", "never"]
        if pkgs:
            cmd += ["--needed"] + pkgs
        rc, lines = await self._run(cmd, **kw)
        return rc == 0, lines

    async def _phase_pacman(self, **kw):
        rc, lines = await self._run(
            [
                "pacman", "-Su", "--noconfirm",
                "--noprogressbar", "--color", "never",
            ],
            **kw,
        )
        return rc == 0, lines

    async def _phase_aur(self, **kw):
        rc, lines = await self._run(
            [
                _which("yay") or "yay", "-Syu", "--noconfirm", "--removemake",
                "--noprogressbar", "--color", "never",
            ],
            env=_yay_env(),
            **kw,
        )
        return rc == 0, lines

    async def _phase_flatpak_system(self, **kw):
        # --system is not optional here: without an explicit scope flatpak
        # updates the system installation *and the calling user's* one, which
        # as root means it would start populating /root/.local/share/flatpak.
        # --noninteractive already implies -y and picks the quiet transaction.
        rc, lines = await self._run(
            [
                _which("flatpak") or "flatpak",
                "update", "--system", "--noninteractive",
            ],
            **kw,
        )
        return rc == 0, lines

    async def _phase_flatpak_user(self, **kw):
        rc, lines = await self._run(
            _flatpak_user_cmd(
                _which("flatpak") or "flatpak",
                decky.DECKY_USER,
                decky.DECKY_USER_HOME,
                ["update", "--user", "--noninteractive"],
            ),
            **kw,
        )
        return rc == 0, lines

    async def _phase_fwupd(self, **kw):
        fwupdmgr = _which("fwupdmgr") or "fwupdmgr"
        # No --force: it would re-download the metadata on every single run,
        # which is wasteful on a battery-powered handheld. Fresh metadata
        # simply exits 2 (EXIT_NOTHING_TO_DO), which is a success here.
        rc, lines = await self._run([fwupdmgr, "refresh", "--assume-yes"], **kw)
        if rc == 101:  # ENETUNREACH
            await self._log("!! Firmware server unreachable - skipped.")
            return True, lines
        rc, update_lines = await self._run(
            [
                fwupdmgr, "update", "--assume-yes", "--no-reboot-check",
                "--no-unreported-check", "--no-remote-check",
            ],
            **kw,
        )
        lines += update_lines
        # 0 = done, 2 = EXIT_NOTHING_TO_DO, 101 = network unreachable.
        return rc in (0, 2, 101), lines

    async def start_update(self, dry_run=False):
        if self.lock.locked():
            return {"started": False, "reason_id": "already_running"}
        if not _is_root() and not dry_run:
            return {"started": False, "reason_id": "not_root"}
        if _pacman_busy() and not dry_run:
            return {"started": False, "reason_id": "pacman_busy"}
        asyncio.create_task(self._do_update(dry_run))
        return {"started": True, "reason_id": ""}

    async def _do_update(self, dry_run=False):
        async with self.lock:
            self._dry_run = dry_run
            self.state.log.clear()
            self.state.error_text = ""
            self.state.hint_id = ""
            self.state.warnings = []
            self.state.failed_phases = []
            self.state.progress = 0.0
            await self._set_status("updating")

            phases = self._enabled_phases()
            total_weight = sum(PHASE_WEIGHTS[p] for p in phases)
            done_weight = 0
            all_output = []

            handlers = {
                "keyring": self._phase_keyring,
                "pacman": self._phase_pacman,
                "aur": self._phase_aur,
                "flatpak_system": self._phase_flatpak_system,
                "flatpak_user": self._phase_flatpak_user,
                "fwupd": self._phase_fwupd,
            }

            try:
                for phase in phases:
                    weight = PHASE_WEIGHTS[phase]
                    await self._set_phase(phase, done_weight, total_weight)
                    ok, lines = await handlers[phase](
                        phase_base=done_weight / total_weight,
                        phase_span=weight / total_weight,
                    )
                    all_output += lines
                    done_weight += weight
                    if not ok:
                        self.state.failed_phases.append(phase)
                        await self._log(f"!! Phase failed: {phase}")
                        # The repo upgrade is the foundation - if it breaks,
                        # building AUR packages on top of it is a bad idea.
                        if phase in ("keyring", "pacman"):
                            break

                self.state.progress = 1.0
                _invalidate_installed_cache()
                self.state.reboot_required = self._needs_reboot(all_output)
                if "fwupd" in phases and not self.state.reboot_required:
                    self.state.reboot_required = await self._firmware_reboot_pending()
                self.state.pacnew = await self._find_pacnew()
                self.state.last_update = time.time()

                if self.state.failed_phases:
                    self.state.status = "error"
                    self.state.hint_id = self._hint_for(all_output)
                else:
                    self.state.status = "done"
            except Exception as exc:
                decky.logger.exception("Update failed")
                self.state.status = "error"
                self.state.error_text = str(exc)
            finally:
                self._dry_run = False
                await decky.emit("cachyos_update_state", self.state.status)
                await decky.emit(
                    "cachyos_update_finished",
                    self.state.status == "done",
                    self.state.failed_phases,
                    self.state.reboot_required,
                )

            # Only a successful run means "nothing pending" - after a failure
            # the old counts stay, otherwise the panel would claim the system
            # is up to date while the updates are still sitting there.
            if not dry_run and self.state.status == "done":
                self.state.details = {k: [] for k in self.state.details}
                self.state.counts = {k: 0 for k in self.state.counts}
                self._save_state()

    # -- post-update analysis ---------------------------------------------

    def _hint_for(self, lines):
        blob = "\n".join(lines).lower()
        for marker, hint_id in ERROR_HINTS:
            if marker in blob:
                return hint_id
        return ""

    def _needs_reboot(self, output_lines):
        """True if the running kernel no longer matches what is installed.

        Primary signal is the same one cachy-update uses: the module tree of
        the running kernel still has its vmlinuz. Skipped when
        kernel-modules-hook or mkmm is installed, because those keep old module
        trees around and would make the test always pass.
        """
        try:
            if not (_installed("kernel-modules-hook") or _installed("mkmm")):
                release = os.uname().release
                if not Path(f"/usr/lib/modules/{release}/vmlinuz").is_file():
                    return True
        except Exception as exc:
            decky.logger.warning("Kernel check failed: %s", exc)

        # Cross-check: did this run upgrade anything from the reboot list?
        upgraded = {
            match.group(1)
            for line in output_lines
            for match in [UPGRADING_RE.search(line)]
            if match
        }
        return any(
            pattern.match(pkg)
            for pkg in upgraded
            for pattern in REBOOT_TRIGGER_PATTERNS
        )

    async def _firmware_reboot_pending(self):
        """Firmware is staged to the ESP and flashed on the next boot.

        `--no-reboot-check` suppresses the prompt but not the need, so ask
        fwupd explicitly. Exit 0 means a reboot IS pending, 2 means clean.
        """
        fwupdmgr = _which("fwupdmgr")
        if not fwupdmgr or self._dry_run:
            return False
        rc, _ = await self._run(
            [fwupdmgr, "check-reboot-needed"], stream=False, merge_stderr=False
        )
        return rc == 0

    async def _find_pacnew(self):
        if not _which("pacdiff"):
            return []
        env = _base_env()
        env["DIFFPROG"] = "/bin/true"
        rc, lines = await self._run(["pacdiff", "--output"], env=env, stream=False)
        if rc != 0:
            return []
        return [l.strip() for l in lines if l.strip().startswith("/")]

    # -- frontend API ------------------------------------------------------

    async def get_state(self):
        return self.state.to_dict()

    async def get_settings(self):
        return self.settings

    async def set_settings(self, settings):
        for key in DEFAULT_SETTINGS:
            if key in settings:
                self.settings[key] = settings[key]
        self._save_settings()
        return self.settings

    async def reboot(self):
        if self._dry_run:
            return True
        await self._run(["systemctl", "reboot"], stream=False)
        return True

    async def self_test(self):
        tools = {}
        # (binary, version args, substrings that identify the relevant line)
        for name, args, match in (
            ("pacman", ["-Q", "pacman"], ()),
            ("yay", ["--version"], ()),
            ("paru", ["--version"], ()),
            ("checkupdates", None, ()),
            ("pacdiff", None, ()),
            ("flatpak", ["--version"], ()),
            # fwupdmgr prints a compile/runtime table, not a single line.
            ("fwupdmgr", ["--version"], ("runtime", "org.freedesktop.fwupd")),
            ("runuser", None, ()),
            ("systemd-run", ["--version"], ()),
            ("fakeroot", None, ()),
        ):
            path = _which(name)
            entry = {"found": bool(path), "path": path or "", "version": ""}
            if path and args:
                try:
                    rc, lines = await self._run(
                        [path] + args, stream=False, merge_stderr=False
                    )
                    if rc == 0:
                        for line in lines:
                            text = " ".join(line.split())
                            if not text:
                                continue
                            if match:
                                if all(m in text for m in match):
                                    entry["version"] = text[:80]
                                    break
                            elif any(c.isdigit() for c in text):
                                entry["version"] = text[:80]
                                break
                except Exception:
                    pass
            tools[name] = entry

        return {
            "root": _is_root(),
            "uid": os.geteuid(),
            "standalone": STANDALONE,
            "decky_user": getattr(decky, "DECKY_USER", ""),
            "decky_user_home": getattr(decky, "DECKY_USER_HOME", ""),
            "kernel": os.uname().release,
            "kernel_vmlinuz_present": Path(
                f"/usr/lib/modules/{os.uname().release}/vmlinuz"
            ).is_file(),
            "pacman_busy": _pacman_busy(),
            "tools": tools,
            "settings": self.settings,
        }

    # -- background check --------------------------------------------------

    async def _auto_check_loop(self):
        # Let the session settle before hitting the network.
        await asyncio.sleep(30)
        while True:
            try:
                if self.settings["auto_check"]:
                    await self.check_updates()
                    total = sum(self.state.counts.values())
                    if total and self.settings["notify_on_updates"]:
                        await decky.emit("cachyos_update_available", total)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                decky.logger.warning("Automatic check failed: %s", exc)
            hours = max(1, int(self.settings.get("check_interval_hours", 6)))
            await asyncio.sleep(hours * 3600)


# --------------------------------------------------------------------------
# standalone entry point (development only)
# --------------------------------------------------------------------------


async def _cli():
    plugin = Plugin()
    plugin._load_settings()
    args = sys.argv[1:]

    if "--selftest" in args:
        result = await plugin.self_test()
        print(json.dumps(result, indent=2, ensure_ascii=False))
        if not result["root"]:
            print("\nNote: not running as root - expected during development.")
        return

    if "--check" in args:
        state = await plugin.check_updates()
        print(json.dumps(
            {
                "counts": state["counts"],
                "total": state["total"],
                "warnings": state["warnings"],
                "details": state["details"],
            },
            indent=2,
            ensure_ascii=False,
        ))
        return

    if "--update" in args:
        dry = "--dry-run" in args
        if not dry and not _is_root():
            print("A real update needs root. Use --dry-run to test.")
            return
        if dry:
            # Enable every phase so the dry run shows all command lines.
            plugin.settings["enable_fwupd"] = True
        await plugin._do_update(dry_run=dry)
        print(f"\nStatus: {plugin.state.status}")
        if plugin.state.failed_phases:
            print(f"Failed phases: {', '.join(plugin.state.failed_phases)}")
        if plugin.state.error_text:
            print(f"Error: {plugin.state.error_text}")
        return

    print(__doc__)


if __name__ == "__main__":
    asyncio.run(_cli())