memgroups.py

#!/usr/bin/env python3
"""memgroups - group RAM consumption by application.

Reads /proc, folds processes into logical groups (Chrome, Devin Desktop, MCP
servers, ...) and prints a table. Read-only, works without root.

One file, stdlib only, no runtime dependencies. That is a constraint, not a
stage: the whole tool is meant to be readable end to end by one person, and to
run on a machine where installing anything is the last thing you want to do.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import time

try:
    import tomllib  # Python 3.11+
except ModuleNotFoundError:  # 3.10, which is what Ubuntu 22.04 ships
    # Deliberately NOT falling back to the `tomli` PyPI package: stdlib-only is a
    # design constraint, and the tool's whole job is answerable without a config.
    # Degrade to defaults and say so only if a config actually exists.
    tomllib = None
from dataclasses import dataclass, field
from pathlib import Path


def _detect_version() -> str:
    """The installed package's version, or a marker when there is no package.

    The number lives in exactly one place - the git tag - and setuptools-scm
    writes it into the wheel's metadata at build time. Reading it back here
    means the tag, the wheel, the Arch package and `--version` can never
    disagree, which is the failure a hand-edited literal invites.

    importlib.metadata is stdlib, so this adds no dependency. Running straight
    out of a checkout there is no metadata to read, and saying so is more honest
    than printing a number that was true at some point in the past.
    """
    try:
        from importlib.metadata import PackageNotFoundError, version
    except ImportError:  # pragma: no cover - Python < 3.8
        return "unknown"
    try:
        return version("memgroups")
    except PackageNotFoundError:
        return "unknown (running from a source checkout)"


__version__ = _detect_version()

DOCKER_TIMEOUT = 3.0
DEFAULT_TOP = 15
DEFAULT_CGROUPFS = "/sys/fs/cgroup"
ZSWAP_SOURCE = "cgroup2 memory.stat"
TMPFS_SOURCE = "statvfs of tmpfs mounts"
NAME_WIDTH_MAX = 56  # enough for "Docker: ghcr.io/modelcontextprotocol/inspector:latest"
CMDLINE_MAX = 4096  # cap argv: bound the per-process regex work against a hostile process
FILE_HINT_MIN_KB = 1024 * 1024  # 1 GiB held by ONE process - see file_hint()
FILE_HINT_RATIO = 2  # ... and the group is 2x more FILE than SIZE
FILE_HINT_NAME_MAX = 32  # keeps the hint line inside 80 columns

METRICS = ("data", "anon", "rss", "pss")
# The default. "anon" cannot see a hypervisor's guest RAM, which VirtualBox and
# a file-backed QEMU map as FILE pages: seven running VMs read as 0.78 GiB
# instead of 20.45 GiB on the machine this was measured on.
DEFAULT_METRIC = "data"
# Metrics whose SIZE can be checked against /proc/meminfo, so "ram" scope means
# something: anon claims AnonPages, data claims AnonPages plus part of Mapped.
# "rss" cannot - it counts a shared page once per process holding it.
RECONCILABLE = ("data", "anon")
SCOPES = ("ram", "processes")
SORTS = ("size", "swap", "zswap", "name")

UNCLASSIFIED = "Unclassified"
REST_FMT = "Other groups ({n})"
DOCKER_PREFIX = "Docker: "
COMPOSE_PREFIX = "Docker compose: "
COMPOSE_LABEL = "com.docker.compose.project"
ROW_TMPFS = "tmpfs / shm"
ROW_KERNEL = "Kernel (slab, page tables, stacks)"
ROW_ZSWAP = "zswap (compressed pool)"  # exactly 23 chars: fits the 80-column name field

# xterm-256 palette, picked to stay readable on both dark and light terminals.
PALETTE = (39, 214, 76, 203, 141, 45, 220, 170, 111, 208, 84, 167)

CONTAINER_RE = re.compile(r"docker[-/]([0-9a-f]{12,64})")
# Recognises the collapsed row by the same format string that builds it, so the
# two can never drift apart. Used to keep the headline off a row that is not one
# application.
REST_RE = re.compile(re.escape(REST_FMT).replace(re.escape("{n}"), r"\d+") + r"$")
# One row per virtual machine, the same way one row per container works. The
# name is free text the VM's owner chose, so it is read to the next option and
# not to the next space: "FS: Ub_Mate" is a real machine name on the host this
# was written against. Shared host daemons - VBoxSVC, libvirtd, its dnsmasq -
# carry no machine name and stay in the plain Virtualization row, exactly as
# "Docker engine" stays separate from the containers.
# Checked first, and anchored to argv[0]: "-name" is an ordinary option, and
# without this gate "find /home/u/obs -name *.mkv" became a virtual machine
# called *.mkv. The name patterns are only ever applied to a hypervisor.
VM_HOST_RE = re.compile(r"^(?:\S*/)?(?:qemu-system-[a-z0-9_]+|VirtualBoxVM|VBoxHeadless)(?:\s|$)")
VM_NAME_RE = (
    re.compile(r"--comment\s+(.+?)\s+--startvm\b"),  # VirtualBox, VBoxHeadless
    re.compile(r"-name\s+(?:guest=)?([^,\s]+)"),  # QEMU, plain and libvirt forms
)
# The name comes from the process's own command line, so it is not ours to
# trust with the table's width. Control characters are already stripped in
# collect_procs; this is the length half of the same argument.
VM_NAME_MAX = 48
DOCKER_CLIENT_RE = re.compile(r"(?:^|/)docker\s+(?:run|exec|start)\b")
# Control characters, incl. ESC/CR/LF. A process controls its own argv and comm,
# so strip these before printing - otherwise a hostile process can forge table
# rows or inject terminal escape sequences into our output. Same idea as ps.
CONTROL_RE = re.compile(r"[\x00-\x1f\x7f]")
# Processes whose job is to LAUNCH other processes. Their children are unrelated
# applications, so inheriting a group across one is always wrong. Without this the
# ancestor walk hands every unmatched desktop app to "System services" - a modern
# desktop starts apps from `systemd --user` and from D-Bus activation - which is
# worse than saying nothing, because a wrong answer arrives with the same
# confidence as a right one, and it makes the "how much is Unclassified on a
# foreign machine" question unanswerable.
#
# The list has four kinds of launcher on it. Service managers and D-Bus come
# first. Then the remote and console entry points: a per-connection `sshd:
# user@pts/0` matches the System services rule, so without the barrier every
# single thing run over SSH inherits System services - measured on a real host,
# where this tool classified *itself* that way. `login` does the same for a text
# console, since agetty above it is a system service too. Then display managers,
# which own a graphical session but are not what runs in it. Last, Wayland
# compositors: they keep the clients they spawn as direct children, so the
# moment a compositor is classified it starts absorbing every unmatched
# application on the desktop.
SUPERVISOR_RE = re.compile(
    r"(?:^|/)(?:systemd|init|dbus-daemon|dbus-broker(?:-launch)?|"
    r"xdg-desktop-portal(?:-[a-z]+)?)\b"
    r"|(?:^|/)sshd(?:-session)?\b"
    r"|(?:^|/)login(?:$|\s)"
    # Not \b: GDM and SDDM run an Xorg of their own whose arguments carry
    # "/run/user/120/gdm/Xauthority" and "/run/sddm/xauth_HTgLMj". That Xorg is
    # a real X server and belongs in Desktop / X, so the name must be followed
    # by the end of the word, not by another path component.
    r"|(?:^|/)(?:lightdm|gdm[0-9]?|sddm(?:-helper)?|greetd|plasmalogin(?:-helper)?)"
    r"(?:$|[- ])"
    r"|(?:^|/bin/)(?:sway|[Hh]yprland|river|niri|wayfire|labwc|weston|dwl|cage)(?![\w./-])"
    # Job schedulers and the session's key agent. cron/crond/anacron run
    # arbitrary user jobs, and ssh-agent is the classic ~/.xinitrc wrapper that
    # execs the whole session - all of them are classified below, so without a
    # barrier every backup script under cron and every window of an
    # ssh-agent-wrapped session would inherit "System services".
    #
    # This one alternative is looser than the rest: the name is matched after
    # any slash, so "vim /home/me/cron" is a barrier too. That is deliberate,
    # not an oversight. A barrier costs nothing unless a process has children
    # that would otherwise inherit through it, and an editor does not. Tighten
    # this to argv[0] and the real case is lost: "/bin/sh -c /home/me/job.sh"
    # under cron is reached through the PARENT's cmdline, which is where the
    # scheduler's name sits.
    r"|(?:^|/)(?:(?:ana)?crond?|ssh-agent)(?:$|\s)"
)
# /proc/mounts escapes exactly these four characters in the path fields
# (fs/proc_namespace.c mangles " \t\n\\"). Decode in ONE pass: chained
# str.replace() would turn the escaped backslash "\134040" into a space.
MOUNT_ESCAPE_RE = re.compile(r"\\(040|011|012|134)")
MOUNT_ESCAPES = {"040": " ", "011": "\t", "012": "\n", "134": "\\"}

# name, cmdline patterns, comm patterns, container patterns.
# Order matters: the first matching rule wins. Anything running inside a
# container is labelled "Docker compose: <project>" / "Docker: <image>" before
# these rules are consulted, so they only ever see host processes.
DEFAULT_RULES: tuple[tuple[str, list[str], list[str], list[str]], ...] = (
    # Backups come FIRST, and every pattern below is anchored to argv[0]. Both
    # halves of that are load-bearing, and together they close a defect that ran
    # in both directions.
    #
    # A backup names what it copies. With the application rules above these, 83
    # patterns could steal a running backup: "rsync -a /etc/nginx /backup" was a
    # web server, "borg create ::snap /etc/systemd" a system service, and
    # "restic backup /home/u/.config/kwin" a desktop. Anchoring 83 patterns one
    # by one is 83 chances to break something; moving five rules is one change,
    # and it is also the correct statement - a running backup is a backup no
    # matter whose directory is in its arguments.
    #
    # The anchor is what makes the move safe in the other direction. Unanchored,
    # "\brsync\b" first would take "vim /etc/rsync.conf" and "code ~/src/rsync"
    # (it takes them today, from further down the list). Anchored, only the
    # program itself matches. borg and duplicity get a second pattern because
    # they are Python programs: the kernel puts the interpreter in argv[0], so
    # the name sits in the second word.
    #
    # Container membership is decided before any of these rules run, so a restic
    # inside a container is still "Docker: <image>".
    ("rsync", [r"^(?:\S*/)?rsync(?:\s|$)"], [r"^rsync$"], []),
    ("restic", [r"^(?:\S*/)?restic(?:\s|$)"], [r"^restic$"], []),
    (
        "borg",
        [
            r"^(?:\S*/)?borg(backup)?(?:\s|$)",
            r"^(?:\S*/)?python[0-9.]*(?: -\w+)* /\S+/borg(backup)?(?:\s|$)",
        ],
        [r"^borg$"],
        [],
    ),
    (
        "duplicity",
        [
            r"^(?:\S*/)?duplicity(?:\s|$)",
            r"^(?:\S*/)?python[0-9.]*(?: -\w+)* /\S+/duplicity(?:\s|$)",
        ],
        [r"^duplicity$"],
        [],
    ),
    ("rclone", [r"^(?:\S*/)?rclone(?:\s|$)"], [r"^rclone$"], []),
    ("Claude Code CLI", [r"/claude-code/bin/claude\b"], [r"^claude$"], []),
    # Browsers.
    (
        "Google Chrome",
        [r"/opt/google/chrome", r"/chromium(-browser)?\b", r"ms-playwright/chromium"],
        [],
        [],
    ),
    ("Mozilla Firefox", [r"/firefox\b", r"/librewolf\b", r"/waterfox\b", r"/floorp\b"], [], []),
    # IDE / AI editors.
    (
        "JetBrains IDE",
        [
            r"/(webstorm|idea|pycharm|goland|phpstorm|clion|rider|rubymine)\b",
            r"JetBrains",
            r"/jbr/bin/java",
            r"jcef",
            r"cef_server",
        ],
        [],
        [],
    ),
    ("Windsurf", [r"/usr/share/windsurf", r"/electron39/"], [], []),
    ("Devin Desktop", [r"/devin-desktop", r"/electron42/", r"\.config/Devin"], [], []),
    ("VS Code / Cursor", [r"/usr/share/code\b", r"/cursor\b", r"vscode-server"], [], []),
    # Messengers.
    (
        "Messengers",
        [
            # Anchored to the binary, not the directory: a bare "/Telegram\b"
            # also matched "rsync -a ~/Downloads/Telegram/ /backup/" and sent it here.
            r"/Telegram/Telegram\b",
            # What every distro package, snap and flatpak actually installs.
            r"/telegram-desktop\b",
            r"/usr/lib/slack",
            r"[Dd]iscord",
            r"signal-desktop",
            r"/viber\b",
            r"/bin/element-desktop\b",
            r"im\.riot\.Riot",
            r"/opt/zoom/",
            r"/bin/zoom\b",
            r"teams-for-linux",
            r"com\.microsoft\.Teams",
        ],
        [r"^(element-desktop|teams-for-linux|telegram-desktop)", r"^zoom$"],
        [],
    ),
    # Host infrastructure.
    (
        "Docker engine",
        [r"/usr/bin/dockerd", r"\bcontainerd\b", r"docker-proxy", r"containerd-shim"],
        [r"^(dockerd|containerd|containerd-shim.*|docker-proxy|docker-init)$"],
        [],
    ),
    (
        "Virtualization (KVM/VirtualBox)",
        # The VirtualBox directory catches the rest of the host-side family -
        # VBoxSVC, VBoxXPCOMIPCD, VBoxNetDHCP. Guest additions install to
        # /usr/bin and /usr/sbin instead, which is what keeps them out of this
        # row and in "Guest agents (VM/cloud)".
        [
            # All three anchored to argv[0]: the bare words used to file
            # "grep -rn VirtualBox /etc" - and this tool's own command line -
            # as a hypervisor.
            r"^(?:\S*/)?qemu-system-[a-z0-9_]+(?:\s|$)",
            r"^(?:\S*/)?(?:VBoxHeadless|VirtualBox(?:VM)?)(?:\s|$)",
            # The host installation directory, both spellings: Debian's
            # /usr/lib/virtualbox and Oracle's /opt/VirtualBox. Requiring a
            # file component after it keeps "rsync -a /usr/lib/virtualbox/ /b"
            # in the rsync row.
            r"^(?:\S*/)?[Vv]irtual[Bb]ox/[A-Za-z]",
            # Deliberately NOT argv[0]-anchored: libvirt's dnsmasq is named
            # /usr/bin/dnsmasq and only its --conf-file argument says libvirt.
            r"/libvirt",
        ],
        # qemu-system-x86_64-spice cannot match the cmdline pattern (the hyphen
        # ends [a-z0-9_]+), so the comm half is what catches the wrappers.
        [r"^qemu-system-", r"^(?:VirtualBox(?:VM)?|VBoxHeadless)$"],
        [],
    ),
    # Services, both on the host and inside containers.
    (
        "Databases",
        [r"\bpostgres\b", r"\bmysqld\b", r"\bmariadbd\b", r"redis-server", r"\bmongod\b"],
        [],
        [],
    ),
    (
        "Web servers and apps",
        [r"\bnginx\b", r"\bapache2\b", r"\bhttpd\b", r"php-fpm", r"\bgunicorn\b", r"\buwsgi\b"],
        [],
        [],
    ),
    # Desktop session.
    (
        "Desktop / X",
        [
            r"/Xorg\b",
            # Every Wayland session runs one, and it holds the pixmaps of every
            # X11 client on the desktop, so it is routinely the largest process
            # on the machine. /Xorg above cannot reach it. Anchored to /bin/ or
            # to the start of argv, or "rsync -a ~/src/xwayland /backup" would
            # be filed as a desktop.
            r"(?:^|/bin/)[Xx]wayland(?:-satellite)?(?![\w./-])",
            r"\bxfce",
            r"\bmate-",
            r"\bmarco\b",
            r"\bcaja\b",
            r"plasmashell",
            r"\bkwin",
            r"gnome-shell",
            # GNOME's settings daemons: about fifteen of them on a stock
            # session. Anchored to argv[0] and to the directory they install
            # into (/usr/libexec on Arch and Fedora, a gnome-settings-daemon
            # subdirectory on Debian), because "gsd-" alone is short enough to
            # appear inside a backup path.
            r"^\S*/(?:libexec|gnome-settings-daemon)/gsd-[a-z0-9-]+(?:\s|$)",
            # The KDE daemon zoo. ^\S* so the directory has to be part of
            # argv[0]. The backup rules at the top of the list already protect
            # "rsync -a /usr/libexec/kded6 /backup", but the anchor is what
            # keeps a plain "vim /usr/libexec/kded6" out. Names are enumerated,
            # never a k<word>d shape - /usr/sbin/knotd and /usr/bin/kadmind are
            # not KDE. /kf[0-9]* covers the KF6 helper directory.
            r"^\S*/(bin|lib(64|exec)?)(/kf[0-9]*)?/k(ded|globalacceld|walletd"
            r"|activitymanagerd|access|runner|systemstats"
            r"|screen_backend_launcher|screenlocker_greet|deconnectd|secretd"
            r"|nighttimed|iod|smserver|unifiedpush-[a-z-]+|cminit|deinit"
            r"|launcher)[0-9]*(?:\s|$)",
            # Session helpers that are not k-prefixed: two Plasma protocol
            # bridges, the crash handler family, and the XSETTINGS bridge (the
            # X-side twin of gsd-xsettings, which is already in this row).
            r"^(?:\S*/)?(?:xembedsniproxy|gmenudbusmenuproxy"
            r"|drkonqi(?:-[a-z-]+)?|xsettingsd)(?:\s|$)",
            # The control centres. They belong here rather than in a row of
            # their own: "settings" is not something this tool can detect - it
            # sees a named binary, and systemsettings is a component of the KDE
            # shell exactly as plasmashell is. A generic Settings row would have
            # to claim that a chess program's preferences window belongs in it,
            # which nothing in /proc supports.
            r"^(?:\S*/)?(?:systemsettings|kcmshell|gnome-control-center)"
            r"[0-9]*(?:\s|$)",
            # Cinnamon: the shell, its session binary and its helpers. Anchored
            # to the END of argv[0], so "/home/me/src/cinnamon/build.sh" is not
            # a desktop.
            r"^(?:\S*/)?cinnamon(?:-[a-z-]+)?(?:\s|$)",
            # The Cinnamon Settings Daemon, the csd-* fork of gsd-*. Both
            # directories are needed: csd-* sit in /usr/bin, except csd-printer,
            # which does not.
            r"^\S*/(?:bin|libexec)/csd-[a-z0-9-]+(?:\s|$)",
            # XApp, the component library Cinnamon, MATE and Xfce share. Its own
            # directory is part of the anchor, which makes this tighter than the
            # csd line above - worth keeping the two apart for that alone.
            r"^\S*/xapps?/xapp-[a-z0-9-]+(?:\s|$)",
            r"^(?:\S*/)?touchegg(?:\s|$)",
            # Evolution Data Server and GNOME Online Accounts. Background
            # infrastructure of a session rather than an application the user
            # opened, which is why they go here and not in a row of their own:
            # "personal data" is a category, and a category is not something
            # /proc can show us.
            r"^\S*/(?:libexec|lib)/(?:evolution-data-server/)?evolution-[a-z-]+(?:\s|$)",
            r"^\S*/(?:libexec|lib)/goa-[a-z-]+(?:\s|$)",
            # The GNOME session manager and its two helpers.
            r"^\S*/libexec/gnome-session-[a-z-]+(?:\s|$)",
            # The snap-to-desktop bridge, which ships as a snap of its own, and
            # the launcher that snap uses. The launcher name is generic enough
            # that it is pinned to a /snap path.
            r"^(?:\S*/)?snapd-desktop-integration(?:\s|$)",
            r"^/snap/\S+/user-session-helper(?:\s|$)",
            # Display managers, replacing a bare "lightdm" that used to swallow
            # "rsync -a /etc/lightdm /backup". Second directory alternative is
            # their private libexec (/usr/lib/*/sddm/sddm-helper,
            # /usr/lib/gdm3/gdm-x-session): those are not in bin or libexec.
            r"^\S*/(?:bin|s?bin|lib(?:64|exec)?|sddm|lightdm|gdm[0-9]?)"
            r"/(?:sddm|gdm|lightdm|greetd|plasmalogin)(?:-[a-z][a-z-]*)?[0-9]*(?:\s|$)",
            # plasma-discover is a software centre, not session furniture, and
            # it has a row of its own further down.
            r"/(bin|lib(64|exec)?)/(start)?plasma[-_](?!discover)",
            r"/org_kde_[A-Za-z]",
            r"/polkit-[a-z]+-authentication-agent",
            r"^(?:\S*/)?(?:ibus|fcitx5?)(?:-[a-z0-9-]+)?(?:\s|$)",
            r"\bpicom\b",
            r"pipewire",
            r"pulseaudio",
            r"wireplumber",
            r"xdg-desktop-portal",
            r"\bblueman",
            r"[-_]applet\b",
            r"applet\.py",
            r"[-_]tray\b",
            r"\bxclip\b",
            r"\bclipit\b",
            r"\bcopyq\b",
            r"\bflameshot\b",
            r"\bspectacle\b",
            r"gnome-screenshot\b",
        ],
        [
            # comm carries the whole Wayland half of the desktop. These programs
            # are started from a session file or a D-Bus service by bare name,
            # so argv[0] has no directory for a cmdline pattern to anchor to,
            # and comm - the executable's basename, truncated by the kernel to
            # fifteen characters - is the only thing that identifies them. The
            # patterns are written to survive that truncation.
            r"^k(ded|globalacceld|walletd|activitymanage|access|runner"
            r"|systemstats|screen_backend|screenlocker_g)",
            # The daemons added below. Closed with $ rather than left open like
            # the line above, because none of these names reaches the kernel's
            # 15-character limit (kunifiedpush-di is the truncation of
            # kunifiedpush-distributor and is exactly 15): an open end would
            # file ~/bin/ksecretd-backup.sh as a desktop.
            r"^k(deconnectd|secretd|nighttimed|iod|smserver|unifiedpush-di"
            r"|cminit|deinit|launcher)[0-9]*$",
            # Display managers, whose cmdline patterns all require a directory.
            # gdmap, lightdmx and sddmconfig are excluded by the terminator.
            r"^(sddm|gdm[0-9]?|lightdm|greetd)(-|$)",
            # gnome-control-center is 20 characters, so the kernel truncates
            # its comm to "gnome-control-c" - shorter than the name the cmdline
            # pattern needs. This is the only control centre that needs a comm
            # pattern; systemsettings and kcmshell are short enough that the
            # cmdline pattern's own ^ branch reaches them.
            r"^gnome-control-c",
            r"^(start)?plasma[-_](?!discover)",
            # The compositors themselves. They are also in SUPERVISOR_RE, and
            # the two must ship together: a classified compositor absorbs every
            # unmatched application on the desktop through the ancestor walk,
            # because a Wayland client is a direct child of its compositor.
            r"^(?:sway|[Hh]yprland|river|niri|wayfire|labwc|weston|dwl|cage)$",
            # Bars and wallpaper daemons. The bars are a few megabytes and ride
            # along; the wallpaper daemons are the reason - they hold a decoded
            # full-resolution buffer per output, which is hundreds of megabytes
            # on a multi-monitor 4K desktop.
            r"^(?:waybar|eww|yambar|ironbar|sfwbar|polybar|mako|dunst|swaybg"
            r"|swww-daemon|hyprpaper|wpaperd|mpvpaper)$",
            # Spelled out rather than a bare ^cosmic-, which would swallow
            # cosmic-term and cosmic-edit from Terminals and Editors.
            r"^cosmic-(?:comp|panel|bg|osd|idle|greeter|launcher|notifica"
            r"|workspa|settings|applet|session|dock)",
        ],
        [],
    ),
    (
        "System services",
        [
            r"^/sbin/init",
            r"systemd",
            r"dbus-daemon",
            r"dbus-broker",
            r"NetworkManager",
            r"polkitd",
            r"udisksd",
            r"upowerd",
            r"accounts-daemon",
            r"rtkit-daemon",
            r"\bgeoclue",
            r"bluetoothd\b",
            r"\bvirt[a-z]*d\b",
            r"\bcupsd\b",
            r"\bsmbd\b",
            r"\bsshd\b",
            r"\bcrond\b",
            r"\bavahi",
            r"wpa_supplicant",
            r"\b(chronyd|ntpd)\b",
            r"irqbalance",
            r"thermald",
            r"\bacpid\b",
            r"\bagetty\b",
            r"spacenavd",
            r"rsyslogd",
            r"journald",
            # D-Bus and systemd --user activate these, so before the supervisor
            # boundary existed they reached this group by inheritance. Named
            # explicitly now: matched on purpose beats matched by luck, and they
            # are session plumbing on every GTK/GNOME/XFCE/MATE desktop, not just
            # on the machine this was written on.
            r"/gvfsd?(-[a-z0-9]+)*\b",
            r"gvfs-[a-z-]+-volume-monitor",
            r"gnome-keyring-daemon",
            r"/at-spi(2)?-[a-z-]+",
            r"/dconf-service\b",
            r"xdg-permission-store",
            r"xdg-document-portal",
            r"/obexd\b",
            r"\(sd-pam\)",
            r"/wsdd\b",
            # Ubuntu/Fedora daemons that were left Unclassified. argv[0]-only,
            # so that an ordinary "vim /usr/libexec/colord" is not a service.
            # Backups are protected by rule order, not by this anchor.
            r"^(?:\S*/)?(?:cups-browsed|colord(?:-sane)?|ModemManager"
            r"|power-profiles-daemon|switcheroo-control|mpris-proxy"
            r"|(?:ana)?cron|ssh-agent)(?:\s|$)",
            # Shebang scripts: the kernel rewrites argv[0] to the interpreter,
            # so the program name is the SECOND word and no argv[0] anchor can
            # reach it. Requiring a python interpreter in front, an absolute
            # path, and a closed list of names is what keeps
            # "vim /usr/bin/networkd-dispatcher" out.
            r"^(?:\S*/)?python[0-9.]*(?: -\w+)* /\S+/"
            r"(?:networkd-dispatcher|unattended-upgrade-shutdown)(?:\s|$)",
            # Seen on two different distributions. The path has the name twice,
            # /usr/libexec/fwupd/fwupd, which the trailing anchor handles.
            r"^\S*/fwupd(?:\s|$)",
            # GNOME's own ssh agent, beside gnome-keyring-daemon above.
            r"^\S*/gcr-ssh-agent(?:\s|$)",
            r"^(?:\S*/)?kerneloops(?:\s|$)",
        ],
        # Only the two names the kernel truncates: for every other name above,
        # an empty cmdline is replaced by comm and the argv[0] patterns match
        # the bare basename on their own.
        [r"^(?:power-profiles-|switcheroo-cont)$"],
        [],
    ),
    (
        "Terminals",
        [
            r"mate-terminal",
            r"gnome-terminal",
            r"\bkonsole\b",
            r"\balacritty\b",
            r"\bkitty\b",
            r"xfce4-terminal",
            # The default terminal on Ubuntu 26.04 and Fedora 40+, where it
            # replaced gnome-terminal. The agent is the process that owns the
            # shell, so the shell reaches this row through the ancestor walk.
            r"^(?:\S*/)?ptyxis(?:-agent)?(?:\s|$)",
            r"\btmux\b",
        ],
        [r"^(tmux.*|screen)$"],
        [],
    ),
    # Desktop applications, the software a machine that is not a developer's
    # runs. Deliberately placed after everything above: a first-match-wins list
    # is only safe to extend at the end, so nothing here can outrank a system or
    # container process. A backup of ~/.local/share/Steam stays "rsync" because
    # the backup rules sit at the very top of the list. Only the main process of
    # an app needs a rule; helpers, Proton and Wine children and Electron
    # zygotes reach their group through the ancestor walk.
    # Two matchers, two jobs. A cmdline pattern ties a bare application
    # name to a directory that really holds executables ("/bin/<name>" covers
    # /usr/bin, /usr/local/bin, ~/.local/bin, Flatpak /app/bin, Snap
    # /snap/x/current/usr/bin and an AppImage's /tmp/.mount_xxx/usr/bin in one
    # go), never to a bare word - "/home/me/steamdeck" and "~/dev/brave-new-world"
    # are not applications. comm carries the other half: .desktop files launch
    # apps by bare name (Exec=gimp-3.2 %U, Exec=thunderbird %u), so argv[0] has
    # no directory at all and only comm identifies the process. comm is the
    # executable's basename truncated to 15 characters, so these patterns are
    # truncation-safe.
    # Steam, Lutris and Heroic launch games and nothing else, so "Games" here is
    # a fact about the software rather than a guess about the user. Their own
    # bundled Wine builds are listed here too, so that a game does not leak into
    # the Wine row: this rule is checked first and first match wins.
    (
        "Games (Steam/Lutris/Heroic)",
        [
            r"/Steam/[^ ]*steam",  # ~/.local/share/Steam/ubuntu12_64/steam
            r"/(bin|games)/steam\b",
            r"/usr/lib/steam/",
            r"steamapps/common/",
            r"compatibilitytools\.d/",  # Proton and GE-Proton builds
            r"SteamLinuxRuntime|pressure-vessel",
            r"/bin/lutris\b",
            r"lutris/runners/",  # ~/.local/share/lutris/runners/wine/lutris-GE/bin/wine
            r"/(bin|opt)/[Hh]eroic\b",
            r"[Hh]eroic[^ /]*\.[Aa]ppImage",
            r"heroic/tools/",  # Heroic's own Wine and Proton builds
            r"com\.heroicgameslauncher",
        ],
        [r"^(steam|steamwebhelper|lutris|heroic)$"],
        [],
    ),
    # Wine is a Windows compatibility layer, not a game. Office suites, CAD,
    # trading terminals and banking clients all run under it, so the row names
    # what the process is instead of guessing what it is for. Games launched
    # through Steam, Lutris or Heroic never reach this rule - the rule above
    # claims them, bundled Wine builds included. Bottles belongs here for the
    # same reason: it is a general-purpose Wine prefix manager.
    (
        "Wine (Windows apps)",
        [
            r"/bin/wine(server|boot|cfg|console|dbg|tricks)?\b",
            r"/bin/bottles\b",
            r"com\.usebottles\.bottles",
        ],
        [
            r"^wine(server|boot|cfg|console|dbg|tricks|64|32|-preloader)?$",
            r"^bottles$",
        ],
        [],
    ),
    # Its own row, not part of "Creative apps": a scene or a render can be the
    # largest thing on the machine, and then the row has to name the app.
    ("Blender", [r"/bin/blender\b", r"/blender[^ /]*/blender\b"], [r"^blender$"], []),
    (
        "Creative apps",
        [
            r"/bin/gimp(-console)?(-[0-9.]+)?\b",  # gimp-3.2, not gimptool
            r"/bin/(inkscape|krita|darktable|kdenlive|audacity|obs)\b",
            r"/(opt|usr/lib)/obs-studio/",
        ],
        [
            r"^gimp(-console)?(-[0-9.]*)?$",
            r"^(inkscape|krita|darktable|kdenlive|audacity|obs)$",
        ],
        [],
    ),
    (
        "Media players",
        [
            r"/bin/(vlc|mpv|spotify)\b",
            r"/(usr/share|opt|app)/spotify/",
            r"spotify-launcher",
        ],
        [r"^(vlc|mpv|spotify)$"],
        [],
    ),
    (
        "Office and documents",
        [
            r"/soffice(\.bin)?\b",  # the real LibreOffice process
            r"/(bin|lib)/libreoffice\b",
            r"/libreoffice/program/",
            r"/opt/onlyoffice/",
            r"/[Dd]esktop[Ee]ditors\b",  # OnlyOffice ships as "desktopeditors"
            r"/bin/(okular|evince|papers|calibre|ebook-(viewer|edit))\b",
        ],
        [r"^(soffice|soffice\.bin|libreoffice|okular|evince|papers|calibre)$"],
        [],
    ),
    (
        "Mozilla Thunderbird",
        [r"/bin/thunderbird\b", r"/thunderbird/thunderbird\b"],
        [r"^thunderbird$"],
        [],
    ),
    (
        "Editors and notes",
        [
            r"/sublime_text\b",
            # Never a bare "zed": /usr/bin/zed is the ZFS event daemon on any
            # box with ZFS. The editor is "zeditor" when packaged, and lives
            # under its own directory when installed by the official script.
            r"/bin/zeditor\b",
            # Official installer: $HOME/.local/zed.app/libexec/zed-editor
            r"/zed\.app/",
            r"/zed-editor\b",
            r"/zed/bin/zed\b",
            r"dev\.zed\.Zed",
            r"/bin/emacs(-[0-9.]+)?\b",
            r"/(bin|opt/[Oo]bsidian|opt/[Ll]ogseq)/(obsidian|logseq)\b",
            r"[Oo]bsidian[^ /]*\.[Aa]ppImage",
        ],
        [
            # comm only for these two: "kate" is also a common first name, so a
            # cmdline pattern would match every path under /home/kate.
            r"^(kate|gedit|gnome-text-edit)",
            r"^(sublime_text|plugin_host|zeditor|obsidian|logseq)",
            r"^emacs(-[0-9.]*)?$",
        ],
        [],
    ),
    (
        "Browsers (Brave/Vivaldi/Edge)",
        [
            r"brave\.com/brave",
            r"/bin/brave(-browser)?\b",
            r"/vivaldi(-snapshot)?/vivaldi",
            r"/bin/vivaldi\b",
            r"/opera(-beta|-developer)?/opera\b",  # never a bare "opera": ~/Music/opera
            r"/bin/opera\b",
            r"/msedge\b",
            r"/bin/microsoft-edge",
            r"zen[-_]browser",
            r"/zen/zen\b",
        ],
        [r"^(brave|vivaldi|msedge|microsoft-edge)", r"^opera(-|$)"],
        [],
    ),
    (
        "Password managers",
        [
            r"/bin/keepassxc(-proxy)?\b",
            r"/opt/Bitwarden/",
            r"/bin/bitwarden(-desktop)?\b",
            r"/opt/1Password/",
            r"/bin/1password\b",
        ],
        [r"^(keepassxc|bitwarden|1password)"],
        [],
    ),
    (
        "Sync and torrent clients",
        [
            r"/bin/(syncthing|nextcloud)\b",
            r"/bin/qbittorrent(-nox)?\b",
            r"/bin/deluge(-gtk|-console|d)?\b",
            r"/bin/transmission-(gtk|qt|daemon|cli|remote)\b",
        ],
        [r"^(syncthing|nextcloud|qbittorrent)", r"^(transmission|deluge)-?"],
        [],
    ),
    # These names are ordinary enough to turn up inside a path being copied.
    # The backup rules at the top of the list are what keeps
    # "borg create ::snap ~/baloo_files" a backup rather than an indexer; the
    # argv[0] anchors below are what keeps "vim ~/baloo_files" out of the row.
    (
        # Often the largest surprise on a GNOME or KDE desktop, and always a
        # surprise, because nobody starts it deliberately.
        "File indexing (Tracker/Baloo)",
        [
            r"(?:^|/)(?:localsearch(?:-[a-z]+)?-3"
            r"|tracker-(?:miner|extract|writeback|store)[a-z0-9-]*)\b",
            r"/(bin|lib(64|exec)?)/baloo(_file|runner|ctl)[a-z_]*",
        ],
        [r"^(baloo_file|baloorunner|tracker-|localsearch)"],
        [],
    ),
    (
        "File managers",
        [
            # dolphin-emu is a GameCube emulator, not the KDE file manager.
            r"/(bin|lib(64|exec)?)/(dolphin(?!-emu)|nautilus|[Tt]hunar"
            r"|nemo(-desktop)?|pcmanfm(-qt)?|krusader)\b",
        ],
        [r"^(nautilus|nemo(-desktop)?|[Tt]hunar|dolphin|pcmanfm(-qt)?|krusader)$"],
        [],
    ),
    (
        "Software center and updates",
        [
            # Both call setproctitle(), which is why argv[0] carries no
            # directory at all and why the capitalisation differs between them:
            # mintUpdate.py sets "mintUpdate", mintinstall.py sets "mintinstall".
            r"^(?:\S*/)?(?:gnome-software|snap-store"
            r"|plasma-discover(?:-[a-z-]+)?|DiscoverNotifier"
            r"|packagekitd|snapd|mintUpdate|mintinstall|update-notifier)(?:\s|$)",
            # The package updater itself, behind its interpreter. Spelled
            # separately from unattended-upgrade-shutdown in System services on
            # purpose: they are two different programs and one pattern for both
            # would put the updater in the wrong row.
            r"^(?:\S*/)?python[0-9.]*(?: -\w+)* /\S+/"
            r"(?:unattended-upgrade|check-new-release-gtk)(?:\s|$)",
        ],
        # plasma-discover is exactly 15 characters; the shipped "plasma-discove"
        # was 14 with a $ after it and could therefore never match.
        [
            r"^(gnome-software|snap-store|plasma-discover|DiscoverNotifie"
            r"|packagekitd|snapd)$"
        ],
        [],
    ),
    (
        # Deliberately not merged into "Virtualization (KVM/VirtualBox)": that
        # row means this machine RUNS virtual machines and is measured in
        # gigabytes, while these agents mean this machine IS one and never grow
        # past a few dozen megabytes. Merging them would answer the question
        # backwards on a guest.
        "Guest agents (VM/cloud)",
        [
            r"/(bin|s?bin|lib(64|exec)?)/VBox(?:Client|DRMClient|Service|Control)"
            r"(?:-all)?(?:\s|$)",
            r"^(?:\S*/)?(?:vmtoolsd|vmware-user(?:-suid-wrapper)?|VGAuthService)(?:\s|$)",
            r"^(?:\S*/)?(?:qemu-ga|spice-vdagentd?)(?:\s|$)",
        ],
        # Named one by one, not a bare ^VBox: the host side has VBoxSVC,
        # VBoxManage and VBoxXPCOMIPCD, and calling those guest agents would
        # announce "this machine is a VM" on the machine running the VMs.
        [
            r"^VBox(Client|DRMClient|Service|Control)",
            r"^(vmtoolsd|VGAuthService|qemu-ga|spice-vdagent)",
        ],
        [],
    ),
    # Last on purpose: this says how a tool was installed, not what it is, so it
    # must not preempt any of the rules above. Whatever it does not catch falls
    # through to the ancestor walk and lands on whichever app spawned it.
    (
        "Tools: npx / uv / pipx",
        [r"/_npx/", r"\.cache/uv/archive", r"/pipx/venvs/"],
        [],
        [],
    ),
)

EXAMPLE_CONFIG = """\
# memgroups - user grouping rules.
# Goes to ~/.config/memgroups.toml (or $XDG_CONFIG_HOME/memgroups.toml).
# Create with: memgroups --init-config

[options]
# data - anonymous memory plus resident file pages that are not program text
#        (default). The extra part is what a hypervisor's guest RAM, an mmap'd
#        database and a memfd live in; without it seven running VMs read as
#        0.78 GiB instead of 20.45 GiB
# anon - anonymous memory only, which is what data was before the file half
# rss  - full resident set (overstates: a shared page counts once per process)
# pss  - proportional set size (fairer for fork families, but without root
#        less than half of the processes are readable)
metric = "data"

# ram       - processes plus tmpfs and kernel memory, percentages of total RAM.
#             Disk cache is not listed: the kernel drops it under pressure,
#             so it is free memory, not consumption
# processes - processes only, 100% = their sum
scope = "ram"

# How many groups to show, the rest is collapsed into one row. 0 shows all.
top = 15

# Display order of the table: size (default), swap, zswap, name.
# Order only - it changes no figure, and it does not change which groups survive
# `top` (that cut is by SIZE + SWAP, so a swapped-out group cannot drop off).
sort = "size"

# Containers of one docker-compose project are grouped together as
# "Docker compose: <project>". A container started by plain `docker run` has no
# compose label and becomes "Docker: <image>".

# Your own groups. Checked FIRST - before the built-in rules and before the
# automatic Docker grouping. So a rule of your own is the only way to pull a
# container out of its Docker group and name it yourself.
# The first match wins.
#
# match           - regexes against the full command line
# match_comm      - regexes against /proc/PID/comm (short name)
# match_container - regexes against the image name, the container name and the
#                   compose project (any of the three matching is enough)
# after_defaults  - true if the rule should come AFTER the built-in ones

# [[group]]
# name = "MCP stack"
# match_container = ["^mcp$"]

# [[group]]
# name = "Heavy JVMs"
# match_comm = ["^java$"]
# after_defaults = true
"""


# --------------------------------------------------------------------------- #
# Data model
# --------------------------------------------------------------------------- #


def docker_group(image: str) -> str:
    return f"{DOCKER_PREFIX}{image}"


def compose_group(project: str) -> str:
    return f"{COMPOSE_PREFIX}{project}"


@dataclass(slots=True, frozen=True)
class Container:
    """A docker container as seen by `docker ps`."""

    name: str = ""
    image: str = ""
    project: str = ""  # com.docker.compose.project label, empty for plain `docker run`

    @property
    def label(self) -> str:
        """Containers of one compose project are one thing - group them as one.

        Otherwise 31 containers of the same stack become 31 rows of 40 MiB each
        and their real 2 GiB never shows up anywhere.
        """
        if self.project:
            return compose_group(self.project)
        return docker_group(self.image or self.name)

    def fields(self) -> tuple[str, ...]:
        return tuple(f for f in (self.image, self.name, self.project) if f)


@dataclass(slots=True)
class Proc:
    pid: int
    ppid: int
    rss: int  # KiB
    anon: int
    # RssFile: file-backed resident memory. Read by aggregate() for the FILE column
    # (--file), by the hint that points at it, and by data_file() for the default
    # metric, which counts the part of it that is not program text. RssShmem is
    # collected for completeness and is not read by the current grouping.
    file: int
    shmem: int
    pss: int
    comm: str
    cmdline: str
    container: Container | None = None
    swap: int = 0  # KiB shown in the SWAP column: SwapPss in pss mode, else VmSwap
    cgroup: str = ""  # cgroup v2 path from /proc/PID/cgroup, "" under cgroup v1
    # Always plain VmSwap, whatever the metric. The zswap split weights use this so
    # every process in a cgroup is weighed on one scale: in pss mode `swap` is
    # SwapPss only where smaps_rollup was readable, which would mix two models.
    # Left at 0 it mirrors `swap`, which is exactly right outside pss mode.
    swap_vm: int = 0
    text_vm: int = 0  # VmExe + VmLib: a virtual CEILING on text, never a size

    def __post_init__(self) -> None:
        if not self.swap_vm:
            self.swap_vm = self.swap

    @property
    def data_file(self) -> int:
        """Resident file pages that are not program text: guest RAM, mmap'd
        databases, memfd.

        VmExe and VmLib are VIRTUAL sizes, so subtracting them from a resident
        figure always takes away at least as much as the resident text really
        is. That makes this a lower bound and never an overstatement, which is
        the only direction the tool is allowed to be wrong in. It also needs
        nothing but /proc/PID/status - smaps is unreadable for exactly the
        processes this exists for, because VirtualBoxVM is not dumpable.
        """
        return max(0, self.file - self.text_vm)

    def metric(self, name: str) -> int:
        if name == "data":
            return self.anon + self.data_file
        if name == "anon":
            return self.anon
        if name == "rss":
            return self.rss
        return self.pss


@dataclass(slots=True)
class Rule:
    name: str
    match: list[re.Pattern[str]] = field(default_factory=list)
    match_comm: list[re.Pattern[str]] = field(default_factory=list)
    match_container: list[re.Pattern[str]] = field(default_factory=list)
    # User rules are checked before the "Docker: <image>" fallback so that own
    # containers can still be grouped by hand. Defaults are checked after it.
    pre_container: bool = False

    def test(self, p: Proc) -> bool:
        if any(rx.search(p.cmdline) for rx in self.match):
            return True
        if any(rx.search(p.comm) for rx in self.match_comm):
            return True
        if p.container and self.match_container:
            fields = p.container.fields()
            if any(rx.search(f) for rx in self.match_container for f in fields):
                return True
        return False


@dataclass(slots=True)
class Row:
    name: str
    size: int  # KiB of RAM, may be negative for the delta row
    procs: int  # -1 for synthetic rows
    swap: int = 0  # KiB swapped out (sum of VmSwap). Never folded into size.
    # Share of the zswap pool charged to this row. A breakdown of the zswap row's
    # size, never extra RAM, so it is never folded into size either.
    zswap: int = 0
    zswap_estimated: bool = False  # True if any part of zswap came from a split
    # LAST on purpose, even though it reads better next to `swap`: aggregate() and
    # apply_top() build Rows positionally through zswap_estimated, and many tests
    # construct Rows positionally too. A field inserted in the middle would silently
    # reinterpret every one of those. The same applies to `file` and `file_max`
    # below: they are appended after swap_estimated, never spliced in next to swap.
    swap_estimated: bool = False  # True when swap contains the tmpfs estimate
    file: int = 0  # summed RssFile of the group's processes; never part of size
    # Largest single RssFile in the group. Never displayed. It is the only file
    # figure with no cross-process double counting, which is exactly what
    # file_hint() needs and what the hint line quotes.
    file_max: int = 0


# --------------------------------------------------------------------------- #
# /proc parsing
# --------------------------------------------------------------------------- #


def _read(path: Path) -> str | None:
    """Read a /proc file, tolerating processes that vanish mid-scan."""
    try:
        return path.read_text(encoding="utf-8", errors="replace")
    except (OSError, ValueError):
        return None


def sanitize(text: str) -> str:
    """Replace control characters so a hostile process's argv/comm cannot forge
    output rows or inject terminal escape sequences."""
    return CONTROL_RE.sub("?", text)


_STATUS_KEYS = {
    "PPid": "ppid",
    "VmRSS": "rss",
    "RssAnon": "anon",
    "RssFile": "file",
    "RssShmem": "shmem",
    "VmSwap": "swap",  # KiB swapped out: NOT resident, never added into a SIZE
    "VmExe": "exe",  # virtual size of the text segment - see Proc.data_file
    "VmLib": "lib",  # virtual size of mapped libraries - see Proc.data_file
}


def parse_status(text: str) -> dict[str, int]:
    """Pull the handful of numbers we need out of /proc/PID/status."""
    out: dict[str, int] = {}
    for line in text.splitlines():
        key, _, rest = line.partition(":")
        field_name = _STATUS_KEYS.get(key)
        if field_name is None:
            continue
        parts = rest.split()
        if not parts:
            continue
        try:
            out[field_name] = int(parts[0])
        except ValueError:
            continue
        if len(out) == len(_STATUS_KEYS):
            break
    return out


def parse_pss(text: str) -> tuple[int, int]:
    """Sum Pss and SwapPss from /proc/PID/smaps_rollup (KiB).

    SwapPss is the proportional counterpart of VmSwap. In PSS mode the SIZE
    column is proportional, so the SWAP column must be too - mixing a
    proportional size with a full-charge swap overstates swap by ~25% on a
    fork-heavy machine.
    """
    pss = swap_pss = 0
    for line in text.splitlines():
        if line.startswith("Pss:"):
            parts = line.split()
            if len(parts) >= 2:
                try:
                    pss += int(parts[1])
                except ValueError:
                    pass
        elif line.startswith("SwapPss:"):
            parts = line.split()
            if len(parts) >= 2:
                try:
                    swap_pss += int(parts[1])
                except ValueError:
                    pass
    return pss, swap_pss


def container_id_from_cgroup(text: str) -> str | None:
    """Extract a docker container id from /proc/PID/cgroup, if any."""
    m = CONTAINER_RE.search(text)
    return m.group(1) if m else None


def parse_cgroup_path(text: str) -> str:
    """The cgroup v2 (unified) path from a /proc/PID/cgroup dump.

    The unified line is "0::<path>" and the path itself may contain ":", hence
    the maxsplit. A pure cgroup v1 box has no such line and yields "".
    """
    for line in text.splitlines():
        parts = line.split(":", 2)
        if len(parts) == 3 and parts[0] == "0" and parts[1] == "":
            return parts[2].strip() or "/"
    return ""


def read_meminfo(procfs: Path) -> dict[str, int]:
    """Parse /proc/meminfo into a name -> KiB mapping."""
    out: dict[str, int] = {}
    text = _read(procfs / "meminfo") or ""
    for line in text.splitlines():
        key, _, rest = line.partition(":")
        parts = rest.split()
        if not parts:
            continue
        try:
            out[key] = int(parts[0])
        except ValueError:
            continue
    return out


def parse_docker_ps(text: str) -> dict[str, Container]:
    """Parse the tab-separated `docker ps` output into id -> Container."""
    out: dict[str, Container] = {}
    for line in text.splitlines():
        parts = line.split("\t")
        if len(parts) < 4:
            continue
        cid, name, image, project = (p.strip() for p in parts[:4])
        if cid:
            out[cid] = Container(name=name, image=image, project=project)
    return out


def docker_containers(enabled: bool = True) -> dict[str, Container]:
    """Map container id -> Container via a single `docker ps` call."""
    if not enabled or shutil.which("docker") is None:
        return {}
    fmt = '{{.ID}}\t{{.Names}}\t{{.Image}}\t{{.Label "' + COMPOSE_LABEL + '"}}'
    try:
        res = subprocess.run(
            ["docker", "ps", "--no-trunc", "--format", fmt],
            capture_output=True,
            text=True,
            errors="replace",  # non-UTF-8 in a container label must not crash us
            timeout=DOCKER_TIMEOUT,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        return {}
    if res.returncode != 0:
        return {}
    return parse_docker_ps(res.stdout)


def collect_procs(
    procfs: Path = Path("/proc"),
    want_pss: bool = False,
    containers: dict[str, Container] | None = None,
) -> tuple[list[Proc], int]:
    """Scan /proc. Returns the processes and how many of them yielded a PSS."""
    containers = containers or {}
    procs: list[Proc] = []
    pss_ok = 0
    try:
        entries = list(os.scandir(procfs))
    except OSError:
        return procs, pss_ok

    for entry in entries:
        if not entry.name.isdigit():
            continue
        pid = int(entry.name)
        base = procfs / entry.name

        status = _read(base / "status")
        if status is None:
            continue
        fields = parse_status(status)
        if not fields.get("rss") and not fields.get("swap"):
            # Kernel thread or already gone. A process with no RSS but non-zero
            # VmSwap is neither: it is entirely evicted, which is exactly the
            # case this tool exists to surface, so it must be kept.
            continue

        cmdline_raw = _read(base / "cmdline") or ""
        # Cap before matching (bounds the per-process regex work), then strip
        # control chars (a process controls its own argv - keep it off the terminal).
        cmdline = sanitize(cmdline_raw.replace("\0", " ").strip()[:CMDLINE_MAX])
        comm = sanitize((_read(base / "comm") or "").strip())
        if not cmdline:
            cmdline = comm

        container = None
        cgroup_path = ""
        cgroup = _read(base / "cgroup")
        if cgroup:
            # One read, two answers: the container id and the unified path used
            # for zswap attribution.
            cgroup_path = parse_cgroup_path(cgroup)
            cid = container_id_from_cgroup(cgroup)
            if cid:
                # No `docker ps` (missing binary, no permission, --no-docker):
                # a short id still identifies the container.
                container = containers.get(cid) or Container(image=cid[:12])

        pss = 0
        swap_vm = fields.get("swap", 0)  # VmSwap: full charge, per mm
        swap = swap_vm
        if want_pss:
            rollup = _read(base / "smaps_rollup")
            if rollup:
                pss, swap_pss = parse_pss(rollup)
                if pss:
                    pss_ok += 1
                    # Keep both columns on the same accounting model. Falls back
                    # to VmSwap when the rollup is unreadable (needs root for
                    # other users' processes).
                    swap = swap_pss

        procs.append(
            Proc(
                pid=pid,
                ppid=fields.get("ppid", 0),
                rss=fields.get("rss", 0),
                anon=fields.get("anon", 0),
                file=fields.get("file", 0),
                shmem=fields.get("shmem", 0),
                pss=pss,
                comm=comm,
                cmdline=cmdline,
                container=container,
                swap=swap,
                cgroup=cgroup_path,
                swap_vm=swap_vm,
                text_vm=fields.get("exe", 0) + fields.get("lib", 0),
            )
        )
    return procs, pss_ok


# --------------------------------------------------------------------------- #
# Classification
# --------------------------------------------------------------------------- #


def _compile(patterns: list[str]) -> list[re.Pattern[str]]:
    return [re.compile(p) for p in patterns]


def _compile_field(value: object, field: str, rule_name: str) -> list[re.Pattern[str]]:
    """Compile a user-supplied pattern list, tolerating bad config.

    A bare string in TOML (`match = "chrome"`) would otherwise be iterated
    character by character and compiled per letter; a malformed regex would
    crash. Warn and skip instead of silently misbehaving.
    """
    if not value:
        return []
    if not isinstance(value, list):
        print(
            f"memgroups: rule '{rule_name}': {field} must be a list of strings, ignoring it",
            file=sys.stderr,
        )
        return []
    out: list[re.Pattern[str]] = []
    for pat in value:
        if not isinstance(pat, str):
            print(
                f"memgroups: rule '{rule_name}': {field} entry {pat!r} is not a string, skipped",
                file=sys.stderr,
            )
            continue
        try:
            out.append(re.compile(pat))
        except re.error as exc:
            print(
                f"memgroups: rule '{rule_name}': invalid regex {pat!r} ({exc}), skipped",
                file=sys.stderr,
            )
    return out


def build_rules(config: dict) -> list[Rule]:
    """User rules first (they override defaults) unless after_defaults is set."""
    defaults = [
        Rule(name, _compile(cmd), _compile(comm), _compile(cont))
        for name, cmd, comm, cont in DEFAULT_RULES
    ]
    before: list[Rule] = []
    after: list[Rule] = []
    for raw in config.get("group", []):
        name = raw.get("name")
        if not name:
            continue
        after_defaults = bool(raw.get("after_defaults"))
        rule = Rule(
            name=name,
            match=_compile_field(raw.get("match"), "match", name),
            match_comm=_compile_field(raw.get("match_comm"), "match_comm", name),
            match_container=_compile_field(raw.get("match_container"), "match_container", name),
            pre_container=not after_defaults,
        )
        (after if after_defaults else before).append(rule)
    return before + defaults + after


def _docker_client_group(p: Proc, by_image: list[tuple[str, str]]) -> str | None:
    """A host-side `docker run <image>` client belongs with its own container."""
    if not by_image or not DOCKER_CLIENT_RE.search(p.cmdline):
        return None
    return next((label for image, label in by_image if image in p.cmdline), None)


def _vm_group(p: Proc) -> str | None:
    """A hypervisor process running one named guest belongs to that guest."""
    if not VM_HOST_RE.search(p.cmdline):
        return None
    for rx in VM_NAME_RE:
        m = rx.search(p.cmdline)
        if m:
            name = m.group(1).strip()[:VM_NAME_MAX]
            if name:
                return f"VM: {name}"
    return None


def classify(
    procs: list[Proc], rules: list[Rule], containers: dict[str, Container] | None = None
) -> dict[int, str]:
    """Assign every process to a group.

    Order: user rules -> container membership -> host-side docker-run client
    adoption -> named virtual machine -> default rules -> ancestor.

    Anything running inside a container becomes "Docker compose: <project>" or,
    without a compose label, "Docker: <image>"; only user rules from the config
    can override that. A host-side `docker run <image>` client is then adopted
    into its container's group. Electron and Chrome helpers carry no application
    name in their cmdline (`/proc/self/exe --type=utility`), so whatever is still
    unmatched inherits the group of its nearest matching ancestor.
    """
    pre = [r for r in rules if r.pre_container]
    rest = [r for r in rules if not r.pre_container]
    # Longest image first, so "mcp/sonarqube:v2" wins over "mcp/sonarqube".
    by_image = sorted(
        {c.image: c.label for c in (containers or {}).values() if c.image}.items(),
        key=lambda kv: len(kv[0]),
        reverse=True,
    )

    by_pid = {p.pid: p for p in procs}
    direct: dict[int, str | None] = {}
    for p in procs:
        hit = next((r.name for r in pre if r.test(p)), None)
        if hit is None and p.container:
            hit = p.container.label
        if hit is None:
            hit = _docker_client_group(p, by_image)
        if hit is None:
            hit = _vm_group(p)
        if hit is None:
            hit = next((r.name for r in rest if r.test(p)), None)
        direct[p.pid] = hit

    result: dict[int, str] = {}
    for p in procs:
        group = direct[p.pid] or _inherit(p, by_pid, direct)
        result[p.pid] = group or UNCLASSIFIED
    return result


def _inherit(p: Proc, by_pid: dict[int, Proc], direct: dict[int, str | None]) -> str | None:
    """Walk up the parent chain, stopping at PID 1, at container boundaries and at
    supervisors.

    The supervisor stop is what keeps an unknown application honest: `systemd
    --user` and the D-Bus daemon start unrelated programs, so a group inherited
    across one is a guess wearing a fact's clothes. Stopping there sends the
    process to Unclassified instead, which is the true answer and the only one
    that can be counted.
    """
    seen = {p.pid}
    cur = by_pid.get(p.ppid)
    while cur is not None and cur.pid not in seen and cur.pid != 1:
        if cur.container != p.container:
            return None
        if SUPERVISOR_RE.search(cur.cmdline):
            return None
        hit = direct.get(cur.pid)
        if hit is not None:
            return hit
        seen.add(cur.pid)
        cur = by_pid.get(cur.ppid)
    return None


# --------------------------------------------------------------------------- #
# zswap attribution
# --------------------------------------------------------------------------- #
#
# The kernel exports no per-process zswap - only per cgroup, in memory.stat.
# So the pool is attributed the other way round: read what each cgroup holds,
# then hand it to the groups whose processes live there. Where a cgroup maps
# onto exactly one group the value is exact; where it is shared, that cgroup's
# own zswap is split among its groups in proportion to their swap, and every
# group that took part in a split is marked estimated.


@dataclass(slots=True)
class Zswap:
    """Everything the ZSWAP column and the JSON need. Empty means "not attributed"."""

    pool_kb: int = 0  # meminfo Zswap - the whole pool, the zswap row's size
    source: str | None = None  # None when attribution did not run
    attributed: dict[str, int] = field(default_factory=dict)  # group -> KiB
    estimated: set[str] = field(default_factory=set)  # groups fed by a split

    @property
    def attributed_kb(self) -> int:
        return sum(self.attributed.values())

    @property
    def unattributed_kb(self) -> int:
        """The pool minus what reached a group. Never clamped: a negative value means
        /proc/meminfo and the cgroup tree disagree, and that must stay visible."""
        return self.pool_kb - self.attributed_kb

    @property
    def visible(self) -> bool:
        """Show the column only when something was actually attributed. Decided on the
        computed total, so --top and --scope cannot influence it."""
        return self.attributed_kb > 0


def cgroup_zswap_bytes(path: Path) -> int | None:
    """The "zswap" line of one memory.stat, in BYTES. None if not an accounting cgroup.

    memory.stat exists exactly when the memory controller is enabled for that cgroup,
    including at the root - memory.current does not, which is why it cannot be used.
    """
    text = _read(path / "memory.stat")
    if text is None:
        return None
    for line in text.splitlines():
        if line.startswith("zswap "):
            try:
                return int(line.split()[1])
            except (IndexError, ValueError):
                return None
    return None


def cgroup_root(cgroupfs: Path) -> Path | None:
    """The unified hierarchy that accounts zswap, or None.

    A missing "zswap " line rejects cgroup v1, memory-on-v1 hybrids and kernels
    before 5.19, where meminfo has Zswap but memcg does not account it.
    """
    for candidate in (cgroupfs, cgroupfs / "unified"):
        if cgroup_zswap_bytes(candidate) is not None:
            return candidate
    return None


def _rel_cgroup(root: Path, dirpath: str) -> str:
    """Path relative to the hierarchy root, normalised the way /proc/PID/cgroup writes it."""
    rel = os.path.relpath(dirpath, root)
    return "/" if rel == "." else "/" + rel


def collect_cgroup_zswap(root: Path) -> dict[str, int]:
    """Walk the hierarchy once: cgroup path -> recursive zswap bytes.

    os.walk visits every directory exactly once, so the paths are already
    unique; followlinks=False keeps a symlinked subtree from being walked into
    a second time under a different name.
    """
    totals: dict[str, int] = {}
    for dirpath, _dirnames, _filenames in os.walk(root, followlinks=False):
        value = cgroup_zswap_bytes(Path(dirpath))
        if value is None:
            continue  # no memory controller here, or the cgroup went away mid-scan
        totals[_rel_cgroup(root, dirpath)] = value
    return totals


def acct_parent(rel: str, totals: dict[str, int]) -> str | None:
    """Nearest ancestor that is itself an accounting cgroup.

    Levels without memory.stat are skipped: their pages are charged further up,
    but a grandchild that does account is still included in that ancestor's total,
    so it has to be subtracted there.
    """
    cur = os.path.dirname(rel) or "/"
    while cur not in totals and cur != "/":
        cur = os.path.dirname(cur) or "/"
    return cur if cur in totals else None


def own_zswap(totals: dict[str, int]) -> dict[str, int]:
    """Turn the recursive totals into disjoint own values, in BYTES.

    cgroup v2 memory.stat includes every descendant, so summing the tree
    overcounts several times over. Subtract EVERY direct accounting child, not
    only the ones holding processes we can see: bytes nobody claims must end up
    in the residual, not smeared over a neighbouring application.
    """
    kids: dict[str, int] = {}
    for rel, value in totals.items():
        if rel == "/":
            continue
        parent = acct_parent(rel, totals)
        if parent is not None:
            kids[parent] = kids.get(parent, 0) + value
    # max(): a child that grew between the two reads would otherwise go negative.
    return {rel: max(0, value - kids.get(rel, 0)) for rel, value in totals.items()}


def resolve_cgroup(path: str, totals: dict[str, int], root: Path) -> str | None:
    """The accounting cgroup a process is charged to, or None.

    The is_dir() probe comes first and is mandatory: without it a path we cannot
    see (cgroup namespace, host cgroupfs bind-mounted into a container) would walk
    up to "/", which always exists, and collect a share of the root's zswap.
    """
    if not path:
        return None
    if not (root / path.lstrip("/")).is_dir():
        return None
    cur = path
    while cur not in totals:
        # Terminate on lack of progress, not on a literal value: os.path.dirname
        # of "//" is "//", so testing for "/" alone loops forever on a path like
        # "0:://" and hangs the tool instead of crashing it.
        parent = os.path.dirname(cur) or "/"
        if parent == cur:
            return None
        cur = parent
    return cur


def split_zswap(own_kb: int, weights: dict[str, int]) -> dict[str, int]:
    """Largest-remainder split of one cgroup's own zswap. Integers only.

    Rounding each share on its own does not sum back to own_kb, and that drift
    would break the one invariant the user is told to trust. An empty result means
    nothing could be attributed - the caller sends the bytes to the residual.
    """
    total_weight = sum(weights.values())
    if total_weight <= 0:
        # No live process claims any of that swap: it belongs to shmem or to
        # processes that already exited. Splitting it equally or by RSS would
        # invent attribution, so the whole amount goes to the residual instead.
        return {}
    parts: dict[str, int] = {}
    remainders: list[tuple[int, str]] = []
    handed_out = 0
    for group in sorted(weights):  # sorted: the whole split is deterministic
        quotient, remainder = divmod(own_kb * weights[group], total_weight)
        parts[group] = quotient
        handed_out += quotient
        remainders.append((remainder, group))
    remainders.sort(key=lambda item: (-item[0], item[1]))
    for remainder, group in remainders[: own_kb - handed_out]:  # 0 <= left < len(weights)
        parts[group] += 1
    return parts


def _self_cgroup_visible(procfs: Path, root: Path) -> bool:
    """Our own cgroup must be findable in the tree we are about to read.

    Guards against a cgroup namespace or a bind mount that shows us paths which
    do not line up with the ones /proc reports. A /proc/self/cgroup we cannot read
    at all tells us nothing either way, so it is not treated as a failure.
    """
    text = _read(procfs / "self" / "cgroup")
    if text is None:
        return True
    path = parse_cgroup_path(text)
    if not path:
        return True
    return (root / path.lstrip("/")).is_dir()


def attribute_zswap(
    procs: list[Proc],
    groups: dict[int, str],
    mem: dict[str, int],
    cgroupfs: Path | None,
    procfs: Path = Path("/proc"),
) -> Zswap:
    """Charge the zswap pool to application groups. See the section comment above."""
    out = Zswap(pool_kb=mem.get("Zswap", 0))
    if out.pool_kb <= 0 or cgroupfs is None:
        return out
    root = cgroup_root(cgroupfs)
    if root is None or not _self_cgroup_visible(procfs, root):
        return out

    totals = collect_cgroup_zswap(root)
    if not totals:
        return out
    own = own_zswap(totals)

    # Membership is strictly direct - one process, one accounting cgroup - so no
    # VmSwap is ever used as a weight at two levels of the tree.
    members: dict[str, list[Proc]] = {}
    resolved: dict[str, str | None] = {}
    for p in procs:
        if p.cgroup not in resolved:
            resolved[p.cgroup] = resolve_cgroup(p.cgroup, totals, root)
        rel = resolved[p.cgroup]
        if rel is not None:
            members.setdefault(rel, []).append(p)

    out.source = ZSWAP_SOURCE
    for rel, own_bytes in own.items():
        own_kb = own_bytes // 1024  # BYTES -> KiB, floored once, before any split
        if own_kb <= 0:
            continue
        here = members.get(rel, ())
        names = {groups[p.pid] for p in here}
        if not names:
            continue  # zswap with no process of ours behind it: residual
        if len(names) == 1:
            # 1:1 cgroup: an exact figure, not marked.
            name = next(iter(names))
            out.attributed[name] = out.attributed.get(name, 0) + own_kb
            continue
        weights: dict[str, int] = {}
        for p in here:
            name = groups[p.pid]
            # swap_vm, never p.swap: in pss mode p.swap is SwapPss only where the
            # rollup was readable, so weighing with it would mix two scales.
            weights[name] = weights.get(name, 0) + p.swap_vm
        # Marked even when the split awards 0 KiB: part of a non-empty pool sat in
        # a cgroup shared with this group, so its figure may be an undercount.
        out.estimated.update(names)
        for name, part in split_zswap(own_kb, weights).items():
            out.attributed[name] = out.attributed.get(name, 0) + part
    return out


# --------------------------------------------------------------------------- #
# tmpfs occupancy
# --------------------------------------------------------------------------- #
#
# tmpfs is swap-backed and its pages belong to the filesystem, not to any
# process, so per-process VmSwap excludes them by definition: several GiB of
# swapped-out /tmp and /dev/shm are in neither column. Total occupancy comes from
# statvfs (the df definition, which counts deleted-but-still-open files that du
# cannot see); the part still resident is meminfo Shmem; the difference is what
# is out on swap.


@dataclass(slots=True)
class Tmpfs:
    """Everything the tmpfs SWAP cell and the JSON need. source None means
    "not measured" - unreadable mount table, or nothing we could statvfs."""

    used_kb: int = 0  # sum over distinct tmpfs filesystems, from statvfs
    shmem_kb: int = 0  # meminfo Shmem: the part of that which is in RAM
    source: str | None = None
    filesystems: int = 0  # how many distinct st_dev were measured

    @property
    def swapped_kb(self) -> int:
        """Total occupancy minus what is resident. Clamped at 0 - Shmem also covers
        shared anonymous memory and SysV/memfd shm, which live in no tmpfs mount, so
        the difference is a LOWER BOUND and can legitimately go negative."""
        return max(0, self.used_kb - self.shmem_kb)

    @property
    def visible(self) -> bool:
        """Decided on the computed figure, so --top and --scope cannot influence it."""
        return self.source is not None and self.swapped_kb > 0


def unescape_mount(field: str) -> str:
    """Decode the four escapes /proc/mounts uses, in a single pass.

    re.sub never rescans its own output, which is the whole point: chained
    str.replace() would decode "\\134" to a backslash and then read the following
    "040" as a second escape, turning a literal "\\040" into a space.
    """
    return MOUNT_ESCAPE_RE.sub(lambda m: MOUNT_ESCAPES[m.group(1)], field)


def tmpfs_mountpoints(text: str) -> list[str]:
    """Mount points of every tmpfs filesystem in a /proc/mounts dump.

    The fstype match is exact. devtmpfs holds only device nodes, ramfs is not
    shmem and can never be swapped, hugetlbfs pages are unswappable and are not in
    Shmem - counting any of them would subtract bytes from a Shmem that never
    contained them.
    """
    out: list[str] = []
    for line in text.splitlines():
        fields = line.split()
        if len(fields) < 3 or fields[2] != "tmpfs":
            continue
        # Unescape AFTER splitting - escaping the space is what makes the split safe.
        out.append(unescape_mount(fields[1]))
    return out


def tmpfs_probe(mountpoint: str) -> tuple[int, int] | None:
    """(st_dev, used KiB) for one mount point, or None when it cannot be read.

    Two syscalls on purpose: statvfs exposes no portable filesystem id in Python
    (f_fsid is 0 on many kernels), so the dedupe key has to come from stat(). One
    try around both: a mount that vanishes between them must be skipped, not crash.
    /run/credentials/* is root-only and another user's /run/user/<uid> is 0700, so
    EACCES is routine - it skips one mount and never ends the scan.
    """
    try:
        st = os.stat(mountpoint)
        vfs = os.statvfs(mountpoint)
    except OSError:
        return None
    used = (vfs.f_blocks - vfs.f_bfree) * vfs.f_frsize
    return st.st_dev, max(0, used) // 1024  # BYTES -> KiB, floored once


def tmpfs_used_by_dev(mountpoints: list[str]) -> dict[int, int]:
    """st_dev -> used KiB. Keyed by device so a filesystem mounted in two places
    (bind mount) is counted exactly once - by mount point a single 4 GiB tmpfs
    bind-mounted once would report its full usage twice."""
    out: dict[int, int] = {}
    for mp in mountpoints:
        # Looked up as a module global on purpose: a probe= default argument would
        # be bound at def time and a test's monkeypatch would silently not apply.
        probed = tmpfs_probe(mp)
        if probed is None:
            continue
        dev, used_kb = probed
        # max(), not first-wins: two probes of one filesystem taken microseconds
        # apart can differ, and max() makes the result independent of mount order.
        out[dev] = max(out.get(dev, 0), used_kb)
    return out


def measure_tmpfs(mem: dict[str, int], procfs: Path = Path("/proc")) -> Tmpfs:
    """Total tmpfs occupancy and how much of it is swapped out.

    Degrades silently to a zeroed Tmpfs() on every failure path - unreadable mount
    table, no tmpfs mounts, nothing statvfs-able - so source is set only when at
    least one filesystem was actually measured. Takes the meminfo dict already read
    by the caller: re-reading it here would mix two snapshots for no gain, and the
    figure is non-atomic by construction anyway.
    """
    text = _read(procfs / "mounts")
    if text is None:
        return Tmpfs()
    points = tmpfs_mountpoints(text)
    if not points:
        return Tmpfs()
    by_dev = tmpfs_used_by_dev(points)
    if not by_dev:
        return Tmpfs()
    return Tmpfs(
        used_kb=sum(by_dev.values()),
        shmem_kb=mem.get("Shmem", 0),
        source=TMPFS_SOURCE,
        filesystems=len(by_dev),
    )


# --------------------------------------------------------------------------- #
# Aggregation
# --------------------------------------------------------------------------- #


def _row_sort_key(r: Row) -> tuple[int, int, str]:
    """Display order: RAM first. Swap only breaks ties, so the SIZE column and the
    bars stay monotone down the table."""
    return (-r.size, -r.swap, r.name)


def sort_rows(rows: list[Row], key: str) -> list[Row]:
    """Display order of the table and of the JSON "groups" array. Nothing else.

    Applied by main() AFTER apply_top, never before: the --top cut ranks by
    SIZE + SWAP on purpose, so a group that is mostly swapped out cannot vanish
    from the table, and apply_top re-sorts by _row_sort_key on the way out - an
    earlier sort would be silently undone.

    Every key is a full tuple ending in the name, so two runs on identical data
    render identically and `watch` does not jitter.
    """
    if key == "swap":
        return sorted(rows, key=lambda r: (-r.swap, -r.size, r.name))
    if key == "zswap":
        return sorted(rows, key=lambda r: (-r.zswap, -r.size, r.name))
    if key == "name":
        return sorted(rows, key=lambda r: r.name)
    return sorted(rows, key=_row_sort_key)


def aggregate(
    procs: list[Proc], groups: dict[int, str], metric: str, zswap: Zswap | None = None
) -> list[Row]:
    sizes: dict[str, int] = {}
    counts: dict[str, int] = {}
    swaps: dict[str, int] = {}
    # Summed unconditionally, not gated on --file: JSON exposes file_kb for every
    # group whatever the flags say, and the cost is one add per process.
    files: dict[str, int] = {}
    file_max: dict[str, int] = {}
    for p in procs:
        name = groups[p.pid]
        sizes[name] = sizes.get(name, 0) + p.metric(metric)
        counts[name] = counts.get(name, 0) + 1
        swaps[name] = swaps.get(name, 0) + p.swap
        files[name] = files.get(name, 0) + p.file
        file_max[name] = max(file_max.get(name, 0), p.file)
    z = zswap or Zswap()
    rows = [
        Row(
            name,
            size,
            counts[name],
            swaps[name],
            z.attributed.get(name, 0),
            name in z.estimated,
            # By keyword, skipping swap_estimated: padding with a positional False
            # would tie these two fields to that flag's position forever.
            file=files[name],
            file_max=file_max[name],
        )
        for name, size in sizes.items()
    ]
    rows.sort(key=_row_sort_key)
    return rows


def kernel_kb(mem: dict[str, int]) -> int:
    """Kernel memory that is never reclaimed. Formula unchanged."""
    return (
        mem.get("SUnreclaim", 0)
        + mem.get("PageTables", 0)
        + mem.get("KernelStack", 0)
        + mem.get("Percpu", 0)
    )


def non_process_rows(
    mem: dict[str, int], zswap_residual_kb: int = 0, tmpfs_swap_kb: int = 0
) -> list[Row]:
    """Memory consumed outside of any process, from /proc/meminfo.

    Page cache, buffers and reclaimable slab are deliberately absent: the kernel
    drops them under pressure, so they are free memory, not consumption. What
    stays is tmpfs/shmem (never reclaimed), the kernel's own footprint and the
    zswap pool. These rows are sorted together with the process groups.

    tmpfs is the only synthetic row that can carry swap, because tmpfs pages are
    swap-backed and belong to the filesystem, not to any process - no VmSwap
    anywhere accounts for them. Its SIZE stays exactly meminfo Shmem: the swapped
    part is not RAM and must never enter a size or a share.

    None of these rows has a FILE figure, deliberately: tmpfs is RssShmem and not
    RssFile, and the kernel and zswap rows have no process behind them at all. They
    keep file == 0, so the FILE column renders "-" for them.
    """
    rows = [
        Row(
            ROW_TMPFS,
            mem.get("Shmem", 0),
            -1,
            swap=tmpfs_swap_kb,
            swap_estimated=tmpfs_swap_kb > 0,
        ),
        Row(ROW_KERNEL, kernel_kb(mem), -1, 0),
    ]
    # Zswap is RAM occupied *right now* by the compressed swap pool: it is not in
    # AnonPages, not in Cached, not in Slab (zsmalloc allocates raw pages), so no
    # other row contains it. Zswapped (its uncompressed size) is NEVER used here -
    # those bytes are already counted, from the owning process side, in the SWAP
    # column. Absent key (no CONFIG_ZSWAP) or an empty pool -> no row at all.
    zswap = mem.get("Zswap", 0)
    if zswap > 0:
        # Its SIZE is the whole pool; its own ZSWAP cell carries only what could not
        # be charged to a group. That is what makes the column a breakdown of this
        # row rather than a second copy of it.
        rows.append(Row(ROW_ZSWAP, zswap, -1, 0, zswap_residual_kb, False))
    return rows


def apply_top(rows: list[Row], top: int) -> list[Row]:
    """Collapse everything past the Nth row into a single "rest" row."""
    if top <= 0 or len(rows) <= top:
        return rows
    # Pick what survives by RAM + swap: a group whose pages were swapped out must not
    # drop off the ranking just because its resident part shrank (idle duplicate JVMs
    # are swapped out first). The tmpfs estimate is part of r.swap, so the tmpfs row
    # now competes for a slot with its swapped-out content included - the same rule,
    # applied to the same kind of memory. Display order below is still by RAM alone.
    ranked = sorted(rows, key=lambda r: (-(r.size + r.swap), -r.size, r.name))
    head, tail = ranked[:top], ranked[top:]
    counted = [r.procs for r in tail if r.procs >= 0]
    rest = Row(
        REST_FMT.format(n=len(tail)),
        sum(r.size for r in tail),
        sum(counted) if counted else -1,
        sum(r.swap for r in tail),
        # Carry the collapsed zswap along, or the column would stop summing to the
        # zswap row - and if that row is itself collapsed, the residual would vanish.
        sum(r.zswap for r in tail),
        any(r.zswap_estimated for r in tail),
        # Keep the "~" on the collapsed cell: sum(r.swap) already carried the tmpfs
        # estimate into "Other groups (n)", and an unmarked cell would present a
        # lower bound as a measurement. The figure also stays at JSON top level, so
        # collapsing never hides it from a consumer.
        any(r.swap_estimated for r in tail),
        # FILE is already a per-process sum, so collapsing rows changes nothing
        # about how wrong it is - sum it like SWAP and let the footer say why.
        sum(r.file for r in tail),
        # A MAX, not a sum: file_max must stay "one process's RssFile" or it stops
        # being the one figure with no double counting in it.
        max((r.file_max for r in tail), default=0),
    )
    merged = head + [rest]
    merged.sort(key=_row_sort_key)
    return merged


def memory_model(mem: dict[str, int]) -> dict[str, int]:
    """Split MemTotal the way the table splits it. Meminfo only - no process data,
    so the result is independent of --metric and of the RssAnon COW overcount.

        MemTotal == free + reclaimable + mapped + model + unaccounted
                                                          (exact, integers)
    """
    total = mem.get("MemTotal", 0)
    free = mem.get("MemFree", 0)
    shmem = mem.get("Shmem", 0)
    # KReclaimable >= SReclaimable on kernels that report non-slab reclaimable pools;
    # max() keeps the identity exact and degrades to SReclaimable when it is absent.
    recl_slab = max(mem.get("SReclaimable", 0), mem.get("KReclaimable", 0))
    cache = mem.get("Buffers", 0) + mem.get("Cached", 0) + recl_slab
    kernel = kernel_kb(mem)
    zswap = mem.get("Zswap", 0)
    # Cached includes Shmem; the table counts Shmem as a consumer, so it must not
    # also be counted as reclaimable cache.
    #
    # Mapped is pages that are in someone's page tables right now. MOST of them are
    # page cache and sit inside Cached, and the old model handed all of Cached to
    # "reclaimable" - which is how a VirtualBox guest's 4.47 GiB of file-backed
    # guest RAM read as droppable disk cache. But Mapped is NOT strictly a subset
    # of Cached: driver-allocated pages with no address_space (io_uring rings,
    # and the vboxdrv allocations behind that very guest RAM) count in
    # NR_FILE_MAPPED while never entering Cached. The clamp below is what keeps
    # that honest, and it is why unaccounted can legitimately go negative on a box
    # where such pages dominate: the model then attributes more than meminfo's
    # cache buckets can fund, and the residual says so instead of hiding it.
    # Shmem is subtracted because mapped shmem IS inside
    # Mapped and Shmem already has its own row and its own subtraction below;
    # counting it here would count it twice. Assuming ALL of Shmem is mapped is the
    # conservative direction: the result is a LOWER BOUND on mapped, short of the
    # truth by exactly the unmapped part of Shmem - at most Shmem itself, 937 MiB
    # (about 0.8% of RAM on the machine this was measured on).
    mapped = max(0, mem.get("Mapped", 0) - shmem)
    # Clamped only for a meminfo that disagrees with itself: a namespaced /proc
    # (lxcfs rewrites Cached from the cgroup and passes Mapped through from the
    # host), or a torn per-cpu vmstat read. Subtracting from Buffers + Cached +
    # SReclaimable rather than from Cached alone also cancels the mmap'd block
    # device case, where the pages are in Mapped but counted under Buffers.
    # unaccounted is the residual below, so the partition stays exact whether or
    # not this clamp fires.
    reclaimable = max(0, cache - shmem - mapped)
    model = mem.get("AnonPages", 0) + shmem + kernel + zswap
    return {
        "mem_total_kb": total,
        "mem_free_kb": free,
        "cache_kb": cache,
        "reclaimable_kb": reclaimable,
        "mapped_kb": mapped,
        "anon_pages_kb": mem.get("AnonPages", 0),
        "tmpfs_kb": shmem,
        "kernel_kb": kernel,
        "zswap_kb": zswap,
        "zswapped_kb": mem.get("Zswapped", 0),
        "swap_total_kb": mem.get("SwapTotal", 0),
        "swap_used_kb": mem.get("SwapTotal", 0) - mem.get("SwapFree", 0),
        "model_kb": model,
        # Computed from the CLAMPED reclaimable, so the residual absorbs a fired
        # clamp by construction and the partition stays exact. Never clamped
        # itself: a negative residual is a torn read and must show as negative.
        "unaccounted_kb": total - free - reclaimable - mapped - model,
    }


def swap_visible(mem: dict[str, int], rows: list[Row]) -> bool:
    """Show the SWAP column only when swap is actually in use. An all-zero column on a
    swapless box is noise. The second clause covers a namespaced /proc (lxcfs) whose
    meminfo hides the host's swap while VmSwap still reports it."""
    return (mem.get("SwapTotal", 0) - mem.get("SwapFree", 0)) > 0 or any(r.swap for r in rows)


# --------------------------------------------------------------------------- #
# Rendering
# --------------------------------------------------------------------------- #


def human(kib: int) -> str:
    sign = "-" if kib < 0 else ""
    value = float(abs(kib))
    for unit in ("KiB", "MiB", "GiB"):
        if value < 1024:
            precision = 0 if unit == "KiB" else 2
            return f"{sign}{value:.{precision}f} {unit}"
        value /= 1024
    return f"{sign}{value:.2f} TiB"


def fit(name: str, width: int, unicode_ok: bool = True) -> str:
    """Pad to width, or cut with a visible marker - never truncate silently."""
    if len(name) <= width:
        return name.ljust(width)
    return name[: width - 1] + ("…" if unicode_ok else "~")


def swap_cell(kib: int, estimated: bool = False) -> str:
    """A leading "~" marks a figure that is an estimate, not a measurement, so an exact
    value and an estimated one can never be confused. An empty cell carries no marker."""
    if kib == 0:
        return "-"
    return ("~" if estimated else "") + human(kib)


def zswap_cell(kib: int, estimated: bool = False) -> str:
    """The ZSWAP column marks proportional splits with the same "~" as SWAP - one
    convention, one implementation."""
    return swap_cell(kib, estimated)


def file_cell(kib: int) -> str:
    """FILE reuses the SWAP/ZSWAP cell format and never carries the "~". The
    number is an exact sum; it is just a sum of numbers that must not be added.
    The "~" means "estimate" in this tool, which would be the wrong warning."""
    return swap_cell(kib)


def use_color(stream) -> bool:
    if os.environ.get("NO_COLOR") is not None:
        return False
    if os.environ.get("TERM", "") == "dumb":
        return False
    return stream.isatty()


def use_unicode(stream) -> bool:
    encoding = (getattr(stream, "encoding", "") or "").lower()
    return "utf" in encoding


def render_table(
    rows: list[Row],
    total: int,
    mem: dict[str, int],
    metric: str,
    scope: str,
    proc_count: int,
    stream=None,
    zswap: Zswap | None = None,
    tmpfs: Tmpfs | None = None,
    # Appended last, with defaults, so every existing positional call site keeps
    # working unchanged.
    show_file: bool = False,
    hint_row: Row | None = None,
) -> None:
    stream = stream if stream is not None else sys.stdout
    color = use_color(stream)
    unicode_ok = use_unicode(stream)
    fill, empty = ("█", "·") if unicode_ok else ("#", ".")

    width = shutil.get_terminal_size((100, 24)).columns
    show_zswap = zswap is not None and zswap.visible
    # A pool that holds swapped-out pages means swap is in use by definition, but
    # keep the two columns together explicitly rather than by inference.
    show_swap = swap_visible(mem, rows) or show_zswap
    longest = max((len(r.name) for r in rows), default=10)
    share_label = "% RAM" if scope == "ram" else "% TOTAL"
    # FILE sits immediately after SIZE, before the share column: the question is
    # "SIZE says 604 MiB, what am I missing", so the answer has to be next to SIZE.
    # Empty strings when the column is off, so the format strings below stay
    # byte-identical to the ones that shipped without it.
    file_extra = 12 if show_file else 0  # 11-wide cell + 1 separator
    file_head = f" {'FILE':>11}" if show_file else ""

    if show_zswap:
        # Five numeric columns no longer leave room for both a wide name and a bar
        # at 80 columns, so the bar is dropped before any number: it is decoration,
        # the numbers are the product. Fixed cost is name + 55, plus whatever the
        # share header needs beyond its 6-char field ("% TOTAL" is 7).
        share_width = max(6, len(share_label))
        fixed = 55 + share_width - 6 + file_extra
        want = min(longest, NAME_WIDTH_MAX)
        with_bar = max(20, min(want, width - fixed - 8))  # 8 = bar floor 6 + 2 separators
        if width - (with_bar + fixed) - 2 >= 6:
            name_width = with_bar
            bar_width = min(28, width - (name_width + fixed) - 2)
        else:
            name_width = max(20, min(want, width - fixed))
            bar_width = 0
        header = (
            f"{'#':>2}  {'GROUP':<{name_width}}  {'SIZE':>12}{file_head} "
            f"{share_label:>{share_width}} {'SWAP':>11} {'ZSWAP':>11} {'PROCS':>5}"
        )
    else:
        # Fixed columns (index, size, %, processes, separators) take 38 chars; the
        # bar needs at least 6. So cap the name at width - 44 (= 38 + 6) and give the
        # bar whatever is left. Names get as much room as they need - docker image
        # names are long and must stay readable. The SWAP column, when shown, costs
        # 13 more (12 wide plus a separator) and is taken out of the name cap first,
        # then out of the bar cap: the bar floor of 6 must never push the line past
        # the terminal width.
        extra = (13 if show_swap else 0) + file_extra
        name_width = max(20, min(longest, NAME_WIDTH_MAX, width - 44 - extra))
        bar_width = max(6, min(28, width - name_width - 38 - extra))
        if show_swap:
            header = (
                f"{'#':>2}  {'GROUP':<{name_width}}  {'SIZE':>12}{file_head} "
                f"{share_label:>7} {'SWAP':>12} {'PROCESSES':>9}"
            )
        else:
            header = (
                f"{'#':>2}  {'GROUP':<{name_width}}  {'SIZE':>12}{file_head} "
                f"{share_label:>7} {'PROCESSES':>9}"
            )

    rule = "-" * min(len(header) + (bar_width + 2 if bar_width else 0), width)
    print(file=stream)
    # No prose lead above the table. It restated row 1 in words and pushed the
    # answer down two lines; the column relationships it was meant to teach are
    # carried by the swap footer's decomposition and by --help. The sentence still
    # earns its place in --processes, where no single line shows the group total.
    print(header, file=stream)
    print(rule, file=stream)

    for i, row in enumerate(rows, 1):
        # Swap never enters the share or the bar: those describe RAM only.
        share = row.size / total if total else 0.0
        procs = "-" if row.procs < 0 else str(row.procs)
        fcell = f" {file_cell(row.file):>11}" if show_file else ""
        if show_zswap:
            middle = (
                f"{human(row.size):>12}{fcell} {share * 100:>{share_width - 1}.1f}% "
                f"{swap_cell(row.swap, row.swap_estimated):>11} "
                f"{zswap_cell(row.zswap, row.zswap_estimated):>11} {procs:>5}"
            )
        elif show_swap:
            middle = (
                f"{human(row.size):>12}{fcell} {share * 100:>6.1f}% "
                f"{swap_cell(row.swap, row.swap_estimated):>12} {procs:>9}"
            )
        else:
            middle = f"{human(row.size):>12}{fcell} {share * 100:>6.1f}% {procs:>9}"
        line = f"{i:>2}  {fit(row.name, name_width, unicode_ok)}  {middle}"
        if bar_width:
            filled = max(0, min(bar_width, round(share * bar_width)))
            line += "  " + fill * filled + empty * (bar_width - filled)
        if color:
            code = PALETTE[(i - 1) % len(PALETTE)]
            line = f"\033[38;5;{code}m{line}\033[0m"
        print(line, file=stream)

    print(rule, file=stream)
    shown = sum(r.size for r in rows)
    shown_share = shown / total if total else 0.0
    # The TOTAL swap cell is the column sum, never SwapTotal - SwapFree: the two
    # are not comparable (COW inflates one, shmem swap deflates it).
    # The rule is "mark what is an estimate", not "mark per column": the TOTAL SWAP
    # cell is marked when any row is estimated, while the TOTAL ZSWAP cell stays
    # unmarked because sum(ZSWAP) equals the zswap row's SIZE by invariant.
    swap_total_estimated = any(r.swap_estimated for r in rows)
    # Printed rather than blanked: hiding it would only make the user add the column
    # by hand and reach the same wrong number with no correction under it. The
    # correction is file_footer's second line.
    ftotal = f" {file_cell(sum(r.file for r in rows)):>11}" if show_file else ""
    if show_zswap:
        print(
            f"{'':>2}  {'TOTAL':<{name_width}}  "
            f"{human(shown):>12}{ftotal} {shown_share * 100:>{share_width - 1}.1f}% "
            f"{swap_cell(sum(r.swap for r in rows), swap_total_estimated):>11} "
            f"{zswap_cell(sum(r.zswap for r in rows)):>11} {proc_count:>5}",
            file=stream,
        )
    elif show_swap:
        print(
            f"{'':>2}  {'TOTAL':<{name_width}}  "
            f"{human(shown):>12}{ftotal} {shown_share * 100:>6.1f}% "
            f"{swap_cell(sum(r.swap for r in rows), swap_total_estimated):>12} {proc_count:>9}",
            file=stream,
        )
    else:
        print(
            f"{'':>2}  {'TOTAL':<{name_width}}  "
            f"{human(shown):>12}{ftotal} {shown_share * 100:>6.1f}% {proc_count:>9}",
            file=stream,
        )
    print(file=stream)
    print(reconcile_line(mem, metric), file=stream)
    # The 5-line footer budget governs the DEFAULT screen. --file is opt-in and its
    # caveat is not optional: FILE double counts shared mappings by ~5x machine-wide,
    # so a user who asked for the column must be told before summing it. Dropping
    # this to hit a line count would trade honesty for tidiness, which is the one
    # trade this tool does not make.
    if show_file:
        print(file_footer(mem, metric, sum(r.file for r in rows)), file=stream)
    # Gated on show_swap: never explain a column that is not on screen. The ZSWAP
    # column has no footer of its own - --help and the README say what it
    # is, and TOTAL's ZSWAP cell equals the zswap row's SIZE on the same screen.
    if show_swap:
        print(swap_footer(sum(r.swap for r in rows), tmpfs, metric, scope, mem), file=stream)
    # Last, because it is not a column footer - the column is off. It is a call to
    # action, so it sits at the bottom where the reader stops.
    if not show_file and hint_row is not None:
        print(file_hint_footer(hint_row, unicode_ok), file=stream)


def swap_source(metric: str) -> str:
    """The instrument behind the SWAP column. Same rule as render_json's swap_source:
    a proportional size gets a proportional swap."""
    return "SwapPss" if metric == "pss" else "VmSwap"


def swap_footer(
    swap_shown_kb: int,
    tmpfs: Tmpfs | None,
    metric: str,
    scope: str,
    mem: dict[str, int] | None = None,
) -> str:
    """Two lines reconciling SWAP column total with meminfo swap used figure.

    The first line breaks down the SWAP column total into components (process swap
    and tmpfs estimate), making explicit the connection between the TOTAL cell above
    and the components shown here.

    The second line explains why the column total and meminfo "swap X/Y" disagree:
    a shared page swapped by a fork family is charged to every process that maps it
    (column runs high), and swap left behind by exited/unattributable processes
    counts nowhere (column runs low). Both effects are real and pull opposite
    directions.
    """
    label = swap_source(metric)
    t = tmpfs or Tmpfs()
    mem = mem or {}
    # In "ram" scope the tmpfs estimate is inside a row and therefore inside the
    # column total; in "processes" scope there is no tmpfs row and it is not.
    in_column = t.swapped_kb if scope == "ram" else 0
    proc_kb = swap_shown_kb - in_column
    # /proc/meminfo has SwapTotal and SwapFree; there is no SwapUsed field. Reading
    # one made this whole comparison dead code - the branch below that names the
    # meminfo figure could never fire, so the line meant to reconcile the two
    # numbers shipped without either of them in it.
    swap_used_kb = max(0, mem.get("SwapTotal", 0) - mem.get("SwapFree", 0))

    if t.source is None:
        first = f"SWAP = {label} {human(proc_kb)}; swapped-out tmpfs could not be measured."
    elif t.swapped_kb <= 0:
        first = f"SWAP = {label} {human(proc_kb)}; no tmpfs is swapped out."
    elif scope == "ram":
        first = (
            f"SWAP {human(swap_shown_kb)} = {label} {human(proc_kb)} + "
            f"~{human(t.swapped_kb)} tmpfs, which {label} cannot see."
        )
    else:
        first = (
            f"SWAP {human(swap_shown_kb)} = {label} {human(proc_kb)}; "
            f"~{human(t.swapped_kb)} tmpfs is in no row, a floor."
        )
    if swap_used_kb > 0:
        second = (
            f"vs meminfo {human(swap_used_kb)}: shared pages count per holder, "
            "ownerless swap uncounted."
        )
    else:
        second = "vs meminfo swap: a shared page counts per holder, other ownerless swap never."
    return f"{first}\n{second}"


def memory_sentence(
    size_kb: int,
    swap_kb: int = 0,
    swap_estimated: bool = False,
    zswap_kb: int = 0,
    zswap_estimated: bool = False,
) -> str:
    """The one sentence that answers "how much is this really costing me".

    Used by --processes. It was briefly printed above the table too, where it only
    restated row 1; the shared builder stays so any future caller cannot drift.

    "of zswap RAM", not "in zswap": the figure is RAM the kernel spends to hold part
    of that swap compressed, not a slice of the swap total. The word RAM makes it a
    cost rather than a share and makes no containment claim, because "of it in
    zswap" would be false.
    """
    out = f": {human(size_kb)}"
    if swap_kb:
        out += f" + {'~' if swap_estimated else ''}{human(swap_kb)} swapped out"
    if zswap_kb:
        out += f", {'~' if zswap_estimated else ''}{human(zswap_kb)} of zswap RAM"
    return out


def file_footer(mem: dict[str, int], metric: str, shown_file_kb: int) -> str:
    """Two lines, printed whenever the FILE column is on: what FILE is relative
    to SIZE (which changes with --metric), and the live size of the double
    count. The second line is a measurement rather than an adjective on
    purpose: "TOTAL 34.22 GiB double counts shared mappings; meminfo Mapped is
    6.33 GiB" on the user's own machine kills the sum-the-column reflex in a way
    no warning does. meminfo Mapped is the right ceiling - it counts every
    mapped page-cache page once, machine-wide, and it also counts mapped shmem,
    which RssFile excludes, so the true unique total is at most Mapped."""
    relation = {
        # Under the default metric SIZE already holds the non-text half of
        # FILE, so the two overlap rather than stack - which is the opposite of
        # the anon case below and worth spelling out.
        "data": "FILE is all file-backed RSS; SIZE only its non-text part. Read one row.",
        "anon": "FILE is file-backed RSS, on top of SIZE (RssAnon). Read one row, not TOTAL.",
        "rss": "FILE is file-backed RSS, already inside SIZE (VmRSS). Read one row, not TOTAL.",
        "pss": "FILE is full-charge file-backed RSS, SIZE is PSS. Read one row, not TOTAL.",
    }.get(metric, "FILE is file-backed RSS per group. Read one row, not TOTAL.")
    mapped = mem.get("Mapped", 0)
    if mapped <= 0:
        return f"{relation}\nShared mappings count once per process, so FILE double counts."
    return (
        f"{relation}\nTOTAL {human(shown_file_kb)} double counts shared mappings; "
        f"meminfo Mapped is {human(mapped)}."
    )


def file_hint(rows: list[Row], metric: str) -> Row | None:
    """The group most likely to be badly understated by SIZE, or None.

    The absolute floor is per-process (file_max), not per-group: a group-level
    floor fires on ordinary desktop noise - 66 node/uv processes each counting the
    same ~20 MiB of interpreter and shared libraries add up to 1.27 GiB that does
    not exist anywhere. file_max cannot be inflated by sharing.

    Gated on "anon" because the test is meaningless elsewhere: under rss FILE is
    inside SIZE so it can never fire, and under pss a proportional SIZE against a
    full-charge FILE would fire constantly.
    """
    if metric != "anon":
        return None
    hits = [r for r in rows if r.file > FILE_HINT_RATIO * r.size and r.file_max > FILE_HINT_MIN_KB]
    return max(hits, key=lambda r: r.file_max) if hits else None


def file_hint_footer(row: Row, unicode_ok: bool = True) -> str:
    """One line: the flag the user has not discovered yet. Quotes file_max, not
    the row's FILE sum - file_max is one process's RssFile, so it carries no
    cross-process double counting and is the largest claim about this group that
    is certainly true. Stays descriptive: "not in SIZE" is a fact about this
    tool, while whether those pages are reclaimable is not something RssFile can
    answer - pinned VM RAM and a mmap'd search index look identical from here."""
    label = fit(row.name, FILE_HINT_NAME_MAX, unicode_ok).rstrip()
    return f"{label}: {human(row.file_max)} file-backed, not in SIZE (--file)."


def reconcile_line(mem: dict[str, int], metric: str) -> str:
    """Three lines, each within 80 columns. Every number comes from memory_model()."""
    m = memory_model(mem)
    total = m["mem_total_kb"]
    used = total - mem.get("MemAvailable", 0)
    lines = [
        # "droppable", not "cache". The whole cache figure was the wrong number
        # to print here: the table's premise is that cache is free memory
        # because the kernel drops it under pressure, and the mapped part of it
        # is not droppable at all - it is in some process's page tables right
        # now. On the host this was measured on, "cache" read 19.82 GiB while
        # the kernel could hand back exactly nothing, because seven VirtualBox
        # guests held all of it. memory_model() already computed the honest
        # figure; this line was printing the one next to it.
        f"RAM {human(total)} | used {human(used)} | "
        f"free {human(m['mem_free_kb'])} | droppable {human(m['reclaimable_kb'])}",
    ]
    second = (
        f"anon {human(m['anon_pages_kb'])} | "
        f"swap {human(m['swap_used_kb'])}/{human(m['swap_total_kb'])}"
    )
    if m["zswap_kb"] > 0:
        # The only place the two zswap numbers meet: the pool holds that much, it is
        # not extra RAM on top of it.
        second += f" | zswap {human(m['zswap_kb'])} holds {human(m['zswapped_kb'])}"
    lines.append(second)
    # Mapped rides on the third line, not the first: line 1 already measures 66
    # columns on a 128 GiB box and "cache X (mapped Y)" measures 84 - over the
    # 80-column budget this function promises. It also belongs here: line 3 is the
    # memory that has no row in the table above, and mapped is now exactly that.
    # Gated like zswap on line 2 - a /proc without a Mapped key must not grow a
    # token that reads "mapped 0 KiB".
    # ... which is also why "data" drops the token: under that metric these pages
    # ARE in the table, inside the application rows, so naming them again here
    # would read as a second, separate quantity.
    show_mapped = m["mapped_kb"] > 0 and metric != "data"
    head = f"mapped {human(m['mapped_kb'])} | " if show_mapped else ""
    # Never clamped: a negative residual (a torn meminfo read) must show as negative.
    pct = (m["unaccounted_kb"] / total * 100) if total else 0.0
    lines.append(
        f"{head}unaccounted {human(m['unaccounted_kb'])} ({pct:.1f}% of RAM) | metric: {metric}"
    )
    return "\n".join(lines)


def render_processes(
    procs: list[Proc],
    groups: dict[int, str],
    name: str,
    metric: str,
    as_json: bool = False,
    zswap: Zswap | None = None,
    show_file: bool = False,
) -> int:
    """Print the processes of a single group. Returns a shell exit code."""
    available = sorted({g for g in groups.values()})
    exact = [g for g in available if g.lower() == name.lower()]
    partial = [g for g in available if name.lower() in g.lower()]
    candidates = exact or partial
    if not candidates:
        print(f"Group '{name}' not found. Available:", file=sys.stderr)
        for g in available:
            print(f"  {g}", file=sys.stderr)
        return 1
    if len(candidates) > 1:
        print(f"Ambiguous name '{name}'. Matches:", file=sys.stderr)
        for g in candidates:
            print(f"  {g}", file=sys.stderr)
        return 1

    target = candidates[0]
    selected = [p for p in procs if groups[p.pid] == target]
    # RAM first, swap as tie-break, pid for determinism. No footprint sort here: the
    # listing is complete, nothing is collapsed, so nothing can hide.
    selected.sort(key=lambda p: (-p.metric(metric), -p.swap, p.pid))
    total = sum(p.metric(metric) for p in selected)
    swap_total = sum(p.swap for p in selected)
    z = zswap or Zswap()
    # Group level only. The kernel exports no per-process zswap, and splitting the
    # group's figure across its PIDs by VmSwap would fabricate precision.
    zswap_kb = z.attributed.get(target, 0)
    zswap_estimated = target in z.estimated

    if as_json:
        payload = {
            "group": target,
            "metric": metric,
            "total_kb": total,
            "total_swap_kb": swap_total,
            "total_zswap_kb": zswap_kb,
            "zswap_estimated": zswap_estimated,
            "processes": [
                {"pid": p.pid, "kb": p.metric(metric), "swap_kb": p.swap, "cmdline": p.cmdline}
                for p in selected
            ],
        }
        json.dump(payload, sys.stdout, ensure_ascii=False, indent=2)
        print()
        return 0

    show_swap = any(p.swap for p in selected)
    unicode_ok = use_unicode(sys.stdout)
    width = shutil.get_terminal_size((100, 24)).columns
    # 32 fixed + 13 per optional numeric column, so the command never runs off
    # the edge when --file adds a fifth field.
    cmd_width = max(30, width - 32 - (13 if show_swap else 0) - (13 if show_file else 0))
    head = target + memory_sentence(total, swap_total, False, zswap_kb, zswap_estimated)
    print(f"\n{head}, {len(selected)} processes\n")
    # --file must reach this view too: the hint tells the user to look here, and
    # arriving with no file figure is exactly where they would give up.
    swap_head = f" {'SWAP':>12}" if show_swap else ""
    file_head = f" {'FILE':>12}" if show_file else ""
    print(f"{'PID':>8} {'SIZE':>12}{swap_head}{file_head}  COMMAND")
    for p in selected:
        cmd = p.cmdline
        if len(cmd) > cmd_width:
            cmd = cmd[: cmd_width - 1] + ("…" if unicode_ok else "~")
        swap_c = f" {swap_cell(p.swap):>12}" if show_swap else ""
        file_c = f" {human(p.file) if p.file else '-':>12}" if show_file else ""
        print(f"{p.pid:>8} {human(p.metric(metric)):>12}{swap_c}{file_c}  {cmd}")
    return 0


def render_json(
    rows: list[Row],
    total: int,
    mem: dict[str, int],
    metric: str,
    scope: str,
    zswap: Zswap | None = None,
    tmpfs: Tmpfs | None = None,
) -> None:
    z = zswap or Zswap(pool_kb=mem.get("Zswap", 0))
    t = tmpfs or Tmpfs()
    payload = {
        "timestamp": int(time.time()),
        "metric": metric,
        "scope": scope,
        # total_kb is the denominator for percentages (MemTotal in "ram" scope),
        # shown_kb is what the listed groups actually add up to.
        "total_kb": total,
        "shown_kb": sum(r.size for r in rows),
        # Swap is a separate figure on purpose: it is not RAM, so it never enters
        # shown_kb, total_kb or percent.
        "shown_swap_kb": sum(r.swap for r in rows),
        # Matches the metric: proportional swap for a proportional size. One helper
        # with the footer, so the label the screen prints and the label the JSON
        # reports can never disagree.
        "swap_source": swap_source(metric),
        # A breakdown of the zswap row, not extra RAM: never part of shown_kb.
        # null source means no attribution was possible (cgroup v1, no zswap, ...).
        "zswap_source": z.source,
        "zswap_pool_kb": z.pool_kb,
        "zswap_attributed_kb": z.attributed_kb,
        "zswap_unattributed_kb": z.unattributed_kb,
        "shown_zswap_kb": sum(r.zswap for r in rows),
        # Swapped-out tmpfs: not RAM, so it enters no size and no percentage, and it
        # is deliberately absent from "reconcile" - memory_model partitions MemTotal
        # and these bytes are not in MemTotal. tmpfs_shmem_kb duplicates
        # meminfo_kb["Shmem"] on purpose, so tmpfs_swap_kb ==
        # max(0, tmpfs_used_kb - tmpfs_shmem_kb) is verifiable from the payload alone.
        "tmpfs_source": t.source,  # null == no measurement was possible
        "tmpfs_used_kb": t.used_kb,  # statvfs, deduped by st_dev
        "tmpfs_shmem_kb": t.shmem_kb,  # the meminfo side of the subtraction
        "tmpfs_filesystems": t.filesystems,
        # The estimate; a lower bound, never negative. Scope matters for consumers:
        # in "ram" scope this same figure is also the tmpfs row's swap_kb and is
        # therefore already inside shown_swap_kb; in "processes" scope there is no
        # tmpfs row, so it is reported here ONLY and shown_swap_kb excludes it.
        "tmpfs_swap_kb": t.swapped_kb,
        "tmpfs_swap_in_shown": scope == "ram",
        "meminfo_kb": mem,
        "reconcile": memory_model(mem),
        # Always emitted, gate or no gate: a consumer must not have to branch on
        # whether swap_kb exists.
        "groups": [
            {
                "name": r.name,
                "kb": r.size,
                "percent": round(r.size / total * 100, 3) if total else 0.0,
                "procs": None if r.procs < 0 else r.procs,
                "swap_kb": r.swap,
                "swap_estimated": r.swap_estimated,
                "zswap_kb": r.zswap,
                "zswap_estimated": r.zswap_estimated,
                # Always emitted, gate or no gate, like swap_kb. Summing these
                # across groups produces a number that means nothing - shared
                # mappings are counted once per process. reconcile.mapped_kb and
                # meminfo_kb["Mapped"] in this same payload are the honest totals,
                # which is why no top-level shown_file_kb exists.
                "file_kb": r.file,
            }
            for r in rows
        ],
    }
    json.dump(payload, sys.stdout, ensure_ascii=False, indent=2)
    print()


# --------------------------------------------------------------------------- #
# Config + CLI
# --------------------------------------------------------------------------- #


def config_path(explicit: str | None = None) -> Path:
    if explicit:
        return Path(explicit).expanduser()
    xdg = os.environ.get("XDG_CONFIG_HOME")
    base = Path(xdg).expanduser() if xdg else Path.home() / ".config"
    return base / "memgroups.toml"


def load_config(path: Path) -> dict:
    if not path.is_file():
        return {}
    if tomllib is None:
        print(
            f"memgroups: {path} ignored, TOML needs Python 3.11+ "
            f"(running {sys.version_info.major}.{sys.version_info.minor}); using defaults",
            file=sys.stderr,
        )
        return {}
    try:
        with path.open("rb") as fh:
            return tomllib.load(fh)
    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
        print(f"memgroups: cannot read {path}: {exc}", file=sys.stderr)
        return {}


def init_config(path: Path) -> int:
    if path.exists():
        print(f"memgroups: {path} already exists, leaving it alone", file=sys.stderr)
        return 1
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(EXAMPLE_CONFIG, encoding="utf-8")
    except OSError as exc:
        print(f"memgroups: cannot create {path}: {exc}", file=sys.stderr)
        return 1
    print(f"Created {path}")
    return 0


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog="memgroups",
        description="Group RAM consumption by application.",
        # The "~" means two different things and no single line on the screen can
        # say so without being vague, so its per-column meaning lives here.
        epilog=(
            "A leading ~ marks a figure that is not a measurement: in SWAP it is the\n"
            "swapped-out tmpfs estimate, a lower bound; in ZSWAP a proportional split."
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--metric", choices=METRICS, default=None)
    parser.add_argument("--scope", choices=("ram", "processes"), default=None)
    parser.add_argument("--top", type=int, default=None, help="0 shows every group")
    parser.add_argument(
        "--sort",
        choices=SORTS,
        default=None,
        help="display order of the group table (default size); --processes ignores it",
    )
    parser.add_argument("--processes", metavar="GROUP", help="list the processes of one group")
    parser.add_argument(
        "--file",
        action="store_true",
        help="add a FILE column: file-backed RSS per group; double counts shared mappings",
    )
    parser.add_argument("--json", action="store_true", help="machine-readable output")
    parser.add_argument("--no-docker", action="store_true", help="do not call docker ps")
    parser.add_argument("--config", metavar="PATH")
    parser.add_argument("--init-config", action="store_true")
    parser.add_argument("--procfs", default=None, help=argparse.SUPPRESS)
    parser.add_argument("--cgroupfs", default=None, help=argparse.SUPPRESS)
    parser.add_argument("--version", action="version", version=f"memgroups {__version__}")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    path = config_path(args.config)
    if args.init_config:
        return init_config(path)

    if args.config and not path.is_file():
        print(f"memgroups: config file {path} not found, using defaults", file=sys.stderr)

    config = load_config(path)
    options = config.get("options", {})
    metric = args.metric or options.get("metric", DEFAULT_METRIC)
    scope = args.scope or options.get("scope", "ram")
    top = args.top if args.top is not None else options.get("top", DEFAULT_TOP)
    sort = args.sort or options.get("sort", "size")

    # CLI flags are validated by argparse; config values are not, so check the
    # merged result. A typo would otherwise crash (top) or silently mislead
    # (metric -> all-zero table, scope -> wrong denominator).
    if metric not in METRICS:
        print(
            f"memgroups: invalid metric {metric!r} (choose {', '.join(METRICS)})", file=sys.stderr
        )
        return 2
    if scope not in SCOPES:
        print(f"memgroups: invalid scope {scope!r} (choose {', '.join(SCOPES)})", file=sys.stderr)
        return 2
    if isinstance(top, bool) or not isinstance(top, int):
        print(f"memgroups: invalid top {top!r} (must be an integer)", file=sys.stderr)
        return 2
    if sort not in SORTS:
        print(f"memgroups: invalid sort {sort!r} (choose {', '.join(SORTS)})", file=sys.stderr)
        return 2

    if scope == "ram" and metric not in RECONCILABLE:
        print(
            f"memgroups: metric '{metric}' cannot be reconciled with /proc/meminfo, "
            "falling back to --scope processes",
            file=sys.stderr,
        )
        scope = "processes"

    procfs = Path(args.procfs or "/proc")
    # A fake /proc matched against the real cgroup tree would attribute the pool to
    # processes that do not exist, so --procfs disables attribution unless the
    # cgroup tree is pointed at a matching snapshot too.
    if args.cgroupfs is not None:
        cgroupfs: Path | None = Path(args.cgroupfs)
    elif args.procfs is not None:
        cgroupfs = None
    else:
        cgroupfs = Path(DEFAULT_CGROUPFS)

    containers = docker_containers(enabled=not args.no_docker)
    procs, pss_ok = collect_procs(procfs, want_pss=(metric == "pss"), containers=containers)
    # A /proc without VmExe/VmLib cannot tell text from data, so "data" would
    # silently become "anon plus all file RSS" - every shared library counted
    # once per process holding it. Fall back to anon, and say so only if the
    # metric was asked for rather than defaulted into.
    if metric == "data" and not any(p.text_vm for p in procs):
        if args.metric == "data" or options.get("metric") == "data":
            print(
                "memgroups: /proc has no VmExe/VmLib, metric 'data' cannot subtract program "
                "text, falling back to --metric anon",
                file=sys.stderr,
            )
        metric = "anon"
    if not procs:
        print(f"memgroups: cannot read {procfs}", file=sys.stderr)
        return 1
    if metric == "pss" and pss_ok < len(procs):
        print(
            f"memgroups: PSS is readable for {pss_ok} of {len(procs)} processes "
            "(the rest need root), the numbers are understated",
            file=sys.stderr,
        )

    rules = build_rules(config)
    groups = classify(procs, rules, containers)
    mem = read_meminfo(procfs)
    zswap = attribute_zswap(procs, groups, mem, cgroupfs, procfs)

    if args.processes:
        return render_processes(procs, groups, args.processes, metric, args.json, zswap, args.file)

    # After the early return: only the table/JSON path consumes this, so a
    # --processes run must not pay for the statvfs calls.
    # No --procfs gating here, unlike zswap: attribution is disabled under --procfs
    # because it would match a FAKE /proc against the REAL cgroup tree, whereas the
    # mount list comes from inside the same fake tree. A fake /proc without a
    # `mounts` file therefore measures nothing, and one with a `mounts` file is a
    # deliberate opt-in by whoever wrote it.
    tmpfs = measure_tmpfs(mem, procfs)

    rows = aggregate(procs, groups, metric, zswap)

    if scope == "ram":
        # Only the part that reached no group goes into the zswap row's cell; the rest
        # of the pool is already spread over the group rows above it.
        residual = zswap.unattributed_kb if zswap.visible else 0
        rows = sorted(rows + non_process_rows(mem, residual, tmpfs.swapped_kb), key=_row_sort_key)
        total = mem.get("MemTotal", 0) or sum(r.size for r in rows)
    else:
        total = sum(r.size for r in rows)

    # Computed BEFORE apply_top: the worst offender is exactly the kind of small-SIZE
    # group that --top collapses, and naming "Other groups (7)" would help nobody.
    hint_row = None if (args.file or args.json) else file_hint(rows, metric)
    rows = apply_top(rows, top)

    # Sort anyway on a machine with nothing to sort by - the -size tie-break makes
    # the result the default order, so scripts and watch loops keep working - but
    # say so on stderr, in the same voice as the --scope fallback above. stdout and
    # --json stay clean.
    if sort == "swap" and not swap_visible(mem, rows):
        print("memgroups: no swap in use, --sort swap falls back to size", file=sys.stderr)
    elif sort == "zswap" and not zswap.visible:
        print("memgroups: no zswap in use, --sort zswap falls back to size", file=sys.stderr)
    rows = sort_rows(rows, sort)

    if args.json:
        render_json(rows, total, mem, metric, scope, zswap, tmpfs)
    else:
        render_table(
            rows,
            total,
            mem,
            metric,
            scope,
            len(procs),
            zswap=zswap,
            tmpfs=tmpfs,
            show_file=args.file,
            hint_row=hint_row,
        )
    return 0


def _run(argv: list[str] | None = None) -> int:
    """Run main(), swallowing the BrokenPipeError from `memgroups | head`."""
    try:
        code = main(argv)
        # Flush inside the guard: with a small output the pipe error would
        # otherwise surface only at interpreter shutdown, past this try.
        sys.stdout.flush()
        return code
    except BrokenPipeError:
        # The reader (head/less) closed the pipe. Redirect stdout to devnull so
        # the interpreter's final flush does not print "Exception ignored".
        try:
            os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
        except OSError:
            pass
        return 141  # 128 + SIGPIPE, the conventional shell exit code


if __name__ == "__main__":
    sys.exit(_run())