"""Command line entry point.
The shape of a run:
gather repo updates, AUR updates, hygiene, flatpak, vulnerabilities -- all read-only
review one screen, one confirmation
apply repo first, then AUR, then flatpak, then report
Two rules the order encodes:
* `pacman -Sy` refreshes, then the plan is computed, then `-Su` applies it with
no second refresh. What the user confirms is computed from the very databases
the transaction will use, so the plan cannot be a lie.
* A failed repo phase stops the run before any AUR build. Building against a
half-upgraded system is how a tree gets broken.
"""
from __future__ import annotations
import argparse
import contextlib
import json
import os
import re
import shutil
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
import pyalpm
from rich.console import Console
from . import (
__version__,
aur,
changes,
config,
detail,
fetch,
flatpak,
hygiene,
icons,
live,
news,
repo,
security,
stream,
)
from . import (
hooks as hooks_mod,
)
from . import (
queue as queue_mod,
)
from .audit import Report, Severity, audit
from .buildgraph import MissingMetadata, build_graph
from .render import (
AurRow,
CheckRow,
RepoRow,
Review,
Step,
VcsRow,
audit_screen,
human_age,
human_bytes,
review_screen,
)
from .srcinfo import load, parse_pkgbuild
from .theme import Theme
CLONE_DEFAULT = Path.home() / ".cache/paru/clone"
VCS_SUFFIXES = ("-git", "-svn", "-hg", "-cvs", "-bzr", "-darcs", "-nightly")
EXIT_OK = 0
EXIT_FAILED = 1
EXIT_DECLINED = 2
EXIT_BLOCKED = 3
# --------------------------------------------------------------------------
# options
# --------------------------------------------------------------------------
@dataclass
class Options:
command: str = "run"
dry_run: bool = False
assume_yes: bool = False
no_tui: bool = False
offline: bool = False
json: bool = False
tail: bool = False
workers: int = 1
icon_tier: str | None = None
clone_dir: Path = CLONE_DEFAULT
width: int = 100
repo_phase: bool = True
aur_phase: bool = True
flatpak_phase: bool = True
hygiene_phase: bool = True
security_phase: bool = True
news_phase: bool = True
refresh: bool = True
#: AUR packages to leave out, and repo packages to hold back
skips: list[str] = field(default_factory=list)
holds: list[str] = field(default_factory=list)
json_stream: bool = False
def _shared_flags(parser: argparse.ArgumentParser, *, suppress: bool = False) -> None:
"""Flags accepted both before and after the subcommand.
With `suppress` the subparser copy carries no default, so a value given
before the subcommand survives instead of being overwritten by a default.
"""
def add(*flags, **kwargs):
if suppress:
kwargs["default"] = argparse.SUPPRESS
parser.add_argument(*flags, **kwargs)
add("-n", "--dry-run", action="store_true", help="show the plan and stop")
add("-y", "--yes", action="store_true", help="do not ask for confirmation")
add("--no-tui", action="store_true", help="stream plain lines instead of the live dashboard")
add("--offline", action="store_true", help="skip everything that needs the network")
add("--json", action="store_true", help="machine-readable output")
add(
"--icons",
choices=sorted(icons.TIERS),
default=None,
help="force an icon tier instead of detecting one",
)
add("--clonedir", type=Path, default=CLONE_DEFAULT, help="paru's clone directory")
add("--no-repo", action="store_true", help="skip repo upgrades")
add("--no-aur", action="store_true", help="skip AUR updates")
add("--no-flatpak", action="store_true", help="skip flatpak")
add("--no-hygiene", action="store_true", help="skip hygiene checks")
add("--no-security", action="store_true", help="skip the arch-audit vulnerability scan")
add("--no-news", action="store_true", help="skip the arch news feed")
add(
"--no-refresh",
action="store_true",
help="plan from the databases already on disk instead of syncing first",
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="updater",
description="A pretty Arch Linux update orchestrator.",
epilog=(
"exit codes: 0 ok or nothing to do, 1 a phase failed, "
"2 you declined the plan, 3 flagged packages need a decision"
),
)
parser.add_argument("--version", action="version", version=f"updater {__version__}")
_shared_flags(parser)
sub = parser.add_subparsers(dest="command")
sub.add_parser("plan", parents=[], help="show the plan (implies --dry-run)")
sub.add_parser("hygiene", help="cache, config drift and stale processes")
audit_cmd = sub.add_parser("audit", help="audit one AUR package's pending changes")
audit_cmd.add_argument("package")
advisory_cmd = sub.add_parser(
"advisory", help="one security advisory, fetched from the tracker"
)
advisory_cmd.add_argument("advisory_id")
sub.add_parser("security", help="re-run arch-audit and report the result")
show_cmd = sub.add_parser("show", help="everything known about one package (diff, audit, CVEs)")
show_cmd.add_argument("package")
apply_cmd = sub.add_parser("apply", help="run the update")
apply_cmd.add_argument(
"--skip",
action="append",
default=[],
metavar="PKG",
help="leave this AUR package out of the run",
)
apply_cmd.add_argument(
"--hold",
action="append",
default=[],
metavar="PKG",
help="hold this repo package back (a partial upgrade)",
)
apply_cmd.add_argument(
"--json-stream",
action="store_true",
help="emit build events as JSON lines instead of rendering",
)
sub.add_parser("checks", help="re-run only the local checks (after running a fix)")
log_cmd = sub.add_parser("log", help="where the last run's logs are")
log_cmd.add_argument("--tail", action="store_true", help="print the captured logs")
# let the shared flags also appear after the subcommand
for subparser in sub.choices.values():
_shared_flags(subparser, suppress=True)
return parser
def options_from(args: argparse.Namespace) -> Options:
def flag(name: str, default=False):
return getattr(args, name, default)
return Options(
command=args.command or "run",
dry_run=flag("dry_run") or args.command == "plan",
assume_yes=flag("yes"),
no_tui=flag("no_tui"),
offline=flag("offline"),
json=flag("json"),
tail=flag("tail"),
# v1 drives paru, which builds serially. There is deliberately no
# --workers flag: exposing one would let a user halve their build speed
# by dividing cores across workers that never run concurrently. The
# division logic in live.build_env is kept for the parallel scheduler.
workers=1,
icon_tier=getattr(args, "icons", None),
clone_dir=getattr(args, "clonedir", CLONE_DEFAULT),
width=shutil.get_terminal_size((100, 40)).columns,
repo_phase=not flag("no_repo"),
aur_phase=not flag("no_aur"),
flatpak_phase=not flag("no_flatpak"),
hygiene_phase=not flag("no_hygiene"),
security_phase=not flag("no_security"),
news_phase=not flag("no_news"),
refresh=not flag("no_refresh"),
skips=list(getattr(args, "skip", []) or []),
holds=list(getattr(args, "hold", []) or []),
json_stream=flag("json_stream"),
)
# --------------------------------------------------------------------------
# run state
# --------------------------------------------------------------------------
def state_root() -> Path:
base = os.environ.get("XDG_STATE_HOME") or (Path.home() / ".local/state")
return Path(base) / "updater"
def new_run_dir() -> Path:
return state_root() / "runs" / time.strftime("%Y-%m-%dT%H-%M-%S")
def latest_link() -> Path:
return state_root() / "latest"
@dataclass
class Run:
options: Options
theme: Theme = field(default_factory=Theme.default)
console: Console = field(default_factory=Console)
glyphs: object = None
warnings: list[str] = field(default_factory=list)
audits: dict[str, Report] = field(default_factory=dict)
aur_meta: dict[str, aur.AurPackage] = field(default_factory=dict)
#: genuine AUR upgrades: name -> (installed, aur)
updates: dict[str, tuple[str, str]] = field(default_factory=dict)
#: AUR version is *older* than what is installed; `paru -Sua` would not apply
#: these, so reporting them as updates would be wrong
downgrades: dict[str, tuple[str, str]] = field(default_factory=dict)
plan: repo.Plan = field(default_factory=repo.Plan)
hygiene: hygiene.Hygiene = field(default_factory=hygiene.Hygiene)
flatpaks: flatpak.FlatpakState = field(default_factory=flatpak.FlatpakState)
vulns: security.Vulnerabilities = field(default_factory=security.Vulnerabilities)
#: devel packages paru would rebuild, and why not if it could not say
devel_upgrades: dict[str, tuple[str, str]] = field(default_factory=dict)
devel_note: str = ""
foreign: list[str] = field(default_factory=list)
#: devel packages paru would rebuild, for the plan breakdown
vcs: list = field(default_factory=list)
#: structured warnings, so the row reads as a headline and the overlay
#: carries the explanation
warning_notes: list = field(default_factory=list)
config: config.Config = field(default_factory=config.Config)
#: whether the plan came from freshly synced databases
refreshed: bool = False
refresh_note: str = ""
log_dir: Path = field(default_factory=new_run_dir)
@property
def blocked(self) -> list[str]:
return [n for n, report in self.audits.items() if report.verdict is Severity.BLOCK]
# --------------------------------------------------------------------------
# gathering
# --------------------------------------------------------------------------
@contextlib.contextmanager
def _plan_handle(run: Run):
"""A handle for planning.
Preferred: a throwaway database synced under fakeroot, which makes the plan
accurate without asking for a password. Falling back to the databases on
disk is allowed but recorded, because a plan built from stale data is the
one thing this tool must not present as fact.
"""
opts = run.options
if opts.offline or not opts.refresh:
with repo.connect() as handle:
run.refreshed = False
run.refresh_note = (
"offline: databases not refreshed" if opts.offline else "refresh skipped"
)
yield handle
return
with repo.preview() as (handle, refreshed, note):
run.refreshed = refreshed
run.refresh_note = note
yield handle
def gather(run: Run) -> Review:
"""Everything the review screen needs. Read-only, never mutates the system."""
opts = run.options
review = Review(meta=f"arch linux · rolling · {time.strftime('%Y-%m-%d %H:%M')}")
with _plan_handle(run) as handle:
installed = repo.installed(handle)
foreign = repo.foreign(handle)
orphan_names = repo.orphans(handle)
if opts.repo_phase:
run.plan = repo.plan(handle)
review.repo = [
RepoRow("U", u.name, u.old, u.new, human_bytes(u.delta)) for u in run.plan.updates
]
review.repo_total = len(run.plan.updates)
review.repo_download = run.plan.download_total
review.repo_delta = run.plan.installed_delta
if opts.aur_phase and not opts.offline and foreign:
try:
run.aur_meta = aur.fetch(foreign)
except aur.AurError as exc:
run.warnings.append(str(exc))
run.aur_meta = {}
gone = aur.deleted(foreign, run.aur_meta, known_bases=set(installed))
if gone:
run.warning_notes.append(_package_warning(gone, known=set(installed)))
run.warnings.append(f"{len(gone)} package(s) no longer in the AUR")
run.updates, run.downgrades = _classify_aur(installed, run.aur_meta)
review.aur = _aur_rows(run)
review.aur_total = len(review.aur)
# a downgrade is listed but paru will not apply it, so it is not "to build"
review.aur_build = sum(1 for r in review.aur if r.tag not in ("!!", "D"))
if opts.flatpak_phase:
run.flatpaks = flatpak.collect(include_runtimes=False)
if opts.security_phase and not opts.offline:
run.vulns = security.scan()
if opts.hygiene_phase:
run.hygiene = hygiene.collect(
installed=installed,
orphans=orphan_names,
updated={u.name for u in run.plan.updates},
include_processes=True,
)
# the devel check informs a check row, so it has to run before _checks
run.foreign = list(foreign)
if opts.aur_phase and not opts.offline and foreign:
run.devel_upgrades, run.devel_note = aur.devel_upgrades()
review.advisories = [
(a.id, a.severity, a.fixed or "", ", ".join(a.packages), len(a.issues))
for a in run.vulns.sorted_advisories()
]
review.checks = _checks(run)
review.steps = _steps(run)
review.vcs = _vcs(installed, foreign, run.devel_upgrades)
run.vcs = review.vcs
return review
def _is_vcs(name: str) -> bool:
return name.endswith(VCS_SUFFIXES)
def _classify_aur(
installed: dict[str, str], meta: dict[str, aur.AurPackage]
) -> tuple[dict[str, tuple[str, str]], dict[str, tuple[str, str]]]:
"""Split AUR packages into real upgrades and things that only look like them.
Two traps, both of which produce a plausible-looking list of updates that
`paru -Sua` would not actually perform:
VCS packages the AUR records the pkgver at the last commit, which for a
-git package is routinely *older* than the build sitting on
this machine. Version comparison is meaningless for them;
upstream movement is what matters, and that is reported
separately rather than rebuilt automatically.
downgrades an AUR version lower than the installed one is not an
upgrade, and pacman will not apply it.
"""
upgrades: dict[str, tuple[str, str]] = {}
downgrades: dict[str, tuple[str, str]] = {}
for name, pkg in meta.items():
old = installed.get(name)
if not old or not pkg.version or _is_vcs(name):
continue
comparison = pyalpm.vercmp(pkg.version, old)
if comparison > 0:
upgrades[name] = (old, pkg.version)
elif comparison < 0:
downgrades[name] = (old, pkg.version)
return upgrades, downgrades
def _aur_rows(run: Run) -> list[AurRow]:
from .render.review import aur_row_from
state = aur.State.load()
rows: list[AurRow] = []
listed = {**run.updates, **run.downgrades}
for name, (old, new) in sorted(listed.items()):
report = run.audits.get(name)
if report is None:
meta = run.aur_meta.get(name)
# Falls back to fetching the PKGBUILD when paru has not cloned the
# package, so "not audited" means genuinely unavailable rather than
# "nobody looked".
change = changes.change_for_audit(
run.options.clone_dir,
name,
network=not run.options.offline,
old_commit=state.previous_commit(name),
old_maintainer=state.previous_maintainer(name),
new_maintainer=meta.maintainer if meta else None,
)
report = audit(change)
run.audits[name] = report
row = aur_row_from(name, old, new, report)
if name in run.downgrades:
row.tag = "D"
row.note = "AUR is older"
rows.append(row)
return rows
def _vcs(
installed: dict[str, str],
foreign: list[str] | None = None,
upgrades: dict[str, tuple[str, str]] | None = None,
) -> list[VcsRow]:
"""Only the devel packages that actually need rebuilding.
Listing all of them was noise: 21 rows, of which the useful number is
usually zero. A `-git` package's recorded pkgver is whatever the maintainer
last committed and is routinely *behind* the version paru generated when it
built the thing, so a version comparison says nothing about whether there is
anything new. paru's own development database is the authority, and
`devel_upgrades()` asks it.
Restricted to the foreign set: a VCS-suffixed name in a repository is not an
AUR package, and the apply path filters on `foreign`, so filtering on
anything else here would let the list and the command disagree.
"""
if upgrades is None:
return []
candidates = set(foreign if foreign is not None else installed)
# `paru -Qua` lists every AUR upgrade, not only the devel ones, so this has
# to filter to VCS names itself. Without that, each ordinary AUR update
# appeared a second time as a "devel rebuild".
return [
VcsRow(name, *upgrades[name])
for name in sorted(upgrades)
if name in candidates and _is_vcs(name)
]
@dataclass
class WarningNote:
"""One kind of warning.
A warning is a category, not a line of text: several packages can have left
the AUR while something else entirely is also worth saying. Each kind gets a
tab of its own rather than being stacked into one scroll, so the second one
is not buried under the first.
"""
label: str # the row it belongs to
title: str # the tab
headline: str # one line for the row
details: list[str] = field(default_factory=list)
action: list[str] = field(default_factory=list)
flagged: int = 0 # how many things want attention
#: the packages this warning is about, so the front end can scope it to a
#: selection. Free text cannot be filtered; a list can.
packages: list[str] = field(default_factory=list)
def _package_warning(gone: list[str], *, known: set[str]) -> WarningNote:
"""Packages that vanished from the AUR, which is usually not an emergency."""
shown = ", ".join(sorted(gone)[:3])
details = [
f"{len(gone)} installed package(s) are no longer listed in the AUR:",
"",
*[f" {name}" for name in sorted(gone)[:20]],
*([f" + {len(gone) - 20} more"] if len(gone) > 20 else []),
"",
"This usually means the maintainer deleted them, or they were merged",
"into another package. Nothing is removed from your system by this:",
"the installed copies stay exactly as they are.",
"",
"What it does mean is that they will stop receiving updates, and that",
"a dependency they satisfy has nothing to fall back on. Worth knowing",
"before they rot:",
"",
" pacman -Qi <name> what it is and what needs it",
]
return WarningNote(
label="warning",
title="left the AUR",
headline=f"{len(gone)} package(s) left the AUR · {shown}",
details=details,
flagged=len(gone),
packages=sorted(gone),
)
def _looks_failed(note: str) -> bool:
"""Whether a refresh note is an error rather than a plain explanation."""
lowered = note.lower()
return any(
word in lowered for word in ("error", "failed", "could not", "unable", "refused", "denied")
)
def _wrap_note(note: str, width: int = 68) -> list[str]:
"""Wrap a tool's output so it reads in a modal rather than running off."""
import textwrap
lines: list[str] = []
for raw in note.splitlines() or [note]:
lines.extend(textwrap.wrap(raw.strip(), width) or [""])
return lines[:12]
def _sync_metadata(handle, package: str) -> dict | None:
"""Everything the sync databases know about a package."""
for db in handle.get_syncdbs():
pkg = db.get_pkg(package)
if pkg is None:
continue
return {
"repo": db.name,
"version": pkg.version,
"desc": pkg.desc or "",
"packager": pkg.packager or "",
"builddate": pkg.builddate or 0,
"url": pkg.url or "",
"licenses": list(pkg.licenses or []),
"depends": list(pkg.depends or []),
"optdepends": list(pkg.optdepends or []),
"groups": list(pkg.groups or []),
"conflicts": list(pkg.conflicts or []),
"provides": list(pkg.provides or []),
"replaces": list(pkg.replaces or []),
"download_size": pkg.download_size or 0,
"isize": pkg.isize or 0,
"base": pkg.base or pkg.name,
"sha256sum": pkg.sha256sum or "",
}
return None
def _dependents(handle, package: str) -> list[str]:
"""Installed packages that declare a dependency on this one."""
out: list[str] = []
for pkg in handle.get_localdb().pkgcache:
for dep in pkg.depends or ():
name = dep.split(">=")[0].split("=")[0].split("<")[0].strip()
if name == package:
out.append(pkg.name)
break
return sorted(out)
def _hooks_for(package: str):
"""The alpm hooks that fire when this one package is upgraded."""
try:
paths = _installed_paths({package})
before, after = hooks_mod.scheduled(hooks_mod.load(), {package}, paths)
return before + after
except Exception: # noqa: BLE001 - attribution is a bonus
return []
def _emit_repo_detail(run: Run, payload: dict) -> None:
console = run.console
name = payload["package"]
head = name
if (
payload["installed"]
and payload["available"]
and payload["installed"] != payload["available"]
):
head += f" {payload['installed']} -> {payload['available']}"
elif payload["available"]:
head += f" {payload['available']}"
console.print(f"[bold]{head}[/bold]")
for section in payload["sections"]:
console.print()
console.print(f"[bold]{section['title']}[/bold]")
for label, value in section.get("fields", []):
console.print(f" [dim]{label:<12}[/dim] {value}")
for item in section.get("items", []):
console.print(f" {item}")
for note in payload.get("notes", []):
console.print()
console.print(f"[dim]{note}[/dim]")
def _hygiene_checks(run: Run) -> list[CheckRow]:
"""The checks that describe this machine, not the wider world.
Split out because these are the only ones an action can change. Refreshing
after running one therefore needs nothing from the network, and asking for
the whole plan again would re-poll the AUR to learn nothing new.
"""
rows: list[CheckRow] = []
# An empty devel list means "nothing to rebuild" only if the question could
# be asked. Say so when it could not, or the section looks deceptively clean.
if run.devel_note and any(_is_vcs(n) for n in (run.foreign or ())):
rows.append(
CheckRow(
False,
"devel",
"could not check for devel rebuilds",
details=[
"Some installed packages track a branch rather than a release.",
"paru decides whether those need rebuilding, and it could not be",
"asked this time:",
"",
f" {run.devel_note}",
"",
"Nothing is wrong with the system. It means this plan cannot say",
"whether a -git package has moved upstream, so none are offered.",
],
)
)
leftovers = run.hygiene.leftovers
if leftovers.worth_mentioning:
age = human_age(time.time() - leftovers.oldest) if leftovers.oldest else "?"
cache = getattr(run.options, "cache_dir", None) or "/var/cache/pacman/pkg"
rows.append(
CheckRow(
False,
"downloads",
f"{leftovers.count} incomplete download(s), oldest {age} old",
details=leftovers.explain(cache),
action=[
"sudo",
"find",
str(cache),
"-maxdepth",
"1",
"-name",
"download-*",
"-delete",
],
)
)
pacnew = run.hygiene.drift.pacnew
if pacnew:
rows.append(
CheckRow(
False,
"pacnew",
f"{len(pacnew)} unresolved config file(s)",
details=[
"pacman left these behind rather than overwrite a config file",
"you or a package had changed. Your version is still in place;",
"the package's version sits beside it:",
"",
*[f" {p}" for p in pacnew[:16]],
"",
"pacdiff walks through them and shows the difference, which is",
"the only sensible way to decide. It runs interactively.",
],
action=["sudo", "pacdiff"],
)
)
else:
rows.append(
CheckRow(
True,
"pacnew",
"no unresolved config files",
details=["pacman had no config file it needed to", "leave alone on this system"],
)
)
cache = run.hygiene.cache
if cache.has_reclaimable:
keep = cache.keep
rows.append(
CheckRow(
False,
"cache",
f"{human_bytes(cache.reclaimable)} reclaimable · paccache -rk{keep}",
details=[
(
f"{human_bytes(cache.reclaimable)} across "
f"{cache.reclaimable_files} file(s) can go"
),
f"the newest {keep} version(s) of each package are kept",
"",
"this deletes cached packages. They can be downloaded again,",
"so it is safe, but the next install of an old version will",
"fetch it from a mirror instead of the cache.",
],
action=["paccache", f"-rk{keep}"],
)
)
else:
rows.append(CheckRow(True, "cache", "nothing worth reclaiming"))
if run.hygiene.orphans:
orphan_list = sorted(run.hygiene.orphans)
rows.append(
CheckRow(
False,
"orphans",
f"{len(orphan_list)} installed as dependencies and no longer needed",
details=[
f"{len(orphan_list)} package(s) were pulled in as dependencies",
"and nothing depends on them now:",
"",
*[f" {name}" for name in orphan_list[:24]],
"",
"Removing them is the usual advice, but a package can be",
"uninstalled because nothing *declares* a dependency on it and",
"still be something you use. Read the list first.",
],
action=["sudo", "pacman", "-Rns", "--noconfirm", *orphan_list],
warning="removes packages from the system",
)
)
else:
rows.append(CheckRow(True, "orphans", "no orphans"))
return rows
def _checks(run: Run) -> list[CheckRow]:
opts = run.options
rows: list[CheckRow] = []
if opts.offline:
rows.append(CheckRow(True, "network", "offline: only local checks ran"))
age = repo.sync_db_age()
if run.refreshed:
rows.append(CheckRow(True, "databases", "refreshed from mirrors"))
elif age is None:
rows.append(CheckRow(True, "databases", run.refresh_note or "age unknown"))
else:
failed = bool(run.refresh_note) and _looks_failed(run.refresh_note)
if failed:
# pacman's own words are long and shouty. A row is not the place for
# them; the overlay is, where there is room to explain.
rows.append(
CheckRow(
False,
"databases",
f"{human_age(age)} old · refresh failed",
details=[
"pacman could not refresh its databases, so this plan was",
"built from the copies already on disk.",
"",
"what pacman said:",
*[f" {line}" for line in _wrap_note(run.refresh_note)],
"",
"A refresh that fails usually means mirrors are unreachable",
"or pacman cannot write to its scratch directory. The plan",
"below is still valid for the databases you have; it may",
"simply be missing newer versions.",
],
)
)
else:
stale = age > 86400
detail = f"{human_age(age)} old"
if run.refresh_note:
detail += f" · {run.refresh_note}"
rows.append(CheckRow(not stale, "databases", detail))
if opts.news_phase and not opts.offline:
items, news_error = news.fetch_news()
if news_error:
rows.append(CheckRow(False, "news", news_error))
else:
fresh = news.recent(items, since=run.config.news_seen)
rows.append(
CheckRow(
not any(i.needs_action for i in fresh),
"news",
news.summary_line(fresh),
details=news.explain(fresh),
)
)
# An empty devel list means "nothing to rebuild" only if the question could
# be asked. Say so when it could not, or the section looks deceptively clean.
if run.devel_note and any(_is_vcs(n) for n in (run.foreign or ())):
rows.append(
CheckRow(
False,
"devel",
"could not check for devel rebuilds",
details=[
"Some installed packages track a branch rather than a release.",
"paru decides whether those need rebuilding, and it could not be",
"asked this time:",
"",
f" {run.devel_note}",
"",
"Nothing is wrong with the system. It means this plan cannot say",
"whether a -git package has moved upstream, so none are offered.",
],
)
)
leftovers = run.hygiene.leftovers
if leftovers.worth_mentioning:
age = human_age(time.time() - leftovers.oldest) if leftovers.oldest else "?"
cache = getattr(run.options, "cache_dir", None) or "/var/cache/pacman/pkg"
rows.append(
CheckRow(
False,
"downloads",
f"{leftovers.count} incomplete download(s), oldest {age} old",
details=leftovers.explain(cache),
action=[
"sudo",
"find",
str(cache),
"-maxdepth",
"1",
"-name",
"download-*",
"-delete",
],
)
)
# advisories live in the Security section, which both front ends render.
# Repeating the counts here as well was duplication: two rows saying the same
# thing, one of which opened a view of the other.
pacnew = run.hygiene.drift.pacnew
if pacnew:
rows.append(
CheckRow(
False,
"pacnew",
f"{len(pacnew)} unresolved config file(s)",
details=[
"pacman left these behind rather than overwrite a config file",
"you or a package had changed. Your version is still in place;",
"the package's version sits beside it:",
"",
*[f" {p}" for p in pacnew[:16]],
"",
"pacdiff walks through them and shows the difference, which is",
"the only sensible way to decide. It runs interactively.",
],
action=["sudo", "pacdiff"],
)
)
else:
rows.append(
CheckRow(
True,
"pacnew",
"no unresolved config files",
details=["pacman had no config file it needed to", "leave alone on this system"],
)
)
cache = run.hygiene.cache
if cache.has_reclaimable:
keep = cache.keep
rows.append(
CheckRow(
False,
"cache",
f"{human_bytes(cache.reclaimable)} reclaimable · paccache -rk{keep}",
details=[
(
f"{human_bytes(cache.reclaimable)} across "
f"{cache.reclaimable_files} file(s) can go"
),
f"the newest {keep} version(s) of each package are kept",
"",
"this deletes cached packages. They can be downloaded again,",
"so it is safe, but the next install of an old version will",
"fetch it from a mirror instead of the cache.",
],
action=["paccache", f"-rk{keep}"],
)
)
if run.hygiene.orphans:
orphan_list = sorted(run.hygiene.orphans)
shown = ", ".join(orphan_list[:6])
if len(orphan_list) > 6:
shown += f", +{len(orphan_list) - 6} more"
rows.append(
CheckRow(
False,
"orphans",
f"{len(orphan_list)} installed as dependencies and no longer needed",
details=[
f"{len(orphan_list)} package(s) were pulled in as dependencies",
"and nothing depends on them now:",
"",
*[f" {name}" for name in orphan_list[:24]],
"",
"Removing them is the usual advice, but a package can be",
"uninstalled because nothing *declares* a dependency on it and",
"still be something you use. Read the list first.",
],
action=["sudo", "pacman", "-Rns", "--noconfirm", *orphan_list],
warning="removes packages from the system",
)
)
if opts.flatpak_phase:
if run.flatpaks.reason:
# not OK: the glyph has to match the words. A failed repository
# refresh is flagged the same way, and a tick beside "unreachable"
# reads as a contradiction.
rows.append(CheckRow(False, "flatpak", run.flatpaks.short_reason))
elif run.flatpaks.has_updates:
rows.append(CheckRow(True, "flatpak", f"{len(run.flatpaks.updates)} update(s)"))
else:
rows.append(CheckRow(True, "flatpak", "nothing to do"))
if run.hygiene.reboot:
rows.append(
CheckRow(
False,
"reboot",
run.hygiene.reboot,
details=[
"A new kernel or core library is installed but not yet in use.",
"Nothing is broken; it takes effect on the next boot.",
],
)
)
# One row for warnings, whatever kinds there are: the page behind it tabs
# through them. A row each turned a single "look at this" into a list of
# near-identical rows.
kinds = list(run.warning_notes)
replaced = {note.label for note in run.warning_notes}
for warning in run.warnings:
if "warning" in replaced and warning.endswith("no longer in the AUR"):
continue
if any(note.headline == warning for note in run.warning_notes):
continue
kinds.append(
WarningNote(
label="warning", title="general", headline=warning, details=_wrap_note(warning)
)
)
if kinds:
headline = kinds[0].headline
if len(kinds) > 1:
headline = f"{len(kinds)} kinds of warning · " + headline
rows.append(
CheckRow(False, "warning", headline, details=kinds[0].details, action=kinds[0].action)
)
return rows
def _stage_dicts(stages) -> list[dict]:
return [
{
"command": s.command,
"what": s.what,
"target": s.target,
"level": s.level,
"phase": s.phase,
"finding": s.finding,
}
for s in stages
]
def _installed_paths(names: set[str]) -> dict[str, list[str]]:
"""File lists for the installed versions, for matching path-triggered hooks.
An approximation, and the only one available: the new version's file list is
inside a package that has not been downloaded.
"""
try:
with repo.connect() as handle:
local = {p.name: p for p in handle.get_localdb().pkgcache}
return {
name: [entry[0] for entry in (getattr(local[name], "files", []) or [])]
for name in names
if name in local
}
except Exception: # noqa: BLE001 - hook attribution is a bonus, not a need
return {}
def _steps(run: Run) -> list[Step]:
steps: list[Step] = []
index = 1
all_hooks = hooks_mod.load()
def add(command: str, detail: str, *, details: list[str] | None = None, staged=None) -> None:
nonlocal index
steps.append(
Step(
str(index),
command,
detail,
details or [],
_stage_dicts(staged.stages) if staged else [],
)
)
index += 1
if run.plan.updates:
updates = sorted(run.plan.updates, key=lambda u: (u.repo, u.name))
staged = queue_mod.pacman_queue(
updates,
repo_of=lambda n: next((u.repo for u in updates if u.name == n), ""),
refreshing=run.options.refresh,
all_hooks=all_hooks,
paths_by_package=_installed_paths({u.name for u in updates}),
)
add(
"pacman -Syu",
f"{len(updates)} packages · {human_bytes(run.plan.download_total)} download",
staged=staged,
details=[
"sudo pacman -Syu --noconfirm",
"",
"pacman will upgrade these, and anything they need:",
"",
*[f" {u.name} {u.old} -> {u.new}" for u in updates[:24]],
*([f" + {len(updates) - 24} more"] if len(updates) > 24 else []),
"",
(
f"{human_bytes(run.plan.download_total)} to download, "
f"{human_bytes(abs(run.plan.installed_delta))} "
f"{'freed' if run.plan.installed_delta < 0 else 'added'} on disk"
),
"",
"pacman downloads to /var/cache/pacman/pkg, verifies the",
"signatures against its keyring, checks each package against",
"the version in the database, and only then unpacks it.",
],
)
if run.updates:
targets = sorted(run.updates)
lines = [
"paru -S --needed --noconfirm --nobatchinstall <name>"
+ (" ..." if len(targets) > 6 else ""),
"",
f"{len(targets)} AUR package(s) will be built, each by makepkg.",
"paru is invoked once per package, in dependency order, so one",
"that cannot be resolved costs one package instead of all of them:",
"",
*[
f" {name} {run.updates[name][0]} -> {run.updates[name][1]}"
for name in targets[:20]
],
*([f" + {len(targets) - 20} more"] if len(targets) > 20 else []),
"",
"makepkg, for every one of them:",
" 1 clones the AUR repository and reads its PKGBUILD",
" 2 downloads the declared sources and checks their sums",
" 3 runs prepare(), then build(), then check()",
" 4 runs package() and packs the result as a .pkg.tar.zst",
" 5 installs that package with pacman -U",
"",
"A PKGBUILD is shell and those functions run as you. That is what",
"the audit findings above are for: read them before agreeing.",
]
if run.blocked:
lines += [
"",
(
f"{len(run.blocked)} package(s) the audit flagged will be "
"skipped rather than built."
),
]
staged = queue_mod.aur_queue(
run.options.clone_dir,
targets,
meta=run.aur_meta,
audits=run.audits,
all_hooks=all_hooks,
paths_by_package=_installed_paths(set(targets)),
)
add(
"paru -S --needed (per package)",
f"{len(run.updates)} AUR packages to build",
staged=staged,
details=lines,
)
if run.vcs:
add(
"paru (devel)",
f"{len(run.vcs)} devel package(s) to rebuild",
details=[
"paru -S --needed --noconfirm --nobatchinstall <devel packages>",
"",
"These track a branch rather than a release, so paru compares",
"the commit it last built against upstream:",
"",
*[f" {v.name} {v.old} -> {v.new}" for v in run.vcs[:20]],
"",
"Nothing here is guessing a version. A -git package's recorded",
"pkgver lags what paru generated when it last built, which is",
"why only paru can tell you whether upstream has moved.",
],
)
if run.downgrades:
add(
"(held back)",
f"{len(run.downgrades)} AUR version(s) older than installed · not applied",
details=[
"these are not upgrades and will not be applied:",
"",
*[
f" {name} {old} -> {new}"
for name, (old, new) in sorted(run.downgrades.items())[:20]
],
"",
"The AUR records a version lower than what is installed. pacman",
"refuses to move backwards without being told to explicitly,",
"and doing so is rarely what you want.",
],
)
if run.flatpaks.has_updates:
add(
"flatpak update",
f"{len(run.flatpaks.updates)} update(s)",
staged=queue_mod.flatpak_queue(run.flatpaks.updates),
details=[
"flatpak update --noninteractive -y",
"",
*[f" {u.application} {u.version}" for u in run.flatpaks.updates[:20]],
],
)
if run.hygiene.reboot:
add(
"reboot",
run.hygiene.reboot,
details=[
run.hygiene.reboot,
"",
"Kernel modules and running services are the usual reason.",
"Nothing is broken; the new versions simply are not in use",
"until the machine comes back up.",
],
)
return steps
# --------------------------------------------------------------------------
# rendering
# --------------------------------------------------------------------------
def emit_plan(run: Run, review: Review) -> None:
if run.options.json:
_print_json(json.dumps(_plan_payload(run, review)))
return
for line in review_screen(
review,
theme=run.theme,
glyphs=run.glyphs,
width=run.options.width,
dry_run=run.options.dry_run,
):
run.console.print(line)
def _plan_payload(run: Run, review: Review) -> dict:
return {
"repo": [
{
"name": u.name,
"old": u.old,
"new": u.new,
"repo": u.repo,
"download": u.download,
"delta": u.delta,
"ignored": run.config.is_ignored(u.name),
}
for u in run.plan.updates
],
"aur": [
{
"name": r.name,
"old": r.old,
"new": r.new,
"tag": r.tag,
"note": r.note,
"details": r.details,
"audited": r.audited,
"ignored": run.config.is_ignored(r.name),
}
for r in review.aur
],
"vcs": [{"name": v.name, "old": v.old, "new": v.new} for v in review.vcs],
"devel_note": run.devel_note,
"checks": [
{
"ok": c.ok,
"label": c.label,
"detail": c.detail,
"details": c.details,
"action": c.action,
"warning": c.warning,
}
for c in review.checks
],
# Warnings are grouped by kind so the front end can tab through them
# instead of stacking unrelated problems into one scroll.
"warnings": [
{
"title": n.title,
"headline": n.headline,
"details": n.details,
"flagged": n.flagged,
"packages": n.packages,
}
for n in run.warning_notes
],
# The plan's steps come from here rather than being rebuilt by the front
# end. Two renderers each deriving the commands from the same facts is
# how they drift, and the front end cannot know what makepkg will do.
"steps": [
{
"index": s.index,
"command": s.command,
"detail": s.detail,
"details": s.details,
"queue": s.queue,
}
for s in review.steps
],
"flatpak": [
{"application": u.application, "version": u.version, "origin": u.origin}
for u in run.flatpaks.updates
],
"vulnerabilities": {"fixable": run.vulns.fixable, "unfixable": run.vulns.unfixable},
# acknowledged advisories are left out of the list rather than shown
# struck through: the point of acknowledging one is to stop seeing it
"advisories": [
detail.advisory_detail(a)
for a in run.vulns.sorted_advisories()
if not run.config.is_acked(a.id)
],
"acked": sorted(run.config.acked),
"acked_count": len(run.config.acked),
"ignored": sorted(run.config.ignored),
"config_path": str(config.config_path()),
"hygiene": {
"cache_reclaimable": run.hygiene.cache.reclaimable,
"pacnew": [str(p) for p in run.hygiene.drift.pacnew],
"orphans": run.hygiene.orphans,
"reboot": run.hygiene.reboot,
"stale_processes": [
{"pid": s.pid, "name": s.name, "path": s.path} for s in run.hygiene.stale
],
},
"audits": {
name: [
{
"rule": f.rule,
"severity": f.severity.name,
"title": f.title,
"evidence": f.evidence,
}
for f in report.findings
]
for name, report in run.audits.items()
},
}
# --------------------------------------------------------------------------
# applying
# --------------------------------------------------------------------------
def confirm_prompt(run: Run) -> str:
"""The one prompt, owned by the confirming step rather than the plan screen."""
if run.blocked:
n = len(run.blocked)
return f"Apply the rest, skipping {n} flagged package{'s' if n != 1 else ''}?"
return "Apply this plan?"
def confirm(run: Run, prompt: str) -> bool:
if run.options.assume_yes:
return True
if not sys.stdin.isatty():
# never assume consent on a pipe or a timer; that is what --yes is for
run.console.print("[dim]not a terminal; use --yes to proceed[/dim]")
return False
try:
answer = run.console.input(f"[bold]{prompt}[/bold] [dim][y/N][/dim] ")
except (EOFError, KeyboardInterrupt):
return False
return answer.strip().lower() in ("y", "yes")
def _runner(run: Run, total: int, graph=None) -> live.LiveRunner:
observer = live.BuildObserver(
graph or build_graph({}, set()), run.options.clone_dir, total=total
)
return live.LiveRunner(
observer,
theme=run.theme,
glyphs=run.glyphs,
console=run.console,
dashboard=not run.options.no_tui,
workers=1,
)
def apply_repo(run: Run) -> live.RunResult | None:
if not run.plan.updates:
return None
# -Syu, not -Su: refresh and upgrade in a single transaction. Splitting them
# means a failed sync silently applies a different set to the one confirmed,
# and on Arch a partial upgrade is the thing to avoid above all.
return _runner(run, len(run.plan.updates)).run(
["sudo", "pacman", "-Syu", "--noconfirm"], run.log_dir / "logs" / "pacman.log"
)
def apply_aur(run: Run) -> live.RunResult | None:
if not run.updates:
return None
candidates = set(run.updates)
infos = {}
for name in candidates:
info, _ = load(run.options.clone_dir, name)
if info:
infos[name] = info
try:
graph = build_graph(infos, candidates, strict=True)
except MissingMetadata as exc:
# keep going: an incomplete order is reported on screen rather than
# aborting a run the user already approved
run.warnings.append(str(exc))
graph = build_graph(infos, candidates)
return _runner(run, len(candidates), graph).run(
["paru", "-Sua", "--noconfirm"], run.log_dir / "logs" / "paru.log"
)
def apply_flatpak(run: Run) -> live.RunResult | None:
if not run.flatpaks.has_updates:
return None
return _runner(run, len(run.flatpaks.updates)).run(
flatpak.update_command(), run.log_dir / "logs" / "flatpak.log"
)
def record_baselines(run: Run) -> int:
"""Remember the commit we just built, so the next audit has a baseline.
Without this every audit would forever be a first audit, and the whole
"what changed since last time" premise would never come true.
"""
state = aur.State.load()
recorded = 0
for name in run.updates:
repo_dir = run.options.clone_dir / name
commit = changes.head(repo_dir).sha
meta = run.aur_meta.get(name)
state.record(
name,
maintainer=meta.maintainer if meta else None,
commit=commit,
version=run.updates[name][1],
)
recorded += 1
if recorded:
try:
state.save()
except OSError as exc:
run.warnings.append(f"could not record build baselines: {exc}")
return recorded
def write_summary(run: Run, results: dict[str, live.RunResult | None]) -> Path:
run.log_dir.mkdir(parents=True, exist_ok=True)
lines = [f"updater run {time.strftime('%Y-%m-%d %H:%M:%S')}"]
for phase, result in results.items():
if result is None:
lines.append(f"{phase:8} skipped")
else:
lines.append(
f"{phase:8} exit={result.returncode} {result.elapsed:.0f}s log={result.log_path}"
)
summary = run.log_dir / "summary.txt"
summary.write_text("\n".join(lines) + "\n")
(run.log_dir / "summary.json").write_text(json.dumps(_plan_payload(run, Review()), indent=1))
link = latest_link()
link.parent.mkdir(parents=True, exist_ok=True)
try:
if link.is_symlink() or link.exists():
link.unlink()
link.symlink_to(run.log_dir)
except OSError:
pass
return summary
# --------------------------------------------------------------------------
# subcommands
# --------------------------------------------------------------------------
def cmd_hygiene(run: Run) -> int:
report = hygiene.collect(include_processes=True)
if run.options.json:
_print_json(
json.dumps(
{
"cache_total": report.cache.total,
"cache_reclaimable": report.cache.reclaimable,
"pacnew": [str(p) for p in report.drift.pacnew],
"pacsave": [str(p) for p in report.drift.pacsave],
"orphans": report.orphans,
"reboot": report.reboot,
"stale": [{"pid": s.pid, "name": s.name, "path": s.path} for s in report.stale],
}
)
)
return EXIT_OK
console = run.console
console.print(
f"[bold]cache[/bold] {human_bytes(report.cache.total)} total · "
f"{human_bytes(report.cache.reclaimable)} reclaimable "
f"({report.cache.reclaimable_files} files)"
)
console.print(
f"[bold]config[/bold] {len(report.drift.pacnew)} .pacnew · "
f"{len(report.drift.pacsave)} .pacsave · "
f"{len(report.drift.pacorig)} .pacorig"
)
for path in report.drift.pacnew[:5]:
console.print(f" [dim]{path}[/dim]")
console.print(f"[bold]orphans[/bold] {len(report.orphans)}")
console.print(f"[bold]reboot[/bold] {report.reboot or 'not required'}")
console.print(
f"[bold]stale[/bold] {len(report.stale)} process(es) running replaced binaries"
)
for entry in report.stale[:8]:
console.print(f" [dim]{entry.pid:>6} {entry.name:<20} {entry.path}[/dim]")
return EXIT_OK
def cmd_audit(run: Run, package: str) -> int:
state = aur.State.load()
meta: dict[str, aur.AurPackage] = {}
if not run.options.offline:
try:
meta = aur.fetch([package])
except aur.AurError as exc:
run.warnings.append(str(exc))
package_meta = meta.get(package)
change = changes.change_from_clone(
run.options.clone_dir,
package,
old_commit=state.previous_commit(package),
old_maintainer=state.previous_maintainer(package),
new_maintainer=package_meta.maintainer if package_meta else None,
)
report = audit(change)
old_text = new_text = ""
repo_dir = run.options.clone_dir / package
if repo_dir.is_dir():
pkgbuild = repo_dir / "PKGBUILD"
new_text = pkgbuild.read_text(errors="replace") if pkgbuild.is_file() else ""
baseline = state.previous_commit(package)
if baseline and changes.has_commit(repo_dir, baseline):
old_text = changes.show(repo_dir, baseline, "PKGBUILD")
if run.options.json:
_print_json(
json.dumps(
{
"package": package,
"verdict": report.verdict.name,
"findings": [
{
"rule": f.rule,
"severity": f.severity.name,
"title": f.title,
"evidence": f.evidence,
}
for f in report.findings
],
}
)
)
return EXIT_BLOCKED if report.verdict is Severity.BLOCK else EXIT_OK
title = f"{package} · AUR"
if package_meta:
title = f"{package} · {package_meta.version} · AUR"
for line in audit_screen(
report,
theme=run.theme,
glyphs=run.glyphs,
width=run.options.width,
title=title,
old_text=old_text,
new_text=new_text,
baseline=bool(old_text),
):
run.console.print(line)
return EXIT_BLOCKED if report.verdict is Severity.BLOCK else EXIT_OK
#: each phase writes one log, so a failure can name the command's own output
_LOG_FOR = {"repo": "pacman", "aur": "paru", "flatpak": "flatpak"}
def _conflict_reason(lines: list[str]) -> str:
"""Explain a package conflict, which is a decision and not a fault.
pacman refuses to replace a package when told `--noconfirm`, because
replacing one is exactly the thing it should ask about first. That makes
this look like a failure when it is really a question, and the useful part
of the log -- which packages conflict -- sits above the error line that
`_failure_reason` would otherwise report.
"""
try:
start = next(i for i, line in enumerate(lines) if line.startswith(":: Conflicts found"))
except StopIteration:
return ""
pairs: list[str] = []
for line in lines[start + 1 :]:
if line.startswith(("::", "error:")):
break
# e.g. "scenefx: scenefx0.5 (scenefx)"
cleaned = line.strip()
if cleaned:
pairs.append(cleaned)
if not pairs:
return ""
# The build screen shows these on one line and truncates, so the fact that
# matters goes first: which packages conflict, and that nothing happened.
listed = ", ".join(pairs[:3])
return (
f"conflicting packages ({listed}): pacman has to be asked before it "
"replaces one and cannot while unattended, so nothing was installed. "
"Run the AUR part by hand to confirm, or hold the package back."
)
#: a line that reads like a failure, rather than progress or cleanup chatter
_LOOKS_LIKE_ERROR = re.compile(
r"\b(?:error|failed|cannot|can't|unable|not found|no such|refused|denied"
r"|conflict|missing|invalid)\b",
re.IGNORECASE,
)
def _failure_reason(log_path: Path, fallback: str, code: int | None = None) -> str:
"""Why a phase failed, in one line.
"exited 1" is not an error message. The command's own output is the only
thing that can tell someone what went wrong, and it was being written to a
log file nobody was told about.
"""
try:
text = log_path.read_text(errors="replace")
except OSError:
return fallback
lines = [line.strip() for line in text.splitlines() if line.strip()]
if not lines:
return fallback
# sudo's complaint is the one failure with a specific remedy, and it is the
# one that looks most like "nothing happened".
for line in lines:
if "a terminal is required" in line or "a password is required" in line:
return (
"sudo could not ask for your password: it has no terminal. "
"Authenticate first with `sudo -v` in another terminal, then "
"apply again."
)
conflict = _conflict_reason(lines)
if conflict:
return conflict
# A signal exit is a stop, not a fault. makepkg's last line when it is
# interrupted is "==> Cleaning up…", which says nothing about anything --
# quoting it made the message look like a mystery rather than an answer.
# subprocess reports a signal as a negative code; a shell reports 128+n.
if code is not None and (code < 0 or code >= 128):
return "stopped before it finished"
# Prefer a line that reads like a failure, and among those prefer one that
# is not makepkg's own. "==> ERROR: A failure occurred in build()." is the
# wrapper around the failure; the compiler's "error: could not compile
# `foo`" is the failure. Reporting the wrapper tells a reader nothing they
# could act on.
for line in reversed(lines):
if line.startswith("==>"):
continue
if _LOOKS_LIKE_ERROR.search(line):
return f"{fallback} — {_shorten(line)}"
# nothing more specific: makepkg's own complaint is better than nothing
for line in reversed(lines):
if _LOOKS_LIKE_ERROR.search(line):
return f"{fallback} — {_shorten(line)}"
return fallback
def _shorten(line: str, limit: int = 160) -> str:
return line if len(line) <= limit else line[: limit - 1] + "…"
def _package_weight(name: str, clone_dir: Path, *, phase: str) -> int:
"""How much of the bar this package is worth.
A package that builds runs for minutes; one that only downloads and
installs is over in about a second. Counting them alike made the bar race
to nearly full during the repository phase, where every package is fast,
and left the estimate resting on a mean of zero-second builds.
The PKGBUILD is the authority on which is which: a `build()` with anything
in it is work, and its absence is a package that unpacks what it fetched.
"""
if phase != "aur":
return stream.PLAIN_WEIGHT
pkgbuild = clone_dir / name / "PKGBUILD"
try:
funcs = parse_pkgbuild(pkgbuild.read_text(errors="replace")).funcs
except OSError:
return stream.PLAIN_WEIGHT
return stream.BUILD_WEIGHT if (funcs.get("build") or "").strip() else stream.PLAIN_WEIGHT
def _apply_phases(
plan_names: set[str], candidates: set[str], holds: set[str], skips: set[str], flatpaks: bool
) -> tuple[list[str], list[str], int]:
"""What the run will actually do: the phases, the AUR targets, the total.
One owner, because the arithmetic has to agree with itself. A hold names a
package the reader kept, and holds cover repository and AUR packages alike
-- the dashboard's solo mode passes whatever survived. Comparing set *sizes*
rather than intersecting meant a hold naming anything outside the plan
skipped the repository phase outright; and when the caller filtered the
targets and this counted them, the two could disagree about which packages
were held. Both failures are silent: the run does less than it was asked
to, or more.
"""
held = plan_names & holds
targets = sorted(candidates - skips - holds)
phases: list[str] = []
if plan_names and len(held) < len(plan_names):
phases.append("repo")
if targets:
phases.append("aur")
if flatpaks:
phases.append("flatpak")
return phases, targets, (len(plan_names) - len(held)) + len(targets)
def cmd_apply(run: Run) -> int:
"""Run the update, streaming events when asked.
Each phase is a separate command, and the repo phase gates the AUR one: a
failed or interrupted upgrade must not be followed by AUR builds, because
building against a half-upgraded system is how a tree gets broken.
"""
opts = run.options
emit = stream.Emitter()
# only the streamed path: an ordinary run has nothing listening for `s`,
# and leaving SIGUSR1 at its default keeps it a real stop
if opts.json_stream:
stream.listen_for_skip()
# The apply path never planned this, so the phase it advertises was decided
# from a default FlatpakState: has_updates was always False and flatpak was
# never applied, while the review screen listed its updates.
if opts.flatpak_phase:
run.flatpaks = flatpak.collect(include_runtimes=False)
# The same handle the review screen used. Planning this from the databases
# on disk instead meant the screen could show sixty repository updates while
# the apply decided, from a staler copy, that there were none -- and then
# silently skipped the repository phase the user had just approved.
with _plan_handle(run) as handle:
installed = repo.installed(handle)
foreign = repo.foreign(handle)
plan = repo.plan(handle)
aur_meta: dict[str, aur.AurPackage] = {}
if not opts.offline:
try:
aur_meta = aur.fetch(foreign)
except aur.AurError as exc:
run.warnings.append(str(exc))
updates, _downgrades = _classify_aur(installed, aur_meta)
# VCS packages travel with the AUR set. They are named as explicit targets
# and `--needed` is what stops them rebuilding when the AUR version already
# matches what is installed.
#
# A -git package that has been deleted from the AUR is still foreign and
# still ends in -git, but naming it makes paru refuse the entire command:
# "could not find all required packages: X (target)". One dead package took
# the whole AUR phase with it.
gone = set(aur.deleted(foreign, aur_meta, known_bases=set(installed)))
# Only the devel packages paru says have actually moved.
#
# Naming every foreign -git package instead rebuilds all of them: paru
# rebuilds a devel package whenever it is asked for one, because --needed
# does not skip them the way it skips a released version -- what is
# installed is the version generated at the last build. The screen asks
# paru which ones moved; the command must ask the same question, or the two
# disagree about what is going to happen.
devel, devel_note = aur.devel_upgrades()
vcs = [n for n in foreign if _is_vcs(n) and n not in gone and n in devel]
if devel_note and any(_is_vcs(n) for n in foreign):
# asked and could not be told: name nothing rather than everything
run.warnings.append(f"devel rebuilds not offered: {devel_note}")
if gone:
run.warnings.append(
f"{len(gone)} package(s) left the AUR and were left alone: " + ", ".join(sorted(gone))
)
# decisions: what the caller asked for, plus anything configured as ignored
ignored = set(run.config.ignored)
# The plan screen and the confirm prompt both say the audit-flagged
# packages will be skipped rather than built. Nothing carried that into
# the run, so they were built anyway -- a package the audit refused, with
# the reader told otherwise.
flagged = set(run.blocked)
skips = set(opts.skips) | {n for n in updates if n in ignored} | flagged
holds = set(opts.holds) | {u.name for u in plan.updates if u.name in ignored}
# A hold names a package the reader kept, and holds cover both repository
# and AUR packages -- the dashboard's solo mode passes whatever was kept.
# Comparing set *sizes* meant a hold naming anything outside the plan
# skipped the repository phase entirely, and the targets were never
# filtered at all, so a held AUR package was built anyway.
phases, targets, total = _apply_phases(
{u.name for u in plan.updates}, {*updates, *vcs}, holds, skips, run.flatpaks.has_updates
)
# what the run is worth in work, as opposed to in packages: the bar and the
# estimate are about the former
weights = {name: _package_weight(name, opts.clone_dir, phase="aur") for name in targets}
for update in plan.updates:
if update.name not in skips and update.name not in holds:
weights[update.name] = stream.PLAIN_WEIGHT
weight_total = sum(weights.values())
started = time.monotonic()
emit(
"start",
phases=phases,
total=total,
weight=weight_total,
weights={n: w for n, w in sorted(weights.items()) if w > 1},
held=sorted(holds),
skipped=sorted(skips),
config=str(config.config_path()),
)
if opts.json_stream:
for warning in run.warnings:
emit("error", message=warning)
if not phases:
emit("done", returncode=0, elapsed=round(time.monotonic() - started, 2))
return EXIT_OK
# Consent. `apply` changes the system, so it must never run unprompted: a
# stream consumer owns its own confirmation dialog, which is why asking for
# the stream without --yes is refused rather than assumed.
if not opts.assume_yes:
if opts.json_stream:
emit("error", message="--json-stream requires --yes; the front end owns confirmation")
emit("done", returncode=EXIT_DECLINED, elapsed=round(time.monotonic() - started, 2))
return EXIT_DECLINED
if not confirm(run, confirm_prompt(run)):
emit("done", returncode=EXIT_DECLINED, elapsed=round(time.monotonic() - started, 2))
return EXIT_DECLINED
# Both of these phases invoke sudo. Checking here means a run that cannot
# authenticate says so immediately, instead of streaming a build screen and
# failing a second later with an exit code.
if {"repo", "aur"} & set(phases) and not live.can_sudo():
message = (
"sudo cannot act without a password prompt. Run `sudo -v` in "
"a terminal first, then apply again."
)
emit("error", message=message)
emit("done", returncode=EXIT_FAILED, elapsed=round(time.monotonic() - started, 2))
return EXIT_FAILED
results: dict[str, int] = {}
if "repo" in phases:
# Not BuildObserver: that watches /proc for makepkg, and pacman is one
# process no matter how many packages it touches, so the repository
# phase rendered an empty dashboard while hundreds of megabytes moved.
observer = live.RepoObserver(
[u.name for u in plan.updates if u.name not in holds],
log_path=run.log_dir / "logs" / "pacman.log",
)
runner = stream.StreamingRunner(observer, emit)
results["repo"] = runner.run(
stream.repo_command(sorted(holds), refresh=opts.refresh),
"repo",
run.log_dir / "logs" / "pacman.log",
)
if results["repo"] != 0:
emit(
"error",
message=_failure_reason(
run.log_dir / "logs" / "pacman.log",
"the repository phase failed; not building AUR packages",
),
)
emit("done", returncode=results["repo"], elapsed=round(time.monotonic() - started, 2))
return EXIT_FAILED
if "aur" in phases:
infos = {}
for name in targets:
info, _ = load(opts.clone_dir, name)
if info:
infos[name] = info
try:
graph = build_graph(infos, set(targets), strict=True)
except MissingMetadata as exc:
run.warnings.append(str(exc))
graph = build_graph(infos, set(targets))
if opts.json_stream:
emit("error", message=str(exc), details=exc.detail)
observer = live.BuildObserver(graph, opts.clone_dir, total=len(targets))
runner = stream.StreamingRunner(observer, emit, workers=opts.workers)
# One package per invocation, in dependency order.
#
# paru resolves its entire target list before building any of it, so a
# single unresolvable target -- a conflict, or a package pulled from
# the AUR between the check and the run -- failed the command outright
# and *nothing* was installed. Asking for one package at a time is the
# only thing that makes one bad package cost one package: everything
# else still lands, and the failures are reported together at the end.
ordered = [n for n in graph.order if n in set(targets)]
ordered += [n for n in sorted(targets) if n not in set(ordered)]
aur_logs = run.log_dir / "logs" / "aur"
before = _installed_versions()
failures, remaining, passed_over = _apply_aur_each(
runner,
ordered,
aur_logs,
total=len(targets),
since=started,
report=(lambda message: emit("error", message=message)) if opts.json_stream else None,
failed=(
lambda name, reason: emit(
"failed",
name=name,
reason=reason,
weight=weights.get(name, stream.PLAIN_WEIGHT),
)
)
if opts.json_stream
else None,
skipped=(
lambda name: emit(
"skipped", name=name, weight=weights.get(name, stream.PLAIN_WEIGHT)
)
)
if opts.json_stream
else None,
)
# Read after the loop, which is what sets it. Snapshotting it before
# meant `stopped` was always False: Ctrl+C could not stop the run, the
# interrupted package appeared in neither the failures nor the resume
# offer, and baselines were recorded for packages never built.
stopped = getattr(runner, "interrupted", False)
results["aur"] = 0 if not failures and not stopped else 1
# One message, not four. The feed already shows which packages did not
# install, so the summary names them once and says how to finish them
# rather than restating the count in three different ways.
unfinished = [name for name, _ in failures] + remaining
if passed_over:
# said out loud, and kept out of the resume: the reader asked for
# this one to be left
run.warnings.append(
f"{len(passed_over)} package(s) skipped at your request: " + ", ".join(passed_over)
)
if opts.json_stream:
emit(
"error",
message=(
f"skipped {len(passed_over)} at your request: " + ", ".join(passed_over)
),
)
if unfinished:
run.warnings.append(
f"{len(unfinished)} of {len(ordered)} AUR package(s) are not "
f"installed: {', '.join(unfinished)}"
)
if opts.json_stream:
if stopped:
emit(
"error",
message=(
f"stopped · {len(unfinished)} of {len(ordered)} "
"package(s) still to build"
),
)
else:
emit(
"error",
message=(
f"{len(unfinished)} of {len(ordered)} package(s) did "
f"not install: {', '.join(name for name, _ in failures)}"
),
)
emit(
"recovery",
label="resume in a terminal",
argv=stream.aur_command_interactive(unfinished),
note=(
f"builds the remaining {len(unfinished)}; prompts "
"and the PKGBUILD review work as usual"
),
)
if stopped:
emit("done", returncode=EXIT_FAILED, elapsed=round(time.monotonic() - started, 2))
return EXIT_FAILED
_record_aur_baselines(
run,
ordered,
updates,
aur_meta,
before=before,
after=_installed_versions() if failures else None,
complete=not failures,
)
if "flatpak" in phases:
observer = live.BuildObserver(
build_graph({}, set()), opts.clone_dir, total=len(run.flatpaks.updates)
)
runner = stream.StreamingRunner(observer, emit)
results["flatpak"] = runner.run(
stream.flatpak_command(), "flatpak", run.log_dir / "logs" / "flatpak.log"
)
failed = {name: rc for name, rc in results.items() if rc != 0}
for phase in sorted(failed):
if phase == "aur":
continue # already reported per package, with its own log
log = run.log_dir / "logs" / f"{_LOG_FOR[phase]}.log"
if opts.json_stream:
emit(
"error",
message=_failure_reason(log, f"the {phase} phase failed (exit {failed[phase]})"),
)
code = EXIT_FAILED if failed else EXIT_OK
emit(
"done", returncode=code, elapsed=round(time.monotonic() - started, 2), failed=sorted(failed)
)
return code
def _apply_aur_each(
runner,
ordered: list[str],
log_dir: Path,
*,
total: int = 0,
since: float | None = None,
report=None,
failed=None,
skipped=None,
) -> tuple[list[tuple[str, str]], list[str], list[str]]:
"""Build and install one package per invocation, collecting the failures.
Split out from the phase because "the others still ran" is the entire point
of doing it this way, and that should be provable rather than inferred from
the shape of a loop.
paru resolves its whole target list before building any of it, so naming
every package at once makes one unresolvable target -- a conflict, or a
package pulled from the AUR between the check and the run -- fail the
command outright, installing nothing. One name per invocation is the only
thing that makes one bad package cost one package.
"""
failures: list[tuple[str, str]] = []
remaining: list[str] = []
passed_over: list[str] = []
for index, name in enumerate(ordered):
# asked to stop: do not start the next package. The child already
# running was unwound by the runner.
if getattr(runner, "interrupted", False):
remaining = ordered[index:]
break
log = log_dir / f"{name}.log"
code = runner.run(stream.aur_command([name]), "aur", log, total=total, since=since)
if getattr(runner, "skipped", False):
# Asked for, so not a failure and not something to offer back:
# the reader decided this one was not worth waiting for.
passed_over.append(name)
if skipped is not None:
skipped(name)
continue
if code == 0:
continue
fallback = f"failed (exit {code})"
reason = _failure_reason(log, fallback, code)
failures.append((name, reason))
# Tell the observer too. Without this the watcher files the package as
# *done* -- it saw the process exit and nothing said otherwise -- so
# the stream carried a `finished state=done` and a `failed` for the
# same package, and `done` reached the total with packages missing.
observer = getattr(runner, "observer", None)
if observer is not None:
with contextlib.suppress(Exception):
observer.fail(name, reason)
# The interface watches processes, so it saw this package's makepkg
# exit and filed it as done -- it cannot read an exit code. Without
# this event the failure count stayed at zero while the log said
# otherwise.
if failed is not None:
failed(name, reason)
if report is not None and reason != fallback:
# named once, and only when the reason says more than the feed
# already does: the feed shows which package failed, but a bare
# reason is ambiguous when several have
report(f"{name}: {reason}")
return failures, remaining, passed_over
def _installed_versions() -> dict[str, str]:
"""What is installed right now, for telling an install from an attempt."""
try:
with repo.connect() as handle:
return repo.installed(handle)
except Exception: # noqa: BLE001 - a missing comparison is not fatal
return {}
def _record_aur_baselines(
run: Run,
targets: list[str],
updates: dict[str, tuple[str, str]],
meta: dict[str, aur.AurPackage],
*,
before: dict[str, str] | None = None,
after: dict[str, str] | None = None,
complete: bool = True,
) -> None:
"""Remember the revision each package was built from.
A baseline is a claim that this revision was built and installed, so it is
only recorded for packages that got there. `complete` says the whole phase
succeeded, in which case every target qualifies; otherwise only the ones
whose installed version actually moved.
"""
state = aur.State.load()
if after is None:
after = _installed_versions() if not complete and before else {}
recorded = 0
for name in targets:
if not complete:
changed = before and after and before.get(name) != after.get(name)
if not changed:
continue
commit = changes.head(run.options.clone_dir / name).sha
entry = meta.get(name)
state.record(
name,
maintainer=entry.maintainer if entry else None,
commit=commit,
version=updates.get(name, ("", ""))[1],
)
recorded += 1
if recorded != len(targets):
run.warnings.append(
f"recorded baselines for {recorded} of {len(targets)} package(s); "
"the rest were not installed"
)
try:
state.save()
except OSError as exc:
run.warnings.append(f"could not record build baselines: {exc}")
def cmd_advisory(run: Run, advisory_id: str) -> int:
"""The tracker's own record for one advisory, so the link is a last resort."""
payload = fetch.advisory_detail(advisory_id)
if run.options.json:
_print_json(json.dumps(payload))
return EXIT_OK if payload.get("available") else EXIT_FAILED
if not payload.get("available"):
run.console.print(f"[red]{payload.get('reason')}[/red]")
return EXIT_FAILED
c = run.console
c.print(f"[bold]{payload['id']}[/bold] [dim]{payload['packages']}[/dim]")
c.print(f" severity {payload['severity']}")
c.print(f" type {payload['kind']}")
c.print(f" status {payload['status']}")
c.print(f" affected {payload['affected'] or '—'}")
c.print(f" fixed {payload['fixed'] or 'no fix available'}")
if payload.get("notes"):
c.print(f" notes {payload['notes']}")
c.print()
c.print(f"[bold]{len(payload['issues'])} CVE(s)[/bold]")
for issue in payload["issues"]:
c.print(f" {issue}")
for reference in payload.get("references", []):
c.print(f" [dim]{reference}[/dim]")
c.print(f"\n[dim]{payload['url']}[/dim]")
return EXIT_OK
def cmd_security(run: Run) -> int:
"""Run arch-audit now and report it, rather than whatever the plan cached."""
report = security.scan(force=True)
payload = {
"available": report.available,
"reason": report.reason,
"total": len(report.advisories),
"fixable": len(report.fixable),
"unfixable": len(report.unfixable),
"advisories": [detail.advisory_detail(a) for a in report.sorted_advisories()],
}
if run.options.json:
_print_json(json.dumps(payload))
return EXIT_OK if report.available else EXIT_FAILED
if not report.available:
run.console.print(f"[red]{report.reason}[/red]")
return EXIT_FAILED
run.console.print(
f"[bold]{len(report.advisories)} advisories[/bold] [dim]{len(report.fixable)} fixable[/dim]"
)
for advisory in report.sorted_advisories():
fix = advisory.fixed or "no fix available"
run.console.print(f" {advisory.severity:<14} {advisory.id:<12} {fix}")
return EXIT_OK
def _print_json(payload: str) -> None:
"""Write JSON to stdout with nothing added.
Deliberately not `console.print_json`: Rich colourises that output, and it
honours FORCE_COLOR, so on a machine that forces colour the JSON arrived
full of ANSI escapes and the front end could not parse a single byte of it.
Machine-readable output must be byte-exact.
"""
sys.stdout.write(payload + "\n")
sys.stdout.flush()
def cmd_show(run: Run, package: str) -> int:
"""Assemble the drill-down for one package: diff, audit, CVEs, metadata.
Deliberately a separate command rather than part of the plan, so the plan
stays small and a diff is only computed when somebody asks for it.
"""
state = aur.State.load()
with repo.connect() as handle:
installed = repo.installed(handle)
is_foreign = package in set(repo.foreign(handle))
sync_meta = None if is_foreign else _sync_metadata(handle, package)
dependents = [] if is_foreign else _dependents(handle, package)
plan = None if is_foreign else repo.plan(handle)
if not is_foreign:
# A repository package has no PKGBUILD to diff. Fetching one and
# reporting the 404 was worse than useless: it looked like an error.
advisories = []
if not run.options.offline:
advisories = security.scan().for_package(package)
detail_payload = detail.repo_package_detail(
package,
installed=installed.get(package, ""),
meta=sync_meta,
plan=plan,
advisories=advisories,
dependents=dependents,
hooks=_hooks_for(package),
)
if run.options.json:
_print_json(json.dumps(detail_payload))
return EXIT_OK
_emit_repo_detail(run, detail_payload)
return EXIT_OK
meta: aur.AurPackage | None = None
if not run.options.offline:
try:
meta = aur.fetch([package]).get(package)
except aur.AurError as exc:
run.warnings.append(str(exc))
advisories = []
if not run.options.offline:
advisories = security.scan().for_package(package)
payload = detail.package_detail(
run.options.clone_dir,
package,
installed=installed.get(package, ""),
available=meta.version if meta else "",
meta=meta,
previous_maintainer=state.previous_maintainer(package),
recorded_commit=state.previous_commit(package),
advisories=advisories,
kind="aur" if is_foreign else "repo",
network=not run.options.offline,
)
payload["ignored"] = run.config.is_ignored(package)
# Which way the version goes, decided here because this is where the
# installed and available versions are both known. `→` does not say
# whether it is an upgrade, and the overlay has no other marker.
installed_version = installed.get(package, "")
if is_foreign and installed_version and meta and meta.version:
payload["downgrade"] = pyalpm.vercmp(meta.version, installed_version) < 0
if run.options.json:
_print_json(json.dumps(payload))
return EXIT_OK
_render_detail(run, payload)
return EXIT_OK
def _render_detail(run: Run, payload: dict) -> None:
"""Human-readable rendering of the same payload."""
c = run.console
c.print(
f"[bold]{payload['package']}[/bold] "
f"[dim]{payload['installed']} → {payload['available']}[/dim]"
)
if payload.get("baseline_note"):
c.print(f"[dim]{payload['baseline_note']}[/dim]")
c.print()
findings = payload.get("findings", [])
if findings:
c.print("[bold]FINDINGS[/bold]")
for f in findings:
glyph = {"BLOCK": "!!", "LOOK": "!", "INFO": "·"}.get(f["severity"], "·")
c.print(f" {glyph} {f['title']}")
if f.get("evidence"):
c.print(f" [dim]{f['evidence']}[/dim]")
c.print()
advisories = payload.get("advisories", [])
if advisories:
c.print("[bold]SECURITY[/bold]")
for a in advisories:
fix = a["fixed"] or "no fix available"
c.print(f" {a['severity']:<14} {a['id']:<12} {fix}")
c.print(f" [dim]{a['url']}[/dim]")
for issue in a["issues"][:8]:
c.print(f" [dim]{issue}[/dim]")
if len(a["issues"]) > 8:
c.print(f" [dim]+ {len(a['issues']) - 8} more[/dim]")
c.print()
diff = payload.get("diff", {})
lines = diff.get("lines", [])
if lines:
c.print(f"[bold]DIFF[/bold] [dim]+{diff['added']} -{diff['removed']}[/dim]")
for line in lines[:200]:
marker = {"add": "+", "del": "-"}.get(line["kind"], " ")
style = {"add": "green", "del": "red"}.get(line["kind"], "dim")
c.print(f" {marker} [{style}]{line['text']}[/{style}]")
elif payload.get("pkgbuild"):
c.print("[bold]PKGBUILD[/bold] [dim](shown as reference)[/dim]")
for i, text in enumerate(payload["pkgbuild"][:200], 1):
c.print(f" [dim]{i:>4}[/dim] {text}")
def cmd_checks(run: Run) -> int:
"""The machine-local checks only. No network, no AUR, no archive scan."""
run.hygiene = hygiene.collect(include_processes=False)
rows = _hygiene_checks(run)
_print_json(
json.dumps(
{
"checks": [
{
"ok": c.ok,
"label": c.label,
"detail": c.detail,
"details": c.details,
"action": c.action,
"warning": c.warning,
}
for c in rows
],
# This command can only recompute the machine-local rows. Naming them
# lets the reader replace exactly those and leave the ones it cannot
# see -- whether a reboot is pending, whether the archive is reachable
# -- alone, instead of silently dropping them from the screen.
"covers": [c.label for c in rows],
}
)
)
return EXIT_OK
def cmd_log(run: Run) -> int:
link = latest_link()
if not link.exists():
run.console.print("[dim]no previous run[/dim]")
return EXIT_FAILED
target = link.resolve()
run.console.print(str(target))
if run.options.tail:
for path in sorted(target.glob("logs/*.log")):
run.console.print(f"\n[bold]{path.name}[/bold]")
run.console.print(path.read_text(errors="replace")[-4000:])
return EXIT_OK
# --------------------------------------------------------------------------
# main
# --------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
"""Entry point. One owner for turning a failure into an exit code.
Without this an unexpected error -- an unreadable file, an odd revision in
a clone, a pacman database in a state nobody anticipated -- leaves the
terminal with a Python traceback, which tells the person running an update
nothing about what to do next.
"""
try:
return _main(argv)
except KeyboardInterrupt:
return EXIT_FAILED
except Exception as exc: # noqa: BLE001 - the top of the program
if os.environ.get("UPDATER_DEBUG"):
raise
Console(stderr=True).print(f"[red]updater failed:[/red] {type(exc).__name__}: {exc}")
Console(stderr=True).print("[dim]set UPDATER_DEBUG=1 for the full traceback[/dim]")
return EXIT_FAILED
def _main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
opts = options_from(args)
tier, reason = icons.detect(override=opts.icon_tier)
console = Console()
# the renderer lays out to an exact column count; if it disagrees with the
# Console's own width, every line that is wide gets wrapped and the grid is
# destroyed. Let the Console, which knows best, be the authority.
opts.width = console.width
run = Run(options=opts, glyphs=icons.glyphs(tier), console=console, config=config.Config.load())
if os.environ.get("UPDATER_DEBUG"):
run.console.print(f"[dim]icons: {tier} ({reason})[/dim]", highlight=False)
if opts.command == "log":
return cmd_log(run)
if opts.command == "checks":
return cmd_checks(run)
if opts.command == "hygiene":
return cmd_hygiene(run)
if opts.command == "apply":
return cmd_apply(run)
# Everything above answers a question about one thing and does its own
# lookups. Gathered before them, a full plan -- a mirror sync, an RPC for
# every foreign package, an arch-audit run, a walk of /etc -- ran and was
# thrown away, which is what made opening a package take seconds.
if opts.command == "audit":
return cmd_audit(run, args.package)
if opts.command == "show":
return cmd_show(run, args.package)
if opts.command == "advisory":
return cmd_advisory(run, args.advisory_id)
if opts.command == "security":
return cmd_security(run)
review = gather(run)
emit_plan(run, review)
if opts.dry_run:
return EXIT_OK
if not (review.repo or review.aur or run.vcs or run.flatpaks.has_updates):
if not opts.json:
run.console.print("[dim]nothing to do[/dim]")
return EXIT_OK
if not confirm(run, confirm_prompt(run)):
return EXIT_DECLINED
results: dict[str, live.RunResult | None] = {"repo": apply_repo(run)}
repo_result = results["repo"]
if repo_result is not None and not repo_result.ok:
# never build AUR packages against a half-upgraded system
run.console.print("[red]repo phase failed; not building AUR packages[/red]")
write_summary(run, results)
return EXIT_FAILED
results["aur"] = apply_aur(run)
aur_result = results["aur"]
if aur_result is not None and aur_result.ok:
record_baselines(run)
results["flatpak"] = apply_flatpak(run)
summary = write_summary(run, results)
if not opts.json:
run.console.print(f"[dim]summary: {summary}[/dim]")
failed = [name for name, res in results.items() if res is not None and not res.ok]
if failed:
run.console.print(f"[red]failed phases: {', '.join(failed)}[/red]")
return EXIT_FAILED
final = hygiene.collect(include_processes=True)
if final.reboot:
run.console.print(f"[yellow]reboot recommended:[/yellow] {final.reboot}")
if final.stale:
run.console.print(
f"[yellow]{len(final.stale)} process(es) still running replaced binaries[/yellow]"
)
return EXIT_OK
if __name__ == "__main__":
raise SystemExit(main())