#!/usr/bin/python3
# felhom-os-apply — the ROOT half of the agent's operating-system update leg (`11-os-updates.md` §5.4.1, §8.1–8.2).
#
# Install as /usr/local/sbin/felhom-os-apply (0755 root:root). The non-root agent invokes it via `sudo -n`
# (FELHOM_OSAPPLY alias) with EXACTLY:   felhom-os-apply --plan /var/lib/felhom-agent/os/plan-<id>.json
# Nothing else on the command line is accepted. Python 3, standard library only (a JSON plan cannot be parsed
# safely in sh). Tests: configs/test_felhom_os_apply.py (a fake runner; nothing real is executed).
#
# THE TRUST MODEL. The plan is written by the agent, so a broken-into agent writes whatever plan it likes. The
# protection is therefore what this file REFUSES, not where the plan came from: no removal, no downgrade, no new
# package, no package outside the plan, only Debian origin in the fast lane, no kernel / boot package on the host,
# only the box's own customer guest, and the host layer only on a box whose ROOT-OWNED install record says
# "appliance" (a BYO host belongs to its owner, `11` §1). Package signatures stay Debian's: apt checks every Release
# file, including the snapshot.debian.org fallback (decision 79). Nothing here is overridable from the environment.
#
# LAYERS (agent v0.141.0): "guest" (the customer LXC, entered with `pct exec`) and "host" (this Proxmox host, run
# directly). LANE: "fast" for those two. Agent v0.142.0 adds the layer "docker" (the guest's Docker engine set, `11`
# §5.8), which is the SLOW lane: lane "slow" only, the six Docker packages only, origin "Docker CE" only, and only
# with an authority this file checks ITSELF (R3): a signed operator job verified with `ssh-keygen -Y verify` against
# the ROOT-OWNED signers file (TRUST_SIGNERS), bound to this host (TRUST_FILE host_id), unexpired and never replayed;
# or, for an unsigned ring-0 step, the root-owned TRUST_FILE saying `"ring0_slow_lane": true` (set by hand on the demo
# boxes only). The agent's own config is NOT trusted for either: the agent can write it. A Docker step also needs
# `live-restore` ON (R15) — without it every container restarts.
# Agent v0.151.0 (R-812 option A, `09` §3 decision 163) adds the layer "pve": the HOST's Proxmox USERSPACE packages,
# lane "slow" only, origin "Proxmox Debian Repository" only, never a kernel / boot / firmware / microcode name (R14 —
# the kernel is R-836's lane), no removal, no undo, a NEW package only from PVE_NEW_ALLOW, an appliance only (R12), and
# the same authority as the Docker step (R3: a signed `os_pve_step`, or the root-owned ring-0 mark). Select
# "pending-pve" (ring 0): every installed Proxmox-origin package with a pending upgrade, minus HOST_SLOW_RE.
#
# Modes (plan field "mode"):
#   inventory   `apt-get update`, then report what is installed (with origin), what is pending, and health.
#   apply       repair first, pick the packages (select "listed": the plan's name=version list; "pending-fast": every
#               pending Debian / Debian-Security upgrade, for ring 0), check every refusal on an `apt-get -s`
#               simulation of EXACTLY name=version, install, clean, scan for restart-needed, report as inventory.
#   health      report health only (the agent polls it after a run).
#   facts       (v0.142.0) read-only versions for the hub's System page: host Debian, running and next-boot kernel,
#               held packages, kernel taint, the crash guard; guest Debian, Docker engine, containerd, live-restore.
#   live-restore-on  (v0.142.0, layer guest) the ONE-TIME step of `09` decision 87: merge `"live-restore": true`
#               into the guest's /etc/docker/daemon.json and `systemctl reload docker`. NEVER a restart (R-835).
#   oom-check   (R-528, `09` decision 157; layer docker, lane slow) ONLY the memory-kill check below: no apt, no engine
#               change, no authority needed. Wrapper-only — the agent never writes this plan; run it by hand as root.
# R-528 (`09` decision 157): a docker-layer APPLY also runs `oom_check()` after health_after and reports it as
#   "oom_check": {"result": "pass"|"fail"|"error", "oom_killed": bool, "oom_event": bool, "exit_code": int|null,
#                 "image": str|null, "detail": str}
#   — one throwaway container (the controller's own image, no network / volume / port, 64 MB cap) is made to exceed its
#   memory; "pass" only when the engine says OOMKilled=true AND emits the `oom` event. It never changes the step's
#   outcome or health: the hub decides whether the engine set can be approved. Pinned by the OOMCheck tests.
# Output: log lines on stderr and the journal (tag felhom-os-apply); the LAST stdout line is
#   OSAPPLY-REPORT <one JSON object>
# which is what the agent parses. Exit 0 = done; 2 = refused (nothing changed); 3 = failed during install.
#
# SPEED (R-845, agent v0.141.0). Every `pct exec` costs ~0.9 s (measured on demo-hp), and v0.140.0 made one per
# package for version comparisons — 272 packages ≈ 4 minutes. Versions are now compared with the HOST's dpkg (the same
# Debian algorithm), madison/policy run once per pass for all packages, the restart scan runs only after an install,
# and one wrapper call does the whole pass (no separate inventory call before an apply).
import calendar
import json
import os
import re
import stat
import subprocess
import sys
import time

PLAN_DIR = "/var/lib/felhom-agent/os"
PLAN_RE = re.compile(r"^plan-[A-Za-z0-9._-]{1,80}\.json$")
AGENT_USER = "felhom-agent"
FAST_ORIGINS = ("Debian", "Debian-Security")
# Debian package name and version grammar (Debian policy §5.6.1, §5.6.12).
NAME_RE = re.compile(r"^[a-z0-9][a-z0-9+.-]+$")
VERSION_RE = re.compile(r"^(?:[0-9]+:)?[0-9][A-Za-z0-9.+~-]*$")
SNAP_RE = re.compile(r"^[0-9]{8}T[0-9]{6}Z$")
RESERVED_VMIDS = set(range(990000, 990010)) | {9999}
DRIVES_PARENT = "/mnt/felhom-drives"
SNAPSHOT_LIST = "/etc/apt/sources.list.d/felhom-os-snapshot.list"
APT_ENV = ["env", "DEBIAN_FRONTEND=noninteractive", "APT_LISTCHANGES_FRONTEND=none", "NEEDRESTART_MODE=l", "LC_ALL=C"]
DPKG_OPTS = ["-o", "Dpkg::Options::=--force-confold", "-o", "Dpkg::Options::=--force-confdef"]
MIN_FREE = 500 * 1024 * 1024
# R-876: dpkg's state in ONE call — `--audit`, a marker line, then the update journal's file names.
JOURNAL_MARK = "@@FELHOM-DPKG-JOURNAL@@"
DPKG_STATE_SCRIPT = "dpkg --audit; echo " + JOURNAL_MARK + "; ls -A /var/lib/dpkg/updates 2>/dev/null; true"
# The installer's ROOT-OWNED record (felhom-host-install.sh `state_set mode`); the agent cannot write it.
INSTALL_STATE = "/var/lib/felhom-install/state.json"
# R-528: the memory-kill check (oom_check). One 200 MB block under a 64 MB cap: measured on Docker 29.8.2 to be
# OOM-killed with OOMKilled=true and an `oom` event. Every call is bounded: timeouts (the clock read twice) + the
# settle wait stay within 90 s.
OOMCHECK_PREFIX = "felhom-oomcheck-"
OOMCHECK_SCRIPT = "dd if=/dev/zero of=/dev/null bs=200M count=1"
OOMCHECK_TIMEOUTS = {"image": 10, "clock": 5, "run": 30, "inspect": 10, "events": 10, "rm": 15}
# Measured on demo-hp 9201 (Docker 29.8.2, 2026-10-06, audits/readback-2026-10-07/F/F1, F2): an `--until` taken right
# after the run MISSED the oom event although OOMKilled=true; after a 2 s wait and `--until` = guest epoch + 1 it is
# seen. Pinned by test_events_window_ends_after_the_settle_wait.
OOMCHECK_SETTLE = 2
# Kernel, boot and firmware packages are the SLOW lane on the host whatever their origin (`11` C3, §5.2): a host
# reboot is needed for them to take effect, and a bad one can stop the box from booting.
HOST_SLOW_RE = re.compile(r"^(linux-(image|headers|kbuild|modules|base)|proxmox-kernel|proxmox-default-kernel|pve-kernel|"
                          r"pve-firmware|firmware-|grub|shim|systemd-boot|intel-microcode|amd64-microcode|efibootmgr)")
# restart_needed() leaves out processes whose cgroup line matches (grep basic regex). Host: the LXC guests' own
# processes (`0::/lxc/<vmid>/...`) -- NOT lxc-start itself, whose cgroup is `0::/lxc.monitor/<vmid>` (measured
# 2026-10-04 on demo-felhom: the old pattern "lxc" hid lxc-start with 20 deleted maps, so "reboot needed" stayed false
# after a libc6 update). Pinned by test_restart_skip_patterns_against_real_cgroups.
RESTART_SKIP_CGROUP = {"guest": "docker", "host": ":/lxc/"}
HOST_SERVICES = ["pveproxy", "pvedaemon", "pvestatd", "pve-cluster", "felhom-agent"]
# The Docker engine set (`11` §5.8): the only names the docker layer may touch, from the only origin it may use.
DOCKER_NAMES = ("containerd.io", "docker-buildx-plugin", "docker-ce", "docker-ce-cli", "docker-ce-rootless-extras",
                "docker-compose-plugin")
DOCKER_ORIGIN = "Docker CE"
# The Proxmox package lane (R-812 option A): the origin apt prints for download.proxmox.com, and the ONLY new packages
# a pve step may add (measured on demo-felhom 2026-10-07: a full upgrade adds proxmox-firewall-data and the kernel;
# the kernel is refused by R14 whatever this list says).
PVE_ORIGIN = "Proxmox Debian Repository"
PVE_NEW_ALLOW = ("proxmox-firewall-data",)
PVE_SIGNED_OP = "os_pve_step"
# ROOT-OWNED trust anchors (the installer writes them; the demo boxes got them by hand, R-840). Never the agent's config.
TRUST_FILE = "/etc/felhom/os-trust.json"          # {"host_id": "...", "ring0_slow_lane": false}
TRUST_SIGNERS = "/etc/felhom/operator-signers"    # ssh allowed_signers: <key_id> namespaces="felhom-op-v1" <key>
SIG_NAMESPACE = "felhom-op-v1"
SIGNED_OP = "os_docker_step"
NONCE_FILE = "/var/lib/felhom-os-apply/nonces.json"
DAEMON_JSON = "/etc/docker/daemon.json"
# R-858 (v0.142.1): a Docker engine step restarts dockerd, which RECREATES the socket file. With live-restore the
# containers keep running — and one that bind-mounts the socket FILE keeps the deleted inode: measured 2026-10-04 on
# demo-felhom, the controller and traefik were blind to Docker for 1h44m. After a step that installed something, the
# wrapper restarts exactly the containers that mount one of these paths (never the apps, never the engine).
DOCKER_SOCKETS = ("/var/run/docker.sock", "/run/docker.sock")
CRASH_GUARD_STATE = "/var/lib/felhom-crash-guard/state.json"
# ---------- the kernel lane (R-836, `09` §3 decisions 164 + 172, `11` §5.11) ----------
# A new kernel boots ONCE through a flag in a GRUB environment block on the ESP (vfat — GRUB can rewrite it there; on
# the LVM /boot it cannot, R-836). The bundle's two GRUB generators read and clear the flag (01_felhom_oneshot) and
# give each installed kernel a one-shot entry with the lockup-to-panic options (42_felhom_oneshot, option C). The GRUB
# default is the kernel the box RUNS, pinned in KERNEL_DEFAULT_CFG; only "kernel-good" (after a healthy one-shot boot)
# moves it to the new kernel. Measured in the spike on the Tester 1 VM, demo-felhom and demo-hp (Secure Boot on):
# `audits/kernel-spike-2026-10-07/`.
KERNEL_OP = "os_kernel_step"
KERNEL_STATE = "/var/lib/felhom-kernel/state.json"          # 0644 root: the step's phase (the agent reads it)
KERNEL_DEFAULT_CFG = "/etc/default/grub.d/zz-felhom-kernel-default.cfg"  # sourced last: GRUB_DEFAULT = this kernel
ONESHOT_SNIPPET = "/etc/grub.d/01_felhom_oneshot"           # bundle-owned: read + clear the flag, pick the entry
ONESHOT_ENTRIES = "/etc/grub.d/42_felhom_oneshot"           # bundle-owned: the one-shot entries (option C options)
GRUB_CFG = "/boot/grub/grub.cfg"
ESP_MOUNT = "/boot/efi"
ONESHOT_ENV = ESP_MOUNT + "/EFI/felhom/oneshot.env"
ONESHOT_ARGS = "softlockup_panic=1 hardlockup_panic=1 hung_task_panic=1 panic=10"
KERNEL_PIN_FILE = "/etc/kernel/proxmox-boot-pin"            # an operator's `proxmox-boot-tool kernel pin` — never fought
KVER_RE = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+-[0-9]+-pve$")
KERNEL_IMAGE_RE = re.compile(r"^proxmox-kernel-([0-9]+\.[0-9]+\.[0-9]+-[0-9]+-pve)(-signed)?$")
# the packages a kernel step may UPGRADE (never add): the kernel series meta-package, the default-kernel meta, the boot
# helper and the firmware the kernel loads. Everything else in HOST_SLOW_RE (grub, shim, microcode, efibootmgr) stays out.
KERNEL_UPGRADE_RE = re.compile(r"^(proxmox-default-kernel|proxmox-kernel-[0-9]+\.[0-9]+|proxmox-kernel-helper|pve-firmware)$")
KERNEL_MIN_GAP = 20 * 3600     # never two kernel steps in one night
KERNEL_MODES = ("kernel-status", "kernel-reboot", "kernel-boot", "kernel-good", "kernel-revert", "kernel-cancel")
# phases: staged → oneshot → judging → good | reverting → self_reverted | revert_failed; oneshot → fell_back; staged → cancelled
KERNEL_ACTIVE = ("staged", "oneshot", "judging", "reverting")

# ---------- the config bundle (R-840, agent v0.143.0, `11` §5.4.2) ----------
# A signed `agent_config_update` job carries {agent_version, bundle_sha256}; the bundle is ONE JSON file built from this
# repo's configs/ at the agent tag (scripts/build-config-bundle.py) and published beside the binary. The installer
# installs the SAME bundle through this same code (--install-bundle, root only, never reachable through sudo), so a new
# box and an updated box cannot drift. This table is the ONLY set of paths a bundle may write — a signed bundle naming
# any other path is refused (R16) — and it is also the builder's list (one table). Each entry:
#   dest: (source file under configs/, mode, check, policy)
# check:  the content check BEFORE anything is written (R18): sudoers → visudo -cf; sh / bash → -n; python → compile;
#         unit → a [Unit]/[Service]/[Timer] section; unit-nort → that, and never RuntimeDirectory= (the G1 incident);
#         agent-unit → User=felhom-agent; dropin → a [Unit] section (SF-3); nft → nft -c -f; plain → none.
# policy: replace → always written when it differs; if-absent → only when the box has none (a setting the operator may
#         have tuned); oob → only on a box with the OOB belt (/etc/felhom-sshd exists — the installer's --no-oob leaves none).
# Order matters: the sudoers files come LAST, so a referenced wrapper is in place before the line that allows it.
# R-861 (agent v0.146.0): the signed agent update, verified here (mode agent_update), then the A/B flip.
SELFUPDATE_OP = "agent_update"
SELFUPDATE_DIR = "/var/lib/felhom-agent/selfupdate"
SELFUPDATE_WRAPPER = "/usr/local/sbin/felhom-selfupdate-guarded"
# The verified bytes are written HERE (a root-owned directory the agent cannot write) and only this copy reaches the A/B wrapper — never the agent's file.
SELFUPDATE_ROOT_DIR = "/var/lib/felhom-os-apply/agent-update"
SELFUPDATE_MAX_BYTES = 256 * 1024 * 1024
BUNDLE_FORMAT = 1
BUNDLE_OP = "agent_config_update"
BUNDLE_RECORD = "/etc/felhom/config-bundle.json"     # what the box runs (0644 root; the non-root agent reports it)
BUNDLE_PREV_DIR = "/var/lib/felhom-os-apply/bundle-prev"  # the previous copies of every file a bundle replaced
BUNDLE_KEEP = 3
BUNDLE_MAX_BYTES = 4 * 1024 * 1024
BUNDLE_RE = re.compile(r"^bundle-[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.]+)?\.json$")
OOB_DIR = "/etc/felhom-sshd"
BUNDLE_FILES = [
    ("/usr/local/sbin/felhom-mkfs-guarded", "felhom-mkfs-guarded.sh", 0o755, "bash", "replace"),
    ("/usr/local/sbin/felhom-selfupdate-guarded", "felhom-selfupdate-guarded", 0o755, "sh", "replace"),
    ("/usr/local/sbin/felhom-pbs-apply", "felhom-pbs-apply", 0o755, "bash", "replace"),
    ("/usr/local/sbin/felhom-backup-target-apply", "felhom-backup-target-apply", 0o755, "bash", "replace"),
    ("/usr/local/sbin/felhom-os-apply", "felhom-os-apply", 0o755, "python", "replace"),
    ("/usr/local/sbin/felhom-crash-guard", "felhom-crash-guard", 0o755, "python", "replace"),
    # R-861 (agent v0.146.0): the content checker for every agent-staged file a root program reads, and the two FIXED
    # files the agent used to install itself from /tmp (the guest pre-start hook and the shared drive parent).
    ("/usr/local/sbin/felhom-priv-apply", "felhom-priv-apply", 0o755, "python", "replace"),
    ("/var/lib/vz/snippets/felhom-guest-hook.sh", "felhom-guest-hook.sh", 0o755, "sh", "replace"),
    ("/usr/local/sbin/felhom-shared-parent.sh", "felhom-shared-parent.sh", 0o755, "sh", "replace"),
    ("/etc/systemd/system/felhom-shared-parent.service", "felhom-shared-parent.service", 0o644, "unit", "replace"),
    ("/etc/systemd/system/felhom-crash-guard.service", "felhom-crash-guard.service", 0o644, "unit", "replace"),
    ("/etc/systemd/system/felhom-crash-guard-check.service", "felhom-crash-guard-check.service", 0o644, "unit", "replace"),
    ("/etc/systemd/system/felhom-crash-guard-check.timer", "felhom-crash-guard-check.timer", 0o644, "unit", "replace"),
    ("/etc/felhom/crash-guard.conf", "crash-guard.conf", 0o644, "plain", "if-absent"),
    # the kernel lane's two GRUB generators (R-836, `11` §5.11): they take effect at the next update-grub, which the
    # first kernel step runs itself; with no flag on the ESP they change nothing about how the box boots.
    ("/etc/grub.d/01_felhom_oneshot", "felhom-grub-oneshot.sh", 0o755, "sh", "replace"),
    ("/etc/grub.d/42_felhom_oneshot", "felhom-grub-oneshot-entries.sh", 0o755, "sh", "replace"),
    ("/etc/systemd/system/felhom-agent.service", "felhom-agent.service", 0o644, "agent-unit", "replace"),
    ("/etc/systemd/system/felhom-agent-rollback.service", "felhom-agent-rollback.service", 0o644, "unit", "replace"),
    ("/etc/systemd/system/felhom-agent.service.d/felhom-agent-limits.conf", "felhom-agent-limits.conf", 0o644, "dropin", "replace"),
    ("/usr/local/sbin/felhom-mgmt-watchdog", "felhom-mgmt-watchdog.sh", 0o755, "sh", "replace"),
    ("/etc/tmpfiles.d/felhom-privsep.conf", "felhom-privsep.tmpfiles", 0o644, "plain", "replace"),
    ("/etc/systemd/system/felhom-mgmt-watchdog.service", "felhom-mgmt-watchdog.service", 0o644, "unit-nort", "replace"),
    ("/etc/systemd/system/felhom-mgmt-watchdog.timer", "felhom-mgmt-watchdog.timer", 0o644, "unit-nort", "replace"),
    ("/etc/systemd/system/felhom-sshd.service", "felhom-sshd.service", 0o644, "unit-nort", "oob"),
    ("/etc/felhom-oob.nft", "felhom-oob.nft", 0o644, "nft", "oob"),
    ("/etc/systemd/system/felhom-oob-nft.service", "felhom-oob-nft.service", 0o644, "unit", "oob"),
    ("/etc/sudoers.d/felhom-op", "felhom-op.sudoers", 0o440, "sudoers", "oob"),
    ("/etc/sudoers.d/felhom-agent", "felhom-agent.sudoers", 0o440, "sudoers", "replace"),
]
BUNDLE_DESTS = {e[0]: e for e in BUNDLE_FILES}
# The trust root is never a bundle's to change (R17): who may sign is decided by these files, so no bundle may carry them.
TRUST_PATHS = (TRUST_FILE, TRUST_SIGNERS, BUNDLE_RECORD)
# When /etc/felhom/operator-signers is MISSING (a box installed before installer 1.30.0), the job is verified against —
# and the file is created with — exactly this key: the operational key felhom-host-install.sh pins
# (OPERATOR_KEY_OPERATIONAL_*). Never any other. Pinned equal to the installer by
# test_pinned_operator_key_equals_the_installers. Signer rotation is a separate, later act (`04` §3).
PINNED_OPERATOR_KEY_ID = "felhom-op-1"
PINNED_OPERATOR_KEY_LINE = ("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIL8z0qCNgA3x2xxAB0Qj5ro8waFjGZ8Ta/sWB63tlLw+ felhom-op-1")
# The self-check (after install, before the record): the route itself must survive — a bundle whose sudoers no longer
# lets the agent call this wrapper would cut the box off from every later bundle.
SELF_CHECK_SUDO_LINE = "/usr/local/sbin/felhom-os-apply --plan /var/lib/felhom-agent/os/plan-*.json"
PINNED_SIGNERS = "<pinned operator key>"   # verify_sig: check against the pinned key, not a file on the box


def pinned_signers_line():
    """The ssh allowed_signers line the installer writes (felhom-host-install.sh _write_slow_lane_trust), byte for byte."""
    return f'{PINNED_OPERATOR_KEY_ID} namespaces="{SIG_NAMESPACE}" {PINNED_OPERATOR_KEY_LINE}\n'


def sha256_hex(data):
    import hashlib
    return hashlib.sha256(data).hexdigest()


class Refused(Exception):
    def __init__(self, code, reason):
        super().__init__(f"{code} {reason}")
        self.code, self.reason = code, reason


class Runner:
    """Runs commands for real. Tests replace it with a fake. `guest` runs inside the container via pct exec."""

    def host(self, argv, timeout=600, stdin=None):
        p = subprocess.run(argv, capture_output=True, text=True, timeout=timeout, input=stdin)
        return p.returncode, p.stdout, p.stderr

    def guest(self, vmid, argv, timeout=1800):
        return self.host(["/usr/sbin/pct", "exec", str(vmid), "--"] + argv, timeout)

    def read_file(self, path):
        with open(path) as f:
            return f.read()

    def stat(self, path):
        return os.lstat(path)

    def agent_uid(self):
        import pwd
        return pwd.getpwnam(AGENT_USER).pw_uid

    def write_file(self, layer, vmid, path, body):
        """Write a small text file in the target layer — never via a shell string."""
        if layer == "host":
            with open(path, "w") as f:
                f.write(body)
            return
        rc, _, _ = self.host(["/usr/sbin/pct", "exec", str(vmid), "--", "tee", path], 60, stdin=body)
        if rc != 0:
            raise Refused("R7", f"could not write {path} in the guest")

    def save_report(self, plan_path, report):
        """R-868: keep an apply pass's report on disk until the agent has sent it (the agent deletes it). Written
        INTO the agent's own plan dir as root, so: the dir is opened with O_NOFOLLOW and must be a real directory
        owned by the agent (a symlink swapped in for it is refused); the file is created O_EXCL|O_NOFOLLOW after
        removing an old one, then handed to the agent (0600). Any failure only loses the copy — never the run."""
        base = os.path.basename(plan_path)
        name = "report-" + base[len("plan-"):]
        dfd = os.open(PLAN_DIR, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        try:
            st = os.fstat(dfd)
            if not stat.S_ISDIR(st.st_mode) or st.st_uid != self.agent_uid():
                raise OSError(f"{PLAN_DIR} is not the agent's own directory")
            try:
                os.unlink(name, dir_fd=dfd)
            except FileNotFoundError:
                pass
            fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=dfd)
            try:
                os.write(fd, (json.dumps(report, sort_keys=True) + "\n").encode())
                os.fchown(fd, self.agent_uid(), -1)
            finally:
                os.close(fd)
        finally:
            os.close(dfd)
        return os.path.join(PLAN_DIR, name)

    def now(self):
        return time.time()

    def sleep(self, s):
        time.sleep(s)

    def verify_sig(self, signers, key_id, namespace, blob, sig):
        """`ssh-keygen -Y verify` over the EXACT signed bytes. Files in a root-only temp dir; nothing via a shell."""
        import tempfile
        with tempfile.TemporaryDirectory(prefix="felhom-os-apply-") as d:
            sp = os.path.join(d, "sig")
            with open(sp, "w") as f:
                f.write(sig)
            if signers == PINNED_SIGNERS:
                signers = os.path.join(d, "allowed_signers")
                with open(signers, "w") as f:
                    f.write(pinned_signers_line())
            p = subprocess.run(["ssh-keygen", "-Y", "verify", "-f", signers, "-I", key_id, "-n", namespace, "-s", sp],
                               input=blob, capture_output=True, timeout=30)
            return p.returncode

    def read_nonces(self):
        try:
            with open(NONCE_FILE) as f:
                d = json.load(f)
            return d if isinstance(d, dict) else {}
        except (OSError, ValueError):
            return {}

    def write_nonces(self, d):
        os.makedirs(os.path.dirname(NONCE_FILE), mode=0o700, exist_ok=True)
        tmp = NONCE_FILE + ".tmp"
        with open(tmp, "w") as f:
            json.dump(d, f)
        os.replace(tmp, NONCE_FILE)

    def log(self, line):
        # R-868 (v0.144.1): the agent that reads stderr may be GONE (killed mid-pass, measured live on demo-hp
        # 2026-10-05): the write then raises BrokenPipeError, and v0.144.0 died right there — after apt had installed
        # everything, before the report copy was saved. A dead reader must never stop the pass; the journal still
        # gets every line. Pinned by AgentDiesMidPass.
        try:
            print(line, file=sys.stderr, flush=True)
        except OSError:
            pass
        try:
            subprocess.run(["logger", "-t", "felhom-os-apply", line], timeout=10)
        except Exception:
            pass

    def read_staged_once(self, path, owner_uid, limit):
        """R-861: read a file the AGENT staged ONCE, as root, safely: O_NOFOLLOW (the last component may not be a
        symlink), fstat on the opened fd (a regular file owned by owner_uid), at most `limit` bytes. The caller hashes
        and uses exactly these bytes — never the path again (the agent owns the directory and could swap the file)."""
        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
        try:
            st = os.fstat(fd)
            if not stat.S_ISREG(st.st_mode) or st.st_uid != owner_uid:
                raise Refused("R19", f"{path} is not a regular file owned by {AGENT_USER}")
            if st.st_size > limit:
                raise Refused("R19", f"{path} is larger than {limit} bytes")
            chunks, n = [], 0
            while True:
                b = os.read(fd, 1 << 20)
                if not b:
                    break
                chunks.append(b)
                n += len(b)
                if n > limit:
                    raise Refused("R19", f"{path} grew past {limit} bytes while it was read")
            return b"".join(chunks)
        finally:
            os.close(fd)

    # ---------- host files, for the config bundle (R-840). Tests replace these with an in-memory tree. ----------
    def read_bytes(self, path):
        with open(path, "rb") as f:
            return f.read()

    def lexists(self, path):
        return os.path.lexists(path)

    def isdir(self, path):
        return os.path.isdir(path)

    def put_file(self, path, data, mode):
        """Atomic: a root-owned temp file beside the target (same filesystem), fsync, then rename over it."""
        d = os.path.dirname(path)
        os.makedirs(d, mode=0o755, exist_ok=True)
        tmp = os.path.join(d, f".{os.path.basename(path)}.felhom-new.{os.getpid()}")
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        try:
            with os.fdopen(fd, "wb") as f:
                f.write(data)
                f.flush()
                os.fchown(f.fileno(), 0, 0)
                os.fchmod(f.fileno(), mode)
                os.fsync(f.fileno())
            os.replace(tmp, path)
        except BaseException:
            try:
                os.remove(tmp)
            except OSError:
                pass
            raise

    def remove(self, path):
        os.remove(path)

    def list_dir(self, path):
        try:
            return sorted(os.listdir(path))
        except OSError:
            return []

    def rmtree(self, path):
        import shutil
        shutil.rmtree(path, ignore_errors=True)

    def check_content(self, kind, data):
        """Run an external syntax check on a root-only temp copy. Returns (rc, message)."""
        import tempfile
        cmd = {"sudoers": ["visudo", "-cf"], "sh": ["sh", "-n"], "bash": ["bash", "-n"], "nft": ["nft", "-c", "-f"]}[kind]
        with tempfile.TemporaryDirectory(prefix="felhom-bundle-") as d:
            p = os.path.join(d, "f")
            with open(p, "wb") as f:
                f.write(data)
            r = subprocess.run(cmd + [p], capture_output=True, text=True, timeout=60)
            return r.returncode, (r.stdout + r.stderr).strip()[-300:]


class Apply:
    def __init__(self, runner, plan_path):
        self.r = runner
        self.plan_path = plan_path
        self.report = {"refused": None, "mode": None}

    # ---------- checks ----------
    def load_plan(self):
        p = self.plan_path
        d, base = os.path.dirname(p), os.path.basename(p)
        if d != PLAN_DIR or not PLAN_RE.match(base) or ".." in p:
            raise Refused("R1", f"the plan must be {PLAN_DIR}/plan-<id>.json, got {p!r}")
        try:
            st = self.r.stat(p)
        except OSError as e:
            raise Refused("R1", f"cannot stat the plan: {e}")
        if not stat.S_ISREG(st.st_mode):
            raise Refused("R1", "the plan is not a regular file (a symlink or a device is refused)")
        if st.st_uid != self.r.agent_uid():
            raise Refused("R1", f"the plan is not owned by {AGENT_USER}")
        if st.st_size > 2 * 1024 * 1024:
            raise Refused("R1", "the plan is larger than 2 MB")
        try:
            plan = json.loads(self.r.read_file(p))
        except (OSError, ValueError) as e:
            raise Refused("R1", f"the plan is not valid JSON: {e}")
        if not isinstance(plan, dict):
            raise Refused("R1", "the plan is not a JSON object")
        return plan

    def check_plan(self, plan):
        mode = plan.get("mode", "apply")
        if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle", "agent_update", "oom-check") \
                + KERNEL_MODES:
            raise Refused("R11", f"unknown mode {mode!r}")
        if mode == "agent_update":
            if plan.get("layer") != "host":
                raise Refused("R11", "agent_update is a host-layer mode")
            return mode, "host", 0, "agent_update"
        if mode == "bundle":
            if plan.get("layer") != "host":
                raise Refused("R11", "bundle is a host-layer mode")
            return mode, "host", 0, "bundle"
        layer = plan.get("layer")
        if layer not in ("guest", "host", "docker", "pve", "kernel"):
            raise Refused("R12", f"layer {layer!r} is not guest, host, docker, pve or kernel")
        if (mode in KERNEL_MODES) != (layer == "kernel" and mode not in ("apply", "health")):
            raise Refused("R11", f"mode {mode!r} does not fit layer {layer!r}")
        lane = plan.get("lane", "fast")
        if layer == "docker" and lane != "slow":
            raise Refused("R3", "the Docker engine is the slow lane (`11` §5.8); a fast-lane Docker plan is refused")
        if layer == "pve" and lane != "slow":
            raise Refused("R3", "the Proxmox packages are the slow lane (`11` §5.10); a fast-lane pve plan is refused")
        if layer == "kernel" and lane != "slow":
            raise Refused("R3", "the kernel is the slow lane (`11` §5.11); a fast-lane kernel plan is refused")
        if layer not in ("docker", "pve", "kernel") and lane != "fast":
            raise Refused("R3", f"the {layer} layer has no slow lane in this release (kernel, Proxmox: `11` §8 step 6)")
        if mode == "facts" and layer != "host":
            raise Refused("R11", "facts is a host-layer mode (it reads the host and the guest)")
        if mode == "live-restore-on" and layer != "guest":
            raise Refused("R11", "live-restore-on is a guest-layer mode")
        if mode == "oom-check" and layer != "docker":
            raise Refused("R11", "oom-check is a docker-layer mode (it checks the guest's Docker engine)")
        if plan.get("undo") and layer != "docker":  # the pve layer has no undo in this release (R-812 option A)
            raise Refused("R5", "an undo (downgrade) exists only for the Docker layer, inside a signed job")
        vmid = plan.get("vmid")
        if layer == "kernel" and mode in KERNEL_MODES and mode != "kernel-reboot" and vmid == 0:
            pass  # after a boot the guest may not run (that is what is judged); these modes never touch it
        elif not isinstance(vmid, int) or isinstance(vmid, bool) or vmid <= 0:
            raise Refused("R11", f"vmid must be a positive integer, got {vmid!r}")
        rid = plan.get("release_id", "")
        if not isinstance(rid, str) or not re.match(r"^[A-Za-z0-9._:-]{1,80}$", rid):
            raise Refused("R11", f"release_id {rid!r} is not a plain id")
        if plan.get("allow_new"):
            raise Refused("R6", "allow_new is a slow-lane field; the fast lane never adds a package")
        select = plan.get("select", "listed")
        if select not in ("listed", "pending-fast", "pending-docker", "pending-pve", "pending-kernel"):
            raise Refused("R11", f"unknown select {select!r}")
        if (select == "pending-docker") != (layer == "docker" and select != "listed"):
            if select == "pending-docker" or layer == "docker":
                raise Refused("R11", f"select {select!r} does not fit layer {layer!r}")
        if (select == "pending-pve") != (layer == "pve" and select != "listed"):
            raise Refused("R11", f"select {select!r} does not fit layer {layer!r}")
        if (select == "pending-kernel") != (layer == "kernel" and select != "listed"):
            raise Refused("R11", f"select {select!r} does not fit layer {layer!r}")
        ek = plan.get("expect_kver")
        if ek is not None and (layer != "kernel" or not isinstance(ek, str) or not KVER_RE.match(ek)):
            raise Refused("R11", f"expect_kver {ek!r} is not a kernel version of the kernel layer")
        pk = plan.get("packages", [])
        if not isinstance(pk, list):
            raise Refused("R11", "packages must be a list")
        if mode == "apply" and select == "listed" and not pk:
            raise Refused("R11", "packages must be a non-empty list in apply mode (select listed)")
        if select in ("pending-fast", "pending-docker", "pending-pve", "pending-kernel") and pk:
            raise Refused("R11", f"select {select} takes no package list")
        seen = set()
        for e in pk:
            if not isinstance(e, dict):
                raise Refused("R11", "every package entry must be an object")
            n, v, o = e.get("name"), e.get("version"), e.get("origin")
            if not isinstance(n, str) or not NAME_RE.match(n):
                raise Refused("R11", f"package name {n!r} is not a Debian package name")
            if not isinstance(v, str) or not VERSION_RE.match(v):
                raise Refused("R11", f"version {v!r} of {n} is not a Debian version string")
            if n in seen:
                raise Refused("R11", f"package {n} is named twice")
            seen.add(n)
            if layer == "docker":
                if n not in DOCKER_NAMES or o != DOCKER_ORIGIN:
                    raise Refused("R2", f"{n} ({o!r}) is not one of the six Docker packages from {DOCKER_ORIGIN!r}")
                continue
            if n in DOCKER_NAMES:
                raise Refused("R2", f"{n} is a Docker package — the slow lane (`11` §5.8), never in a {layer} plan")
            if layer == "kernel":
                if o != PVE_ORIGIN:
                    raise Refused("R2", f"{n}: origin {o!r} is not {PVE_ORIGIN!r} (the kernel layer)")
                if not (KERNEL_UPGRADE_RE.match(n) or KERNEL_IMAGE_RE.match(n)):
                    raise Refused("R23", f"{n} is not a kernel package (a kernel image, the kernel meta-packages, "
                                         f"proxmox-kernel-helper or pve-firmware)")
                continue
            if layer == "pve":
                if o != PVE_ORIGIN:
                    raise Refused("R2", f"{n}: origin {o!r} is not {PVE_ORIGIN!r} (the pve layer)")
                if HOST_SLOW_RE.match(n):
                    raise Refused("R14", f"{n} is a kernel / boot / firmware package — never the pve lane (R-836)")
                continue
            if o not in FAST_ORIGINS:
                raise Refused("R2", f"{n}: origin {o!r} is not Debian / Debian-Security (the fast lane, `11` C3)")
            if layer == "host" and HOST_SLOW_RE.match(n):
                raise Refused("R14", f"{n} is a kernel / boot / firmware package — the host's slow lane")
        snap = plan.get("snapshot", "")
        if snap and not SNAP_RE.match(snap):
            raise Refused("R11", f"snapshot {snap!r} is not YYYYMMDDTHHMMSSZ")
        return mode, layer, vmid, select

    def check_appliance(self):
        """R12: the host layer only on a box whose ROOT-OWNED install record says appliance (`11` §1: never BYO)."""
        try:
            st = self.r.stat(INSTALL_STATE)
        except OSError:
            raise Refused("R12", f"no install record ({INSTALL_STATE}) — this box cannot prove it is an appliance")
        if st.st_uid != 0 or (st.st_mode & 0o022):
            raise Refused("R12", f"{INSTALL_STATE} is not root-owned and root-only-writable — it proves nothing")
        try:
            mode = json.loads(self.r.read_file(INSTALL_STATE)).get("mode")
        except (OSError, ValueError, AttributeError):
            raise Refused("R12", f"{INSTALL_STATE} is unreadable — this box cannot prove it is an appliance")
        if mode != "appliance":
            raise Refused("R12", f"this box was installed as {mode!r}, not appliance — its host belongs to its owner")

    def load_trust(self):
        """The ROOT-OWNED trust record. Absent or agent-writable → no slow-lane authority at all (R3)."""
        try:
            st = self.r.stat(TRUST_FILE)
        except OSError:
            raise Refused("R3", f"no {TRUST_FILE} — this box has no slow-lane trust anchor")
        if st.st_uid != 0 or (st.st_mode & 0o022):
            raise Refused("R3", f"{TRUST_FILE} is not root-owned and root-only-writable — it proves nothing")
        try:
            t = json.loads(self.r.read_file(TRUST_FILE))
        except (OSError, ValueError):
            raise Refused("R3", f"{TRUST_FILE} is unreadable")
        if not isinstance(t, dict) or not isinstance(t.get("host_id"), str) or not t["host_id"]:
            raise Refused("R3", f"{TRUST_FILE} names no host_id")
        return t

    def verify_signed(self, signed, trust, op_name=SIGNED_OP, signers=None, burn=True):
        """R3: an operator-signed job, checked HERE (not by the agent): signature against the root-owned signers file (or,
        for a bundle on a box that has none, the PINNED key — `signers` names that temp file), op, host binding, time
        window, and a root-owned nonce record (no replay). burn=False leaves the nonce for the caller to burn after its
        own checks (a bundle refused for a wrong sha keeps its job usable — the operator fixes the sha, not the key).
        Returns the params; with burn=False, (params, nonce, expiry)."""
        import base64
        if not isinstance(signed, dict) or not isinstance(signed.get("blob_b64"), str) or not isinstance(signed.get("sig"), str):
            raise Refused("R3", "the signed job is malformed")
        try:
            blob = base64.b64decode(signed["blob_b64"], validate=True)
            op = json.loads(blob)
        except (ValueError, TypeError):
            raise Refused("R3", "the signed blob is not base64 JSON")
        key_id = op.get("key_id", "")
        if not isinstance(key_id, str) or not re.match(r"^[A-Za-z0-9._-]{1,64}$", key_id):
            raise Refused("R3", "the signed blob names no plain key_id")
        if signers is None:
            signers = TRUST_SIGNERS
            try:
                st = self.r.stat(TRUST_SIGNERS)
            except OSError:
                raise Refused("R3", f"no {TRUST_SIGNERS} — no operator key to check a signed job against")
            if st.st_uid != 0 or (st.st_mode & 0o022):
                raise Refused("R3", f"{TRUST_SIGNERS} is not root-owned and root-only-writable")
        rc = self.r.verify_sig(signers, key_id, SIG_NAMESPACE, blob, signed["sig"])
        if rc != 0:
            raise Refused("R3", f"the operator signature does not verify (ssh-keygen rc={rc})")
        if op.get("op") != op_name:
            raise Refused("R3", f"the signed op is {op.get('op')!r}, not {op_name}")
        if (op.get("target") or {}).get("host_id") != trust["host_id"]:
            raise Refused("R3", "the signed job is for another host")
        now = self.r.now()
        try:
            exp = calendar.timegm(time.strptime(op["expires_at"], "%Y-%m-%dT%H:%M:%SZ"))
            iss = calendar.timegm(time.strptime(op["issued_at"], "%Y-%m-%dT%H:%M:%SZ"))
        except (KeyError, ValueError, TypeError):
            raise Refused("R3", "the signed job has no readable time window")
        if now > exp or now < iss - 120:
            raise Refused("R3", "the signed job is expired or not yet valid")
        nonce = op.get("nonce")
        if not isinstance(nonce, str) or not nonce:
            raise Refused("R3", "the signed job has no nonce")
        if nonce in self.r.read_nonces():
            raise Refused("R3", "the signed job was already used (replay)")
        if not burn:
            return op.get("params") or {}, nonce, exp
        self.burn_nonce(nonce, exp)
        return op.get("params") or {}

    def agent_update(self, plan):
        """R-861 (agent v0.146.0): the agent binary is replaced ONLY by an operator-signed agent_update, checked HERE as
        root — signature against the root-owned signers file, op, this host, the time window, the nonce — and only the
        staged file whose sha256 the SIGNED params pin, at the one staging path. Before v0.146.0 the agent handed the
        sha to `felhom-selfupdate-guarded apply` itself, so a compromised agent could install any binary — and the
        binary is what FELHOM_ESCROW and the guest hook run as root. The nonce is burned only after the flip."""
        trust = self.load_trust()
        params, nonce, exp = self.verify_signed(plan.get("signed"), trust, op_name=SELFUPDATE_OP, burn=False)
        ver, sha = params.get("version"), params.get("sha256")
        if not isinstance(ver, str) or not re.match(r"^[0-9]+\.[0-9]+\.[0-9]+$", ver):
            raise Refused("R19", f"the signed version {ver!r} is not bare semver")
        if not isinstance(sha, str) or not re.match(r"^[0-9a-f]{64}$", sha):
            raise Refused("R19", "the signed sha256 is not 64 lowercase hex")
        staged = os.path.join(SELFUPDATE_DIR, "felhom-agent-" + ver)
        if plan.get("staged") != staged:
            raise Refused("R19", f"the staged binary must be {staged}, got {plan.get('staged')!r}")
        # ONE read, then never the agent's path again: hash exactly these bytes and hand the A/B wrapper a ROOT-OWNED
        # copy of them. Hashing the agent's file and then letting the wrapper copy it by path was a race — the agent owns
        # that directory and could swap the file between the check and the copy (found by review 2026-10-05).
        try:
            data = self.r.read_staged_once(staged, self.r.agent_uid(), SELFUPDATE_MAX_BYTES)
        except OSError as e:
            raise Refused("R19", f"cannot read the staged binary: {e}")
        got = sha256_hex(data)
        if got != sha:
            raise Refused("R19", f"the staged binary's sha256 {got[:16]}… is not the signed {sha[:16]}…")
        root_copy = os.path.join(SELFUPDATE_ROOT_DIR, "felhom-agent-" + ver)
        self.r.put_file(root_copy, data, 0o755)
        self.r.log(f"os-apply: AGENT-UPDATE signed by the operator: version={ver} sha={sha[:16]} — handing the root copy to the A/B wrapper")
        try:
            rc, out, err = self.r.host([SELFUPDATE_WRAPPER, "apply", root_copy, sha], 120)
        finally:
            try:
                self.r.remove(root_copy)
            except OSError:
                pass
        self.report["agent_update"] = {"version": ver, "sha256": sha, "wrapper_rc": rc,
                                       "wrapper": (out + err).strip()[-300:]}
        if rc != 0:
            self.report["failed"] = {"rc": rc, "step": "agent_update", "reason": (out + err).strip()[-300:]}
            return 3
        self.burn_nonce(nonce, exp)
        return 0

    def burn_nonce(self, nonce, exp):
        seen = self.r.read_nonces()
        if nonce in seen:
            raise Refused("R3", "the signed job was already used (replay)")
        seen[nonce] = exp
        now = self.r.now()
        self.r.write_nonces({k: v for k, v in seen.items() if v > now})

    def docker_authority(self, plan, op_name=SIGNED_OP):
        """R3 for the docker (and, op_name os_pve_step, the pve) layer: returns (who, undo). A signed job binds the EXACT
        package list and the undo flag."""
        trust = self.load_trust()
        signed = plan.get("signed")
        if signed:
            params = self.verify_signed(signed, trust, op_name=op_name)
            want = sorted(f"{e.get('name')}={e.get('version')}" for e in params.get("packages") or [])
            got = sorted(f"{e['name']}={e['version']}" for e in plan.get("packages", []))
            if not want or want != got:
                raise Refused("R3", "the plan's packages are not exactly the signed job's packages")
            if bool(params.get("undo")) != bool(plan.get("undo")):
                raise Refused("R3", "the plan's undo flag is not the signed job's")
            if params.get("vmid") not in (None, self.vmid):
                raise Refused("R3", "the signed job names another guest")
            return "signed", bool(plan.get("undo"))
        if plan.get("undo"):
            raise Refused("R3", "an undo (downgrade) needs a signed operator job")
        if trust.get("ring0_slow_lane") is True:
            return "ring0", False
        raise Refused("R3", f"a {self.layer} slow-lane step needs a signed operator job (ring 1) or this box's root-owned ring-0 mark")

    def live_restore(self):
        rc, out, _ = self.g(["docker", "info", "--format", "{{.LiveRestoreEnabled}}"], timeout=60)
        return out.strip() if rc == 0 and out.strip() in ("true", "false") else "unknown"

    def container_ids(self):
        rc, out, _ = self.g(["docker", "ps", "-q", "--no-trunc"], timeout=60)
        return sorted(out.split()) if rc == 0 else None

    def live_restore_on(self):
        """`09` decision 87: merge live-restore into daemon.json and RELOAD (C5: a reload turns it on, no restart)."""
        log = self.r.log
        if self.live_restore() == "true":
            self.report["live_restore"] = {"result": "already on"}
            log("os-apply: LIVE-RESTORE already on")
            return 0
        rc, cur, _ = self.g(["cat", DAEMON_JSON], timeout=30)
        try:
            conf = json.loads(cur) if rc == 0 and cur.strip() else {}
        except ValueError:
            raise Refused("R16", f"{DAEMON_JSON} in the guest is not valid JSON — not touched")
        if not isinstance(conf, dict):
            raise Refused("R16", f"{DAEMON_JSON} is not a JSON object — not touched")
        before = self.container_ids()
        conf["live-restore"] = True
        self.r.write_file("guest", self.vmid, DAEMON_JSON, json.dumps(conf, indent=2, sort_keys=True) + "\n")
        rrc, _, rerr = self.g(["systemctl", "reload", "docker"], timeout=120)
        state = "unknown"
        for _ in range(10):
            state = self.live_restore()
            if state == "true":
                break
            self.r.sleep(1)
        after = self.container_ids()
        same = before is not None and before == after
        self.report["live_restore"] = {"result": "on" if state == "true" else "failed", "reload_rc": rrc,
                                       "containers_before": len(before or []), "same_ids": same}
        log(f"os-apply: LIVE-RESTORE reload_rc={rrc} state={state} containers={len(before or [])} same-ids={'yes' if same else 'NO'}")
        if state != "true":
            # put the old file back (and reload again) — still never a restart
            self.r.write_file("guest", self.vmid, DAEMON_JSON, cur if rc == 0 else "{}\n")
            self.g(["systemctl", "reload", "docker"], timeout=120)
            self.report["failed"] = {"rc": 3, "step": "live-restore", "reason": (rerr or "").strip()[-200:]}
            return 3
        return 0

    def kernel_next_boot(self):
        """Which kernel GRUB boots next, read without root-only files (grubenv + /etc/default/grub + /boot)."""
        try:
            dflt = re.search(r'^GRUB_DEFAULT=["\']?([^"\'\n]*)', self.r.read_file("/etc/default/grub"), re.M)
            dflt = dflt.group(1) if dflt else "0"
        except OSError:
            dflt = "0"
        env = {}
        try:
            for l in self.r.read_file("/boot/grub/grubenv").splitlines():
                if "=" in l and not l.startswith("#"):
                    k, v = l.split("=", 1)
                    env[k] = v
        except OSError:
            pass

        def ver(entry):
            m = re.search(r"gnulinux-([0-9][^>\s]*?-pve)-(?:advanced|recovery)", entry)
            return m.group(1) if m else "unknown"
        if env.get("next_entry"):
            return ver(env["next_entry"]), "next_entry (a one-shot GRUB cannot clear on LVM /boot)"
        if dflt == "saved":
            return (ver(env["saved_entry"]), "saved default") if env.get("saved_entry") else ("unknown", "saved default unset")
        if dflt == "0":
            rc, out, _ = self.r.host(["sh", "-c", "ls /boot/vmlinuz-* 2>/dev/null"], 30)
            vers = [l.split("vmlinuz-", 1)[1] for l in out.split() if "vmlinuz-" in l]
            best = None
            for v in vers:
                if best is None or self.dpkg_cmp(v, "gt", best):
                    best = v
            return (best or "unknown"), "GRUB_DEFAULT=0 (the newest installed)"
        return "unknown", f"GRUB_DEFAULT={dflt}"

    def facts(self):
        """Read-only versions for the hub's System page (R-852). A value that cannot be read is "unknown"."""
        def first(cmd, timeout=30):
            rc, out, _ = self.r.host(cmd, timeout)
            v = out.strip().splitlines()[0].strip() if rc == 0 and out.strip() else ""
            return v or "unknown"
        h = {"debian": first(["cat", "/etc/debian_version"]), "kernel_running": first(["uname", "-r"])}
        h["kernel_next_boot"], h["kernel_next_boot_source"] = self.kernel_next_boot()
        try:
            # the kernel lane (R-836): the default and the one-shot flag, read from grub.cfg and the ESP themselves
            kv = Kernel(self, {}).view()
            kv["setup_problems"] = Kernel(self, {}).setup_problems()
            h["kernel_lane"] = kv
            if kv.get("flag"):
                h["kernel_next_boot"], h["kernel_next_boot_source"] = kv["flag"], "felhom one-shot flag (once; then the default)"
            elif kv.get("default") not in (None, "unknown"):
                h["kernel_next_boot"], h["kernel_next_boot_source"] = kv["default"], "grub.cfg default"
        except Exception as e:  # never cost the System page its other facts
            h["kernel_lane"] = {"error": str(e)[:200]}
        rc, out, _ = self.r.host(["apt-mark", "showhold"], 60)
        h["held"] = sorted(out.split()) if rc == 0 else None
        try:
            t = int(self.r.read_file("/proc/sys/kernel/tainted").strip())
            h["tainted"], h["oops_this_boot"], h["warn_this_boot"] = t, bool(t & 128), bool(t & 512)
        except (OSError, ValueError):
            h["tainted"], h["oops_this_boot"], h["warn_this_boot"] = None, None, None
        try:
            h["kernel_panic"] = int(self.r.read_file("/proc/sys/kernel/panic").strip())
        except (OSError, ValueError):
            h["kernel_panic"] = None
        try:
            h["crash_guard"] = json.loads(self.r.read_file(CRASH_GUARD_STATE))
        except (OSError, ValueError):
            h["crash_guard"] = None
        g = {"debian": "unknown", "docker_engine": "unknown", "containerd": "unknown", "live_restore": "unknown"}
        try:
            self.check_guest(self.vmid)
            running = True
        except Refused as e:
            running, g["unknown_reason"] = False, f"{e.code} {e.reason}"
        if running:
            script = ('echo "debian=$(cat /etc/debian_version 2>/dev/null)"; '
                      'echo "engine=$(docker version --format \'{{.Server.Version}}\' 2>/dev/null)"; '
                      'echo "containerd=$(dpkg-query -W -f \'${Version}\' containerd.io 2>/dev/null)"; '
                      'echo "live=$(docker info --format \'{{.LiveRestoreEnabled}}\' 2>/dev/null)"')
            rc, out, _ = self.g(["sh", "-c", script], timeout=60)
            kv = dict(l.split("=", 1) for l in out.splitlines() if "=" in l)
            for k, src in (("debian", "debian"), ("docker_engine", "engine"), ("containerd", "containerd")):
                g[k] = kv.get(src, "").strip() or "unknown"
            g["live_restore"] = {"true": "on", "false": "off"}.get(kv.get("live", "").strip(), "unknown")
        try:
            h["config_bundle"] = Bundle(self).state()
        except Exception as e:  # a read problem must never cost the System page its other facts
            h["config_bundle"] = {"version": "unknown", "error": str(e)[:200]}
        self.report["facts"] = {"host": h, "guest": g}
        return 0

    def check_guest(self, vmid):
        if vmid in RESERVED_VMIDS:
            raise Refused("R10", f"vmid {vmid} is a reserved scratch vmid")
        try:
            conf = self.r.read_file(f"/etc/pve/lxc/{vmid}.conf")
        except OSError:
            raise Refused("R10", f"vmid {vmid} is not a container on this host")
        cur = conf.split("\n[", 1)[0]  # the current config, not a snapshot section
        binds = [l for l in cur.splitlines() if re.match(r"^mp[0-9]+: " + re.escape(DRIVES_PARENT) + r",", l)]
        if not binds:
            raise Refused("R10", f"vmid {vmid} does not bind {DRIVES_PARENT} — it is not this box's customer guest")
        lock = [l for l in cur.splitlines() if l.startswith("lock:")]
        if lock:
            raise Refused("R9", f"vmid {vmid} is locked ({lock[0].split(':', 1)[1].strip()}) — a backup or restore is running")
        rc, out, _ = self.r.host(["/usr/sbin/pct", "status", str(vmid)])
        if rc != 0 or "running" not in out:
            raise Refused("R10", f"vmid {vmid} is not running")

    # ---------- target helpers ----------
    def x(self, argv, timeout=1800):
        """Run in the TARGET layer: the guest via pct exec, or the host directly (host, pve and kernel)."""
        if self.layer in ("host", "pve", "kernel"):
            return self.r.host(argv, timeout)
        return self.r.guest(self.vmid, argv, timeout)  # guest and docker both live in the customer guest

    def g(self, argv, timeout=1800):
        """Run in the customer GUEST whatever the layer (its health)."""
        return self.r.guest(self.vmid, argv, timeout)

    def dpkg_cmp(self, a, op, b):
        # The HOST's dpkg: the same Debian version algorithm, and no `pct exec` (0.9 s) per comparison (R-845).
        rc, _, _ = self.r.host(["dpkg", "--compare-versions", a, op, b], 30)
        return rc == 0

    def installed(self):
        rc, out, _ = self.x(["dpkg-query", "-W", "-f", "${Package}\t${Version}\t${db:Status-Abbrev}\n"])
        res = {}
        for l in out.splitlines():
            parts = l.split("\t")
            if len(parts) == 3 and parts[2].startswith("ii"):
                res[parts[0]] = parts[1]
        return res

    def madison_all(self, names):
        """name -> set of downloadable versions, ONE call for all names."""
        res = {n: set() for n in names}
        if not names:
            return res
        rc, out, _ = self.x(["apt-cache", "madison"] + sorted(names))
        for l in out.splitlines():
            f = [x.strip() for x in l.split("|")]
            if len(f) >= 3 and f[0] in res:
                res[f[0]].add(f[1])
        return res

    def simulate(self, args):
        rc, out, err = self.x(APT_ENV + ["apt-get", "-s", "-q"] + args)
        inst, remv = [], []
        for l in out.splitlines():
            m = re.match(r"^Inst (\S+) (?:\[([^]]*)\] )?\((\S+) (.*?) \[[a-z0-9]+\]\)", l)
            if m:
                inst.append({"name": m.group(1), "from": m.group(2), "to": m.group(3), "origin": m.group(4)})
            m = re.match(r"^Remv (\S+)", l)
            if m:
                remv.append(m.group(1))
        return rc, inst, remv, out + err

    @staticmethod
    def origin_name(origin):
        # "Debian:13.7/stable, Debian-Security:13/stable-security" -> {"Debian", "Debian-Security"}
        return {o.strip().split(":")[0] for o in origin.split(",") if o.strip()}

    def free_bytes(self):
        rc, out, _ = self.x(["df", "-B1", "--output=avail", "/"])
        try:
            return int(out.strip().splitlines()[-1])
        except (ValueError, IndexError):
            return -1

    def apt_lock_held(self):
        rc, out, _ = self.x(["fuser", "/var/lib/dpkg/lock-frontend", "/var/lib/dpkg/lock"])
        return rc == 0 and out.strip() != ""

    def guest_health(self):
        """The guest's signals: every container's state + health, the controller's own health, the network."""
        rc, out, _ = self.g(["docker", "ps", "-a", "--no-trunc", "--format", "{{.Names}}\t{{.State}}\t{{.Status}}\t{{.ID}}"], timeout=60)
        cont = {}
        for l in out.splitlines():
            p = l.split("\t")
            if len(p) >= 3:
                h = "healthy" if "(healthy)" in p[2] else "unhealthy" if "(unhealthy)" in p[2] else \
                    "starting" if "(health: starting)" in p[2] else "none"
                cont[p[0]] = {"state": p[1], "health": h}
                if len(p) >= 4 and p[3]:
                    cont[p[0]]["id"] = p[3]
        nrc, _, _ = self.g(["getent", "hosts", "deb.debian.org"], timeout=30)
        # R-858: the controller's own health check stayed "healthy" while it could not reach Docker at all — so ask
        # the consequence directly: can the controller talk to the engine from inside its container?
        crc, _, _ = self.g(["docker", "exec", "felhom-controller", "docker", "version", "--format", "{{.Server.Version}}"], timeout=60)
        return {"docker_ok": rc == 0, "containers": cont, "controller_docker_ok": crc == 0,
                "controller": cont.get("felhom-controller", {}).get("health", "absent"),
                "network_ok": nrc == 0}

    def health(self):
        if self.layer in ("guest", "docker"):
            return self.guest_health()
        rc, out, _ = self.r.host(["systemctl", "is-active"] + HOST_SERVICES, 30)
        states = out.split()
        svc = {s: (states[i] if i < len(states) else "unknown") for i, s in enumerate(HOST_SERVICES)}
        src, sout, _ = self.r.host(["/usr/sbin/pct", "status", str(self.vmid)], 30)
        running = src == 0 and "running" in sout
        return {"host_services": svc, "guest_running": running, "guest": self.guest_health() if running else None}

    def restart_needed(self):
        """Processes still mapping deleted files, OUTSIDE containers (C11). Guest: outside docker; host: outside the
        LXC guests (the host's /proc shows guest processes too)."""
        skip = RESTART_SKIP_CGROUP["host" if self.layer in ("host", "pve", "kernel") else "guest"]
        script = ('for p in /proc/[0-9]*; do grep -q "(deleted)" $p/maps 2>/dev/null || continue; '
                  'grep -q "%s" $p/cgroup 2>/dev/null && continue; echo "${p#/proc/} $(cat $p/comm 2>/dev/null)"; done' % skip)
        rc, out, _ = self.x(["sh", "-c", script], timeout=120)
        lines = [l for l in out.splitlines() if " " in l]
        procs = sorted({l.split(" ", 1)[1] for l in lines})
        pid1 = any(l.split(" ", 1)[0] == "1" for l in lines)
        return procs, pid1 or "lxc-start" in procs

    def inventory(self, inst=None):
        inst = inst if inst is not None else self.installed()
        names = sorted(inst)
        origins = {}
        if names:
            rc, out, _ = self.x(["apt-cache", "policy"] + names)  # ONE call (R-845)
            cur, star = None, False
            for l in out.splitlines():
                if not l.startswith(" "):
                    cur, star = l.rstrip(":"), False
                    continue
                s = l.strip()
                if s.startswith("*** "):
                    star = True
                    continue
                if star and cur and re.match(r"^[0-9-]+ ", s):
                    if "/var/lib/dpkg/status" in s:
                        origins.setdefault(cur, "local")
                    else:
                        origins[cur] = s
                    continue
                if star and not re.match(r"^[0-9-]+ ", s):
                    star = False

        def oname(src):
            if src in (None, "local"):
                return "unknown"
            if "proxmox" in src:
                return "Proxmox"
            if "security" in src and "debian" in src:
                return "Debian-Security"
            if "docker.com" in src:
                return "Docker"
            if "debian" in src:
                return "Debian"
            return "other"
        rc, pend, remv, _ = self.simulate(["dist-upgrade"])
        self._pending = pend
        return {
            "installed": [{"name": n, "version": inst[n], "origin": oname(origins.get(n))} for n in names],
            "pending": [{"name": p["name"], "from": p["from"], "to": p["to"],
                         "origin": sorted(self.origin_name(p["origin"]))} for p in pend],
        }

    # ---------- the run ----------
    def run(self):
        plan = self.load_plan()
        self.mode, self.layer, self.vmid, self.select = self.check_plan(plan)
        self.report.update(mode=self.mode, layer=self.layer, release_id=plan.get("release_id"), vmid=self.vmid)
        # R-868 (v0.144.0): the agent's run id, trigger and ring travel in the report, so a report the agent never
        # received (it was killed mid-pass) can be sent later from the saved copy. Plain ids only; anything else
        # is dropped, never refused (the plan's other checks decide).
        rid, trig, ring = plan.get("run_id"), plan.get("trigger"), plan.get("ring")
        if isinstance(rid, str) and re.match(r"^[A-Za-z0-9._-]{1,80}$", rid):
            self.report["run_id"] = rid
        if isinstance(trig, str) and re.match(r"^[a-z0-9_-]{1,20}$", trig):
            self.report["trigger"] = trig
        if ring in (0, 1) and not isinstance(ring, bool):
            self.report["ring"] = ring
        if self.mode == "facts":
            return self.facts()
        if self.mode == "bundle":
            return Bundle(self).from_plan(plan)
        if self.mode == "agent_update":
            return self.agent_update(plan)
        if self.layer in ("host", "pve", "kernel"):
            self.check_appliance()
        if self.layer == "kernel" and self.mode != "health":
            # the kernel lane's own modes (R-836): most of them run while the guest is still starting after a boot, so
            # the guest check is the stage's and the reboot's own (Kernel.run), not every mode's
            return Kernel(self, plan).run()
        self.check_guest(self.vmid)
        log = self.r.log
        if self.mode == "live-restore-on":
            return self.live_restore_on()
        if self.mode == "oom-check":
            # R-528: the check alone — no apt, no engine change; check_guest above still applies.
            self.report["oom_check"] = self.oom_check()
            return 0
        self.who, self.allow_downgrade = ("fast", False)
        if self.layer == "pve" and self.mode == "apply":
            self.who, self.allow_downgrade = self.docker_authority(plan, op_name=PVE_SIGNED_OP)
            self.report["authority"] = self.who
        if self.layer == "docker" and self.mode == "apply":
            self.who, self.allow_downgrade = self.docker_authority(plan)
            if self.live_restore() != "true":
                raise Refused("R15", "live-restore is not ON in the guest — a Docker step would restart every container")
            self.report["authority"] = self.who
            self.report["undo"] = self.allow_downgrade
        if self.mode == "health":
            self.report["health"] = self.health()
            return 0
        log(f"os-apply: START release={plan.get('release_id')} layer={self.layer}" +
            (f":{self.vmid}" if self.layer != "host" else "") +
            f" lane={plan.get('lane', 'fast')} mode={self.mode} select={self.select} packages={len(plan.get('packages', []))}" +
            (f" authority={self.who}{' UNDO' if self.allow_downgrade else ''}" if self.layer in ("docker", "pve") else ""))
        if self.apt_lock_held():
            raise Refused("R9", f"another apt/dpkg holds the lock on the {self.layer}")
        self.report["health_before"] = self.health()
        if self.mode == "apply":
            self.repair()
        rc, out, err = self.x(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
        if rc != 0:
            raise Refused("R7", f"apt-get update failed on the {self.layer}: {(out + err).strip().splitlines()[-1:]}")
        installed_after = None
        if self.mode == "apply":
            rc, installed_after = self.apply(plan)
            if rc:
                return rc
        self.report.update(self.inventory(installed_after))
        if "reboot_needed" not in self.report:
            # EVERY layer is scanned on EVERY pass: a reboot (host) or a restart (guest) must CLEAR "restart needed",
            # or the fleet view keeps a stale date (R-849, v0.142.0; the host since v0.141.1). One pct exec, ~1 s.
            # Pinned by test_every_layer_scans_every_pass.
            self.report["restart_needed"], self.report["reboot_needed"] = self.restart_needed()
            self.report["docker_restart_needed"] = any(p in ("dockerd", "containerd") for p in self.report["restart_needed"])
        if self.layer == "docker":
            rc_v, out_v, _ = self.g(["docker", "version", "--format", "{{.Server.Version}}"], timeout=60)
            self.report["docker_engine"] = out_v.strip() if rc_v == 0 and out_v.strip() else "unknown"
        if self.layer == "pve":
            self.report["pve_manager"] = self.pve_manager()
        self.report["reboot_scanned"] = "reboot_needed" in self.report
        self.report["health_after"] = self.health()
        if self.layer == "docker" and self.mode == "apply":
            # R-528 (`09` decision 157): does the engine report a memory kill? Reported only — never the outcome.
            self.report["oom_check"] = self.oom_check()
        return 0

    # ---------- R-528: the memory-kill check ----------
    def oom_check(self):
        """Run one throwaway container over its memory cap in the guest and read what the engine says about it.
        Never raises: any failure becomes result "error". The container is ALWAYS removed (finally), and a failed
        removal is named in the detail."""
        t = OOMCHECK_TIMEOUTS
        res = {"result": "error", "oom_killed": False, "oom_event": False, "exit_code": None, "image": None, "detail": ""}
        log = self.r.log
        try:
            rc, out, err = self.g(["docker", "inspect", "-f", "{{.Config.Image}}", "felhom-controller"], timeout=t["image"])
        except Exception as e:
            rc, out, err = -1, "", str(e)
        img = out.strip() if rc == 0 else ""
        if not img or any(c.isspace() for c in img):
            res["detail"] = f"the controller's image could not be read (rc={rc}): {(err or out).strip()[:200]}"
            log(f"os-apply: OOM-CHECK error — {res['detail']}")
            return res
        res["image"] = img
        name = f"{OOMCHECK_PREFIX}{os.getpid()}-{os.urandom(4).hex()}"
        notes = []
        try:
            t0 = self.guest_epoch()
            if t0 is None:
                raise RuntimeError("the guest clock could not be read")
            rrc, rout, rerr = self.g(["docker", "run", "--name", name, "--pull", "never", "--network", "none",
                                      "--memory", "64m", "--memory-swap", "64m", "--label", "felhom.oomcheck=1",
                                      "--entrypoint", "sh", img, "-c", OOMCHECK_SCRIPT], timeout=t["run"])
            self.r.sleep(OOMCHECK_SETTLE)  # the engine publishes the oom event a moment after the run returns (F1/F2)
            t1 = self.guest_epoch()
            if t1 is None:
                t1 = t0 + t["run"] + OOMCHECK_SETTLE + 1
            irc, iout, ierr = self.g(["docker", "inspect", "-f", "{{.State.OOMKilled}} {{.State.ExitCode}}", name], timeout=t["inspect"])
            if irc != 0:
                raise RuntimeError(f"the check container could not be inspected (run rc={rrc}: {(rerr or rout).strip()[:120]}; "
                                   f"inspect rc={irc}: {(ierr or iout).strip()[:120]})")
            f = iout.split()
            res["oom_killed"] = bool(f) and f[0] == "true"
            try:
                res["exit_code"] = int(f[1]) if len(f) > 1 else None
            except ValueError:
                res["exit_code"] = None
            erc, eout, eerr = self.g(["docker", "events", "--since", str(t0 - 1), "--until", str(t1 + 1),
                                      "--filter", f"container={name}", "--filter", "event=oom",
                                      "--format", "{{.Action}}"], timeout=t["events"])
            if erc != 0:
                notes.append(f"the event read failed (rc={erc}): {(eerr or eout).strip()[:120]}")
            res["oom_event"] = erc == 0 and any(l.strip() == "oom" for l in eout.splitlines())
            if res["oom_killed"] and res["oom_event"]:
                res["result"] = "pass"
                notes.insert(0, "the engine reported the memory kill: OOMKilled=true and the oom event")
            else:
                res["result"] = "fail"
                miss = [w for w, ok in (("OOMKilled=true", res["oom_killed"]), ("the oom event", res["oom_event"])) if not ok]
                notes.insert(0, f"the engine did not report the memory kill: missing {' and '.join(miss)} (exit code {res['exit_code']})")
        except Exception as e:
            res["result"] = "error"
            notes.insert(0, f"the check could not finish: {type(e).__name__}: {str(e)[:200]}")
        finally:
            try:
                mrc, mout, merr = self.g(["docker", "rm", "-f", name], timeout=t["rm"])
                if mrc != 0 and "no such container" not in (merr + mout).lower():
                    notes.append(f"the check container {name} could not be removed (rc={mrc}): {(merr or mout).strip()[:120]}")
            except Exception as e:
                notes.append(f"the check container {name} could not be removed: {type(e).__name__}: {str(e)[:120]}")
        res["detail"] = "; ".join(notes)
        log(f"os-apply: OOM-CHECK result={res['result']} oom_killed={res['oom_killed']} oom_event={res['oom_event']} "
            f"exit={res['exit_code']} image={img} — {res['detail']}")
        return res

    def guest_epoch(self):
        rc, out, _ = self.g(["date", "+%s"], timeout=OOMCHECK_TIMEOUTS["clock"])
        try:
            return int(out.strip()) if rc == 0 else None
        except ValueError:
            return None

    def dpkg_state(self):
        """`dpkg --audit` AND dpkg's update journal, in ONE call (R-876, agent v0.145.0). A crash in the middle of an
        install can leave `/var/lib/dpkg/updates/` non-empty while `--audit` reads clean — measured on demo-hp
        2026-10-05 — and that journal is exactly what apt refuses on ("dpkg was interrupted"). One `sh -c` with a
        constant script keeps R-845's speed: a clean pass still costs one call here, as before."""
        rc, out, _ = self.x(["sh", "-c", DPKG_STATE_SCRIPT])
        audit, _, journal = out.partition(JOURNAL_MARK + "\n")
        return audit, [l for l in journal.split() if l]

    def repair(self, force=False):
        before, journal = self.dpkg_state()
        configured = len([l for l in before.splitlines() if l.startswith(" ")])
        fixed = 0
        after, journal_after = "", []
        # nothing half-done and no update journal → nothing to run (R-845: two calls saved on every clean pass).
        # R-876: the JOURNAL counts too, and `force` (apt said "dpkg was interrupted") always repairs.
        if before.strip() or journal or force:
            self.x(APT_ENV + ["dpkg", "--configure", "-a", "--force-confold"])
            rc2, out, err = self.x(APT_ENV + ["apt-get", "-f", "install", "-y", "-q"] + DPKG_OPTS)
            after, journal_after = self.dpkg_state()
            fixed = len(re.findall(r"^Setting up ", out, re.M))
        self.report["repair"] = {"half_configured_before": configured, "journal_before": len(journal), "fixed": fixed,
                                 "clean_after": after.strip() == "" and not journal_after}
        self.r.log(f"os-apply: REPAIR configured={configured} journal={len(journal)} fixed={fixed}" + (" forced" if force else ""))
        if after.strip():
            raise Refused("R13", "dpkg is still broken after the repair: " + after.strip().splitlines()[0])
        if journal_after:
            raise Refused("R13", f"dpkg's update journal is still not empty after the repair ({len(journal_after)} file(s))")

    def pending_fast(self):
        """Ring 0 (select pending-fast): every pending upgrade of an INSTALLED package whose every origin is Debian /
        Debian-Security — and, on the host, not a kernel / boot / firmware package."""
        rc, pend, remv, _ = self.simulate(["dist-upgrade"])
        out = []
        for p in pend:
            o = self.origin_name(p["origin"])
            if p["from"] is None or not o or not o <= set(FAST_ORIGINS):
                continue
            if self.layer == "host" and HOST_SLOW_RE.match(p["name"]):
                continue
            out.append({"name": p["name"], "version": p["to"], "origin": "Debian-Security" if "Debian-Security" in o else "Debian"})
        return out

    def restart_socket_users(self):
        """R-858: restart ONLY the containers that bind-mount the Docker socket, so they attach to the new one."""
        rc, out, _ = self.g(["docker", "ps", "-q", "--no-trunc"], timeout=60)
        users = []
        for cid in out.split():
            irc, iout, _ = self.g(["docker", "inspect", "-f", "{{.Name}}|{{range .Mounts}}{{.Destination}};{{end}}", cid], timeout=60)
            if irc != 0 or "|" not in iout:
                continue
            name, mounts = iout.strip().split("|", 1)
            if any(m in DOCKER_SOCKETS for m in mounts.split(";")):
                users.append(name.lstrip("/"))
        users.sort()
        if users:
            rrc, _, rerr = self.g(["docker", "restart"] + users, timeout=300)
            self.r.log(f"os-apply: SOCKET-USERS restarted={','.join(users)} rc={rrc} (R-858: they held the old docker socket)")
        return users

    def pending_docker(self):
        """Ring 0 (select pending-docker): the newest pending version of each INSTALLED Docker package, Docker origin."""
        rc, pend, remv, _ = self.simulate(["dist-upgrade"])
        return [{"name": p["name"], "version": p["to"], "origin": DOCKER_ORIGIN} for p in pend
                if p["from"] is not None and p["name"] in DOCKER_NAMES and self.origin_name(p["origin"]) == {DOCKER_ORIGIN}]

    def pending_pve(self):
        """Ring 0 (select pending-pve): the newest pending version of each INSTALLED Proxmox-origin package, never a
        kernel / boot / firmware / microcode name (HOST_SLOW_RE, R-836's lane), never another origin."""
        rc, pend, remv, _ = self.simulate(["dist-upgrade"])
        return [{"name": p["name"], "version": p["to"], "origin": PVE_ORIGIN} for p in pend
                if p["from"] is not None and not HOST_SLOW_RE.match(p["name"]) and p["name"] not in DOCKER_NAMES
                and self.origin_name(p["origin"]) == {PVE_ORIGIN}]

    def pve_manager(self):
        """pveversion's pve-manager version ("unknown" when it cannot be read)."""
        rc, out, _ = self.r.host(["pveversion"], 60)
        m = re.match(r"^pve-manager/([^/\s]+)", out.strip()) if rc == 0 else None
        return m.group(1) if m else "unknown"

    def origin_ok(self, origin):
        o = self.origin_name(origin)
        if self.layer == "docker":
            return o == {DOCKER_ORIGIN}
        if self.layer == "pve":
            return o == {PVE_ORIGIN}
        return bool(o & set(FAST_ORIGINS))

    def apply(self, plan):
        log = self.r.log
        if self.select == "listed":
            packages = plan["packages"]
        elif self.select == "pending-docker":
            packages = self.pending_docker()
        elif self.select == "pending-pve":
            packages = self.pending_pve()
        else:
            packages = self.pending_fast()
        cmp_op = "ne" if self.allow_downgrade else "gt"
        inst = self.installed()
        upgrade, already, notinst = [], 0, 0
        for e in packages:
            n, v = e["name"], e["version"]
            if n not in inst:
                notinst += 1
                continue
            if not self.dpkg_cmp(v, cmp_op, inst[n]):
                already += 1
                continue
            upgrade.append((n, v))
        from_snap = 0
        if upgrade:
            avail = self.madison_all([n for n, _ in upgrade])
            missing = [(n, v) for n, v in upgrade if v not in avail[n]]
        else:
            missing = []
        if missing:
            snap = plan.get("snapshot", "")
            if not snap:
                raise Refused("R7", f"{missing[0][0]}={missing[0][1]} is not downloadable and the plan names no snapshot")
            self.add_snapshot_sources(snap)
            avail = self.madison_all([n for n, _ in missing])
            still = [(n, v) for n, v in missing if v not in avail[n]]
            if still:
                self.remove_snapshot_sources()
                raise Refused("R7", f"{still[0][0]}={still[0][1]} is not downloadable, not even from snapshot {snap}")
            from_snap = len(missing)
        try:
            log(f"os-apply: PLAN upgrade={len(upgrade)} already={already} not-installed={notinst} from-snapshot={from_snap}")
            self.report["plan"] = {"upgrade": len(upgrade), "already": already, "not_installed": notinst, "from_snapshot": from_snap}
            if not upgrade:
                self.report["upgraded"] = []
                log("os-apply: DONE rc=0 seconds=0 upgraded=0 (nothing to do)")
                return 0, inst
            args = ["install", "--only-upgrade", "--no-install-recommends"] + \
                (["--allow-downgrades"] if self.allow_downgrade else []) + [f"{n}={v}" for n, v in upgrade]
            rc, sim, remv, text = self.simulate(args)
            if rc != 0:
                tail = text.strip().splitlines()[-1] if text.strip() else ""
                raise Refused("R7", "the simulation failed: " + tail)
            if remv:
                raise Refused("R4", f"the plan would remove {', '.join(remv[:5])}")
            want = dict(upgrade)
            for p in sim:
                if p["from"] is None and self.layer == "pve" and p["name"] in PVE_NEW_ALLOW and self.origin_ok(p["origin"]) \
                        and not HOST_SLOW_RE.match(p["name"]):
                    self.r.log(f"os-apply: NEW {p['name']}={p['to']} (on the pve lane's allow-list)")
                    self.report.setdefault("added", []).append({"name": p["name"], "version": p["to"]})
                    continue
                if p["from"] is None:
                    raise Refused("R6", f"the plan would add a package that is not installed: {p['name']}")
                if p["name"] not in want:
                    raise Refused("R6", f"the plan would touch {p['name']}, which is not in the plan")
                if p["to"] != want[p["name"]]:
                    raise Refused("R6", f"{p['name']} would go to {p['to']}, not the approved {want[p['name']]}")
                if not self.allow_downgrade and not self.dpkg_cmp(p["to"], "gt", p["from"]):
                    raise Refused("R5", f"{p['name']} would be downgraded {p['from']} -> {p['to']}")
                if not self.origin_ok(p["origin"]):
                    raise Refused("R2", f"{p['name']} would come from {p['origin']}, not the {self.layer} layer's origin")
                if self.layer in ("host", "pve") and HOST_SLOW_RE.match(p["name"]):
                    raise Refused("R14", f"{p['name']} is a kernel / boot / firmware package — the host's slow lane")
            need = self.download_bytes(args)
            free = self.free_bytes()
            if free >= 0 and free < max(MIN_FREE, 3 * need):
                raise Refused("R8", f"free space {free} B is below max(500 MB, 3 x download {need} B)")
            t0 = time.time()
            rc, out, err = self.x(APT_ENV + ["apt-get", "-y", "-q"] + DPKG_OPTS + args)
            if rc != 0 and "dpkg was interrupted" in (out + err):
                # R-876 (belt): apt says dpkg was interrupted although the repair found nothing — repair and
                # retry ONCE. Never a loop.
                self.r.log("os-apply: INTERRUPTED apt says dpkg was interrupted — repairing and retrying once")
                self.repair(force=True)
                rc, out, err = self.x(APT_ENV + ["apt-get", "-y", "-q"] + DPKG_OPTS + args)
            secs = time.time() - t0
            # dpkg says "Installing new version of config file X" when X was NOT changed locally (the package's new
            # version is taken), and "Configuration file 'X'" + "Keeping old config file" when it was (--force-confold
            # keeps the local one; the package's version lands as X.dpkg-dist). Measured live 2026-10-04 (debian_version).
            conflict = None
            for l in (out + err).splitlines():
                m = re.search(r"Installing new version of config file (\S+?)\s*\.\.\.", l)
                if m:
                    log(f"os-apply: CONFFILE updated {m.group(1)} (it was not changed locally)")
                m = re.search(r"Configuration file '([^']+)'", l)
                if m:
                    conflict = m.group(1)
                if conflict and "Keeping old config file" in l:
                    log(f"os-apply: CONFFILE kept {conflict} (changed locally; the package's version is {conflict}.dpkg-dist)")
                    self.report.setdefault("conffiles_kept", []).append(conflict)
                    conflict = None
            self.x(["apt-get", "clean"])
            if rc != 0:
                _, aud, _ = self.x(["dpkg", "--audit"])
                first = aud.strip().splitlines()[0] if aud.strip() else "clean"
                log(f"os-apply: FAILED rc={rc} step=install — dpkg state: {first}")
                self.report["failed"] = {"rc": rc, "dpkg_audit": first, "tail": (out + err).strip().splitlines()[-3:]}
                return 3, None
            self.report["upgraded"] = [{"name": n, "version": v} for n, v in upgrade]
            self.report["seconds"] = round(secs, 1)
            if self.layer == "docker":
                self.report["socket_restarted"] = self.restart_socket_users()
            procs, reboot = self.restart_needed()
            self.report["restart_needed"] = procs
            self.report["docker_restart_needed"] = any(p in ("dockerd", "containerd") for p in procs)
            self.report["reboot_needed"] = reboot
            log(f"os-apply: DONE rc=0 seconds={secs:.1f} upgraded={len(upgrade)} restart-needed={','.join(procs) or '-'} reboot-needed={'yes' if reboot else 'no'}")
            return 0, None
        finally:
            if from_snap:
                self.remove_snapshot_sources()

    def download_bytes(self, args):
        # R-865 (v0.144.0): NO `-s`. With `-s` apt prints the simulation ("Inst …") and no URI list, so this summed
        # 0 B and R8 only ever applied its 500 MB floor (measured 2026-10-04: 0 URIs with -s, 3 URIs without).
        # `--print-uris` alone downloads nothing — measured on 9202 2026-10-05: the archive cache and the versions
        # unchanged. Pinned by test_R8_counts_the_real_download / test_download_bytes_never_simulates.
        rc, out, _ = self.x(APT_ENV + ["apt-get", "-o", "Debug::NoLocking=1", "--print-uris", "-q"] + args)
        total = 0
        for l in out.splitlines():
            m = re.match(r"^'[^']+' \S+ ([0-9]+) ", l)
            if m:
                total += int(m.group(1))
        return total

    def add_snapshot_sources(self, snap):
        rc, out, _ = self.x(["sh", "-c", ". /etc/os-release && echo $VERSION_CODENAME"])
        code = out.strip()
        if not re.match(r"^[a-z]+$", code):
            raise Refused("R7", f"cannot read the {self.layer}'s Debian codename ({code!r})")
        body = (f"deb [check-valid-until=no] http://snapshot.debian.org/archive/debian/{snap} {code} main\n"
                f"deb [check-valid-until=no] http://snapshot.debian.org/archive/debian-security/{snap} {code}-security main\n")
        self.r.write_file(self.layer, self.vmid, SNAPSHOT_LIST, body)
        self.r.log(f"os-apply: SNAPSHOT using snapshot.debian.org/{snap} for versions no longer published (decision 79)")
        rc, out, err = self.x(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
        if rc != 0:
            self.remove_snapshot_sources()
            raise Refused("R7", "apt-get update against snapshot.debian.org failed")

    def remove_snapshot_sources(self):
        self.x(["rm", "-f", SNAPSHOT_LIST])
        self.x(APT_ENV + ["apt-get", "-q", "update"], timeout=600)


def _iso(t):
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t))


def _parse_iso(s):
    try:
        return calendar.timegm(time.strptime(s, "%Y-%m-%dT%H:%M:%SZ"))
    except (TypeError, ValueError):
        return None


class Kernel:
    """The kernel lane (R-836, `09` §3 decisions 164 + 172, `11` §5.11). Refusal codes: R3 (authority), R4 (removal),
    R6 (a package outside the step), R8 (space), R9 (locks), R12 (appliance), R20 (the box's boot setup cannot do a
    one-shot), R21 (the crash guard is tripped or saw an unclean boot within its window), R22 (the step's phase does not
    allow this mode), R23 (the kernel set is not one exact new kernel).

    Modes (plan "mode", layer "kernel", lane "slow"):
      apply          STAGE: install the kernel set, keep the GRUB default on the kernel the box runs, write the flag.
                     Never reboots. select "pending-kernel" (ring 0, the root-owned mark) or "listed" (a signed
                     os_kernel_step). expect_kver: the kernel the hub told the household about — any other is R23.
      kernel-reboot  a STAGED step's reboot (the night leg, after the household was told): phase oneshot, then reboot.
      kernel-boot    after a boot: what became of the step (judging | fell_back | self_reverted | revert_failed).
      kernel-good    the one-shot boot was healthy: the new kernel becomes the GRUB default.
      kernel-revert  the one-shot boot was NOT healthy: reboot ONCE into the old kernel (still the default).
      kernel-cancel  drop a staged step: clear the flag (the package stays installed, the default never moved).
      kernel-status  read only."""

    def __init__(self, apply, plan):
        self.a, self.r, self.plan = apply, apply.r, plan
        self.report = apply.report
        self.log = apply.r.log

    # ---------- reading the box ----------
    def running(self):
        rc, out, _ = self.r.host(["uname", "-r"], 30)
        v = out.strip() if rc == 0 else ""
        return v if KVER_RE.match(v) else ""

    def state(self):
        try:
            s = json.loads(self.r.read_file(KERNEL_STATE))
            return s if isinstance(s, dict) else {}
        except (OSError, ValueError):
            return {}

    def save_state(self, s):
        s["updated_at"] = _iso(self.r.now())
        self.r.put_file(KERNEL_STATE, (json.dumps(s, indent=2, sort_keys=True) + "\n").encode(), 0o644)

    def flag(self):
        """The one-shot flag: the kernel named, "" when the env block holds none, None when there is no env block."""
        rc, out, _ = self.r.host(["grub-editenv", ONESHOT_ENV, "list"], 30)
        if rc != 0:
            return None
        for l in out.splitlines():
            if l.startswith("felhom_next="):
                return l.split("=", 1)[1].strip()
        return ""

    def grub_cfg(self):
        try:
            return self.r.read_file(GRUB_CFG)
        except OSError:
            return ""

    @staticmethod
    def default_kver(cfg):
        """The kernel grub.cfg boots by default (00_header's `set default=`), or "unknown"."""
        m = re.search(r'^\s*set default="(?:gnulinux-advanced-[^>"]*>)?gnulinux-([0-9][^"]*?-pve)-advanced-[^"]*"', cfg, re.M)
        return m.group(1) if m else "unknown"

    @staticmethod
    def entry_id(cfg, kver):
        """The GRUB_DEFAULT value that names kver's normal entry, read from grub.cfg itself (10_linux's ids)."""
        sub = re.search(r"\$menuentry_id_option '(gnulinux-advanced-[^']+)'", cfg)
        m = re.search(r"\$menuentry_id_option '(gnulinux-" + re.escape(kver) + r"-advanced-[^']+)'", cfg)
        if not m:
            return None
        return f"{sub.group(1)}>{m.group(1)}" if sub else m.group(1)

    def setup_problems(self):
        """R20: why this box cannot do a one-shot boot (empty = it can). Measured shape: UEFI, a vfat ESP at /boot/efi,
        /boot on the root filesystem, GRUB with fat + loadenv, the bundle's two generators, no hand pin."""
        why = []
        if not self.r.lexists("/sys/firmware/efi"):
            why.append("the box does not boot UEFI")
        rc, out, _ = self.r.host(["findmnt", "-n", "-o", "FSTYPE", ESP_MOUNT], 30)
        if rc != 0 or out.strip() != "vfat":
            why.append(f"{ESP_MOUNT} is not a mounted vfat ESP ({out.strip() or 'not mounted'})")
        rc, out, _ = self.r.host(["findmnt", "-n", "-o", "TARGET", "/boot"], 30)
        if rc == 0 and out.strip():
            why.append("/boot is a separate filesystem (the one-shot entries assume /boot on the root filesystem)")
        for m in ("fat", "loadenv"):
            if not self.r.lexists(f"/usr/lib/grub/x86_64-efi/{m}.mod"):
                why.append(f"GRUB has no {m} module")
        for p in (ONESHOT_SNIPPET, ONESHOT_ENTRIES):
            if not self.r.lexists(p):
                why.append(f"{p} is missing (the config bundle installs it)")
        if self.r.lexists(KERNEL_PIN_FILE):
            why.append("a kernel is pinned by hand (proxmox-boot-tool kernel pin) — the lane never fights it")
        return why

    def guard(self):
        try:
            return json.loads(self.r.read_file(CRASH_GUARD_STATE))
        except (OSError, ValueError):
            return None

    def check_guard(self):
        """R21: a kernel step only on a box whose crash guard is armed and saw no unclean boot within its window — so
        the step's own reboots (clean), one crash and one self-revert (clean) can never reach the 3rd unclean boot that
        leaves the box off (`11` §5.9). Pinned by test_felhom_crash_guard KernelStepCannotLeaveTheBoxOff."""
        g = self.guard()
        if not isinstance(g, dict):
            raise Refused("R21", f"no crash guard state ({CRASH_GUARD_STATE}) — a kernel step needs the guard")
        if g.get("tripped") or not g.get("armed"):
            raise Refused("R21", "the crash guard is tripped — no kernel step until it re-arms")
        if (g.get("unclean_boots_in_window") or 0) > 0:
            raise Refused("R21", f"{g.get('unclean_boots_in_window')} unclean boot(s) within the guard's window — wait")

    def view(self, st=None, cfg=None):
        st = self.state() if st is None else st
        cfg = self.grub_cfg() if cfg is None else cfg
        return {"running": self.running() or "unknown", "default": self.default_kver(cfg), "flag": self.flag(),
                "phase": st.get("phase", "none"), "from": st.get("from"), "to": st.get("to"),
                "step_id": st.get("step_id"), "self_revert_used": bool(st.get("self_revert_used")), "vmid": st.get("vmid"),
                "staged_at": st.get("staged_at"), "rebooted_at": st.get("rebooted_at"),
                "result_at": st.get("result_at"), "reason": st.get("reason")}

    # ---------- writing the box ----------
    def write_default(self, kver):
        """Pin the GRUB default to kver's normal entry, regenerate grub.cfg, and PROVE it (the default read back)."""
        cfg = self.grub_cfg()
        eid = self.entry_id(cfg, kver)
        if not eid:
            raise Refused("R20", f"grub.cfg has no normal entry for {kver}")
        body = ("# felhom kernel lane (R-836, `11` §5.11) — written by felhom-os-apply; the kernel that booted healthily\n"
                f'GRUB_DEFAULT="{eid}"\n')
        self.r.put_file(KERNEL_DEFAULT_CFG, body.encode(), 0o644)
        rc, out, err = self.r.host(["update-grub"], 300)
        got = self.default_kver(self.grub_cfg())
        if rc != 0 or got != kver:
            raise Refused("R20", f"update-grub rc={rc}: the default reads {got}, not {kver}: {(out + err).strip()[-200:]}")
        self.log(f"os-apply: KERNEL default = {kver} (proved from grub.cfg)")

    def set_flag(self, kver):
        self.r.host(["mkdir", "-p", os.path.dirname(ONESHOT_ENV)], 30)
        if self.flag() is None:
            rc, out, err = self.r.host(["grub-editenv", ONESHOT_ENV, "create"], 30)
            if rc != 0:
                raise Refused("R20", f"cannot create the one-shot env block on the ESP: {(out + err).strip()[-200:]}")
        rc, out, err = self.r.host(["grub-editenv", ONESHOT_ENV, "set", f"felhom_next={kver}"], 30)
        if rc != 0 or self.flag() != kver:
            raise Refused("R20", f"the one-shot flag did not read back as {kver}: {(out + err).strip()[-200:]}")

    def clear_flag(self):
        if self.flag():
            self.r.host(["grub-editenv", ONESHOT_ENV, "unset", "felhom_next"], 30)

    def reboot(self, why):
        self.log(f"os-apply: KERNEL REBOOT — {why}")
        rc, out, err = self.r.host(["systemctl", "reboot"], 60)
        self.report["reboot_rc"] = rc
        if rc != 0:
            self.report["failed"] = {"rc": 3, "step": "reboot", "reason": (out + err).strip()[-200:]}
            return 3
        return 0

    # ---------- the modes ----------
    def run(self):
        mode = self.a.mode
        self.report["kernel_mode"] = mode
        if mode == "kernel-status":
            v = self.view()
            v["setup_problems"] = self.setup_problems()
            self.report["kernel"] = v
            return 0
        fn = {"apply": self.stage, "kernel-reboot": self.reboot_staged, "kernel-boot": self.after_boot,
              "kernel-good": self.good, "kernel-revert": self.revert, "kernel-cancel": self.cancel}[mode]
        rc = fn()
        self.report["kernel"] = self.view()
        return rc

    def phase_is(self, st, *phases):
        if st.get("phase") not in phases:
            raise Refused("R22", f"the kernel step is {st.get('phase', 'none')!r}, not {' or '.join(phases)} — "
                                 f"{self.a.mode} does not apply")

    def stage(self):
        a = self.a
        st = self.state()
        if st.get("phase") in KERNEL_ACTIVE:
            raise Refused("R22", f"a kernel step is already {st['phase']} ({st.get('from')} -> {st.get('to')})")
        last = _parse_iso(st.get("staged_at"))
        if last is not None and self.r.now() - last < KERNEL_MIN_GAP:
            raise Refused("R22", "a kernel step was staged within the last 20 hours — never two in one night")
        why = self.setup_problems()
        if why:
            raise Refused("R20", "; ".join(why))
        self.check_guard()
        a.check_guest(a.vmid)
        who, _ = a.docker_authority(self.plan, op_name=KERNEL_OP)
        self.report["authority"] = who
        old = self.running()
        if not old:
            raise Refused("R20", "the running kernel is not a Proxmox kernel version")
        self.log(f"os-apply: START release={self.plan.get('release_id')} layer=kernel lane=slow mode=apply "
                 f"select={a.select} authority={who} running={old}")
        if a.apt_lock_held():
            raise Refused("R9", "another apt/dpkg holds the lock on the host")
        self.report["health_before"] = a.health()
        a.repair()
        rc, out, err = a.x(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
        if rc != 0:
            raise Refused("R7", f"apt-get update failed on the host: {(out + err).strip().splitlines()[-1:]}")
        inst = a.installed()
        if a.select == "pending-kernel":
            _, pend, _, _ = a.simulate(["dist-upgrade"])
            want = [(p["name"], p["to"]) for p in pend if p["from"] is not None and KERNEL_UPGRADE_RE.match(p["name"])
                    and a.origin_name(p["origin"]) == {PVE_ORIGIN}]
        else:
            want = [(e["name"], e["version"]) for e in self.plan["packages"]]
        # upgrades of installed names (never a downgrade), and at most the listed new kernel image
        args, upg = [], {}
        for n, v in want:
            if n in inst:
                if a.dpkg_cmp(v, "gt", inst[n]):
                    upg[n] = v
                    args.append(f"{n}={v}")
            elif KERNEL_IMAGE_RE.match(n):
                args.append(f"{n}={v}")
            else:
                raise Refused("R6", f"{n} is not installed and is not a kernel image")
        if not args:
            self.report["upgraded"], self.report["outcome_hint"] = [], "nothing"
            self.log("os-apply: DONE rc=0 upgraded=0 (no pending kernel)")
            return 0
        rc, sim, remv, text = a.simulate(["install", "--no-install-recommends"] + args)
        if rc != 0:
            raise Refused("R7", "the simulation failed: " + (text.strip().splitlines()[-1] if text.strip() else ""))
        if remv:
            raise Refused("R4", f"the kernel step would remove {', '.join(remv[:5])}")
        images = []
        listed = dict(want)
        for p in sim:
            if a.origin_name(p["origin"]) != {PVE_ORIGIN}:
                raise Refused("R2", f"{p['name']} would come from {p['origin']}, not {PVE_ORIGIN!r}")
            m = KERNEL_IMAGE_RE.match(p["name"])
            if p["from"] is None:
                if not m:
                    raise Refused("R6", f"the kernel step would add {p['name']}, which is not a kernel image")
                if a.select == "listed" and listed.get(p["name"]) != p["to"]:
                    raise Refused("R23", f"the kernel step would add {p['name']}={p['to']}, not the signed set")
                images.append((m.group(1), p["to"]))
                continue
            if p["name"] not in upg or p["to"] != upg[p["name"]]:
                raise Refused("R6", f"the kernel step would touch {p['name']} ({p['to']}), which is not in the step")
            if not a.dpkg_cmp(p["to"], "gt", p["from"]):
                raise Refused("R5", f"{p['name']} would be downgraded {p['from']} -> {p['to']}")
        if len(images) > 1:
            raise Refused("R23", f"the kernel step would add {len(images)} kernels — one at a time")
        # the target kernel: the new image, else the series meta-package's version (its image is already installed)
        if images:
            new = images[0][0]
            if images[0][1] + "-pve" != new:
                raise Refused("R23", f"the image version {images[0][1]} does not name the kernel {new}")
        else:
            metas = [(n, v) for n, v in upg.items() if re.match(r"^proxmox-kernel-[0-9]+\.[0-9]+$", n)]
            if len(metas) != 1:
                self.report["upgraded"], self.report["outcome_hint"] = [], "nothing"
                self.log("os-apply: DONE rc=0 upgraded=0 (the pending set names no kernel to boot)")
                return 0
            new = metas[0][1] + "-pve"
        if not KVER_RE.match(new) or not a.dpkg_cmp(new[:-4], "gt", old[:-4]):
            raise Refused("R23", f"the kernel {new} is not newer than the running {old}")
        ek = self.plan.get("expect_kver")
        if ek and ek != new:
            raise Refused("R23", f"the step would boot {new}, but the household was told about {ek}")
        need = a.download_bytes(["install", "--no-install-recommends"] + args)
        free = a.free_bytes()
        if free >= 0 and free < max(MIN_FREE, 3 * need):
            raise Refused("R8", f"free space {free} B is below max(500 MB, 3 x download {need} B)")
        # 1. the default = the kernel the box RUNS (it booted healthily), proved from grub.cfg BEFORE the install
        default_before = self.default_kver(self.grub_cfg())
        self.write_default(old)
        # 2. install (the kernel's own postinst runs update-grub; our default file keeps the default on `old`)
        t0 = time.time()
        rc, out, err = a.x(APT_ENV + ["apt-get", "-y", "-q"] + DPKG_OPTS + ["install", "--no-install-recommends"] + args)
        a.x(["apt-get", "clean"])
        if rc != 0:
            _, aud, _ = a.x(["dpkg", "--audit"])
            self.report["failed"] = {"rc": rc, "step": "install", "dpkg_audit": (aud.strip().splitlines() or ["clean"])[0],
                                     "tail": (out + err).strip().splitlines()[-3:]}
            self.log(f"os-apply: FAILED rc={rc} step=install (the default stays {old}; no flag written)")
            return 3
        self.report["upgraded"] = [{"name": x.split("=", 1)[0], "version": x.split("=", 1)[1]} for x in args]
        self.report["seconds"] = round(time.time() - t0, 1)
        # 3. prove: the image is there, the default is still `old`, the one-shot entry for `new` exists
        cfg = self.grub_cfg()
        if f"felhom-oneshot-{new}" not in cfg or self.default_kver(cfg) != old:
            self.r.host(["update-grub"], 300)
            cfg = self.grub_cfg()
        for f in (f"/boot/vmlinuz-{new}", f"/boot/initrd.img-{new}"):
            if not self.r.lexists(f):
                self.report["failed"] = {"rc": 3, "step": "verify", "reason": f"{f} is missing after the install"}
                return 3
        if f"felhom-oneshot-{new}" not in cfg or "felhom_next" not in cfg:
            self.report["failed"] = {"rc": 3, "step": "verify", "reason": f"grub.cfg has no one-shot entry for {new}"}
            return 3
        if self.default_kver(cfg) != old:
            self.report["failed"] = {"rc": 3, "step": "verify",
                                     "reason": f"the install moved the default to {self.default_kver(cfg)} — no flag written"}
            return 3
        # 4. the flag — the ONLY thing that makes the next boot use `new`, and only once
        self.set_flag(new)
        self.save_state({"phase": "staged", "step_id": self.plan.get("release_id"), "from": old, "to": new, "vmid": a.vmid,
                         "staged_at": _iso(self.r.now()), "authority": who, "default_before": default_before,
                         "packages": self.report["upgraded"], "self_revert_used": False})
        self.report["reboot_needed"] = True
        self.log(f"os-apply: KERNEL STAGED {old} -> {new} (default {old}, one-shot flag {new}); upgraded={len(args)} "
                 f"seconds={self.report['seconds']}")
        return 0

    def reboot_staged(self):
        st = self.state()
        self.phase_is(st, "staged")
        if self.flag() != st.get("to"):
            raise Refused("R22", f"the one-shot flag reads {self.flag()!r}, not {st.get('to')!r}")
        if self.running() != st.get("from"):
            raise Refused("R22", f"the box runs {self.running()!r}, not the step's old kernel {st.get('from')!r}")
        cfg = self.grub_cfg()
        if self.default_kver(cfg) != st["from"] or f"felhom-oneshot-{st['to']}" not in cfg:
            raise Refused("R20", "grub.cfg no longer keeps the old default with a one-shot entry for the new kernel")
        if self.setup_problems():
            raise Refused("R20", "; ".join(self.setup_problems()))
        self.check_guard()
        self.a.check_guest(self.a.vmid)
        st["health_before"] = self.a.health()
        st["phase"], st["rebooted_at"] = "oneshot", _iso(self.r.now())
        self.save_state(st)
        return self.reboot(f"one-shot boot of {st['to']} (the default stays {st['from']})")

    def after_boot(self):
        st = self.state()
        ph, run = st.get("phase"), self.running()
        old, new = st.get("from"), st.get("to")
        event = "none"
        if ph in ("oneshot", "staged", "judging") and run == new and new:
            if ph != "judging":
                st["phase"], st["judging_since"], event = "judging", _iso(self.r.now()), "judging"
            else:
                event = "judging"
        elif ph in ("oneshot", "judging") and run == old:
            # the new kernel did not come up, or crashed: GRUB already booted the default (the old kernel)
            self.clear_flag()
            st["phase"], st["result_at"], event = "fell_back", _iso(self.r.now()), "fell_back"
            st["reason"] = "the box came back on the old kernel by itself (the new one did not boot, or crashed)"
        elif ph == "reverting" and run == old:
            st["phase"], st["result_at"], event = "self_reverted", _iso(self.r.now()), "self_reverted"
        elif ph == "reverting" and run == new:
            st["phase"], st["result_at"], event = "revert_failed", _iso(self.r.now()), "revert_failed"
            st["reason"] = "the self-revert came back on the NEW kernel — never retried (one self-revert per step)"
        if event not in ("none",) and st.get("phase") != ph:
            self.save_state(st)
            self.log(f"os-apply: KERNEL AFTER-BOOT {ph} -> {st['phase']} running={run} ({old} -> {new})")
        self.report["kernel_event"] = event
        if event == "judging":
            self.report["health_before"] = st.get("health_before")  # the agent judges the boot against it
        return 0

    def good(self):
        st = self.state()
        self.phase_is(st, "judging")
        if self.running() != st.get("to"):
            raise Refused("R22", f"the box runs {self.running()!r}, not the new kernel {st.get('to')!r}")
        try:
            self.write_default(st["to"])
        except Refused:
            # put the old default back — the box must never be left without a proved default
            self.write_default(st["from"])
            raise
        self.clear_flag()
        st["phase"], st["result_at"] = "good", _iso(self.r.now())
        self.save_state(st)
        self.log(f"os-apply: KERNEL GOOD {st['to']} is the default now (was {st['from']})")
        return 0

    def revert(self):
        st = self.state()
        self.phase_is(st, "judging")
        if st.get("self_revert_used"):
            raise Refused("R22", "this kernel step already used its one self-revert")
        if self.running() != st.get("to"):
            raise Refused("R22", f"the box runs {self.running()!r}, not the new kernel {st.get('to')!r}")
        self.clear_flag()
        cfg = self.grub_cfg()
        if self.default_kver(cfg) != st.get("from"):
            raise Refused("R20", f"the GRUB default reads {self.default_kver(cfg)}, not the old {st.get('from')} — "
                                 f"no self-revert into an unknown kernel")
        reason = self.plan.get("reason")
        st["reason"] = reason[:300] if isinstance(reason, str) else "the one-shot boot was not healthy"
        st["self_revert_used"], st["phase"], st["reverted_at"] = True, "reverting", _iso(self.r.now())
        self.save_state(st)
        return self.reboot(f"self-revert to {st['from']}: {st['reason']}")

    def cancel(self):
        st = self.state()
        self.phase_is(st, "staged")
        self.clear_flag()
        st["phase"], st["result_at"], st["reason"] = "cancelled", _iso(self.r.now()), "cancelled before the reboot"
        self.save_state(st)
        self.log(f"os-apply: KERNEL CANCELLED {st.get('to')} (installed, never the default; flag cleared)")
        return 0


class Bundle:
    """The config bundle (R-840, `11` §5.4.2): every root-owned file the installer's step 5 writes, installed as ONE
    signed unit. Every check runs before the first write; a failed write or a failed self-check puts every previous
    copy back. Refusal codes: R1 (the bundle file), R3 (authority), R16 (a path outside BUNDLE_FILES), R17 (a trust
    file), R18 (a content check or a wrong sha)."""

    def __init__(self, apply):
        self.a, self.r = apply, apply.r
        self.report = apply.report

    # ---------- reading and checking ----------
    def load_file(self, path):
        d, base = os.path.dirname(path or ""), os.path.basename(path or "")
        if d != PLAN_DIR or not BUNDLE_RE.match(base) or ".." in path:
            raise Refused("R1", f"the bundle must be {PLAN_DIR}/bundle-<version>.json, got {path!r}")
        try:
            st = self.r.stat(path)
        except OSError as e:
            raise Refused("R1", f"cannot stat the bundle: {e}")
        if not stat.S_ISREG(st.st_mode):
            raise Refused("R1", "the bundle is not a regular file")
        if st.st_uid != self.r.agent_uid():
            raise Refused("R1", f"the bundle is not owned by {AGENT_USER}")
        if st.st_size > BUNDLE_MAX_BYTES:
            raise Refused("R1", "the bundle is larger than 4 MB")
        return self.r.read_bytes(path)

    def parse(self, data, want_sha, want_version=None):
        """The bundle bytes → [(dest, content, mode, check, policy)], every content check passed. Nothing is written."""
        import base64
        got = sha256_hex(data)
        if got != want_sha:
            raise Refused("R18", f"the bundle's sha256 is {got}, not the pinned {want_sha}")
        try:
            b = json.loads(data)
        except ValueError as e:
            raise Refused("R18", f"the bundle is not JSON: {e}")
        if not isinstance(b, dict) or b.get("format") != BUNDLE_FORMAT:
            raise Refused("R18", f"the bundle format is not {BUNDLE_FORMAT}")
        ver = b.get("agent_version")
        if not isinstance(ver, str) or not re.match(r"^[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.]+)?$", ver):
            raise Refused("R18", f"the bundle names no agent version ({ver!r})")
        if want_version is not None and ver != want_version:
            raise Refused("R18", f"the bundle is for agent {ver}, the signed job pins {want_version}")
        files = b.get("files")
        if not isinstance(files, list) or not files:
            raise Refused("R18", "the bundle lists no files")
        out, seen = [], set()
        for e in files:
            if not isinstance(e, dict) or not isinstance(e.get("path"), str):
                raise Refused("R18", "a bundle entry has no path")
            dest = e["path"]
            if dest in TRUST_PATHS:
                raise Refused("R17", f"{dest} is the trust root — no bundle may add, remove or change a signer")
            if dest not in BUNDLE_DESTS:
                raise Refused("R16", f"{dest} is not a path a bundle may write")
            if dest in seen:
                raise Refused("R18", f"{dest} is named twice")
            seen.add(dest)
            try:
                content = base64.b64decode(e.get("content_b64", ""), validate=True)
            except (ValueError, TypeError):
                raise Refused("R18", f"{dest}: the content is not base64")
            if sha256_hex(content) != e.get("sha256"):
                raise Refused("R18", f"{dest}: the content does not match its sha256")
            _, _, mode, check, policy = BUNDLE_DESTS[dest]
            self.check_content(dest, check, content)
            out.append((dest, content, mode, check, policy))
        order = [x[0] for x in BUNDLE_FILES]
        out.sort(key=lambda x: order.index(x[0]))  # the table's order: the sudoers files last
        return ver, out

    def check_content(self, dest, check, content):
        try:
            text = content.decode("utf-8")
        except UnicodeDecodeError:
            raise Refused("R18", f"{dest} is not UTF-8 text")
        if check in ("sudoers", "sh", "bash", "nft"):
            rc, msg = self.r.check_content(check, content)
            if rc != 0:
                raise Refused("R18", f"{dest} fails its {check} check: {msg}")
        elif check == "python":
            try:
                compile(text, dest, "exec")
            except SyntaxError as e:
                raise Refused("R18", f"{dest} is not valid Python: {e}")
        elif check in ("unit", "unit-nort", "agent-unit", "dropin"):
            if not re.search(r"^\[(Unit|Service|Timer)\]\s*$", text, re.M):
                raise Refused("R18", f"{dest} has no [Unit]/[Service]/[Timer] section")
            if check == "unit-nort" and re.search(r"^\s*RuntimeDirectory\s*=", text, re.M | re.I):
                raise Refused("R18", f"{dest} declares RuntimeDirectory= — the G1 incident; refused")
            if check == "agent-unit" and not re.search(r"^User=" + AGENT_USER + r"\s*$", text, re.M):
                raise Refused("R18", f"{dest} does not run the agent as {AGENT_USER}")
            if check == "dropin" and not re.search(r"^\[Unit\]\s*$", text, re.M):
                raise Refused("R18", f"{dest} must keep its start limits in [Unit] (SF-3)")
        # The route must survive the bundle: a sudoers without this wrapper's line, or a wrapper without the bundle
        # mode, would cut the box off from every later bundle.
        if dest == "/etc/sudoers.d/felhom-agent" and SELF_CHECK_SUDO_LINE not in text:
            raise Refused("R18", "the new sudoers no longer lets the agent call felhom-os-apply — the bundle route would end")
        if dest == "/usr/local/sbin/felhom-os-apply" and f'BUNDLE_OP = "{BUNDLE_OP}"' not in text:
            raise Refused("R18", "the new felhom-os-apply has no bundle mode — the bundle route would end")

    def live(self, dest):
        """(bytes or None, mode or None, uid or None) of what the box has now."""
        if not self.r.lexists(dest):
            return None, None, None
        st = self.r.stat(dest)
        if not stat.S_ISREG(st.st_mode):
            return b"", None, None  # a symlink or a directory: always replaced by the real file
        return self.r.read_bytes(dest), stat.S_IMODE(st.st_mode), st.st_uid

    def plan_writes(self, files):
        """Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt)."""
        oob = self.r.isdir(OOB_DIR)
        plan = []
        for dest, content, mode, check, policy in files:
            if policy == "oob" and not oob:
                plan.append((dest, content, mode, "skipped", None, None))
                continue
            old, old_mode, old_uid = self.live(dest)
            if policy == "if-absent" and old is not None:
                plan.append((dest, content, mode, "kept", old, old_mode))
            elif old == content and old_mode == mode and old_uid == 0:
                plan.append((dest, content, mode, "same", old, old_mode))
            else:
                plan.append((dest, content, mode, "write", old, old_mode))
        return plan

    # ---------- the two entries ----------
    def from_plan(self, plan):
        """A signed agent_config_update, from the agent (sudo, the --plan line)."""
        data = self.load_file(plan.get("bundle"))
        trust = self.a.load_trust()
        signers, bootstrap = TRUST_SIGNERS, False
        if not self.r.lexists(TRUST_SIGNERS):
            signers, bootstrap = PINNED_SIGNERS, True
            self.r.log(f"os-apply: BUNDLE {TRUST_SIGNERS} is missing — verifying against the installer's pinned key "
                       f"{PINNED_OPERATOR_KEY_ID} only")
        params, nonce, exp = self.a.verify_signed(plan.get("signed"), trust, op_name=BUNDLE_OP,
                                                  signers=None if not bootstrap else signers, burn=False)
        sha, ver = params.get("bundle_sha256"), params.get("agent_version")
        if not isinstance(sha, str) or not re.match(r"^[0-9a-f]{64}$", sha) or not isinstance(ver, str):
            raise Refused("R3", "the signed job does not pin agent_version and bundle_sha256")
        ver, files = self.parse(data, sha, ver)
        self.a.burn_nonce(nonce, exp)
        return self.install(ver, sha, files, "signed", bootstrap)

    def from_installer(self, path, sha):
        """The installer (root, never through sudo): the hub manifest pinned the sha; no signature."""
        try:
            data = self.r.read_bytes(path)
        except OSError as e:
            raise Refused("R1", f"cannot read the bundle: {e}")
        ver, files = self.parse(data, sha)
        return self.install(ver, sha, files, "installer", False)

    # ---------- install, self-check, undo ----------
    def install(self, ver, sha, files, authority, bootstrap):
        log = self.r.log
        plan = self.plan_writes(files)
        counts = {k: sum(1 for p in plan if p[3] == k) for k in ("write", "same", "kept", "skipped")}
        log(f"os-apply: BUNDLE START agent={ver} sha={sha[:16]} authority={authority} files={len(plan)} "
            f"write={counts['write']} same={counts['same']} kept={counts['kept']} skipped={counts['skipped']}")
        stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(self.r.now()))
        prev = os.path.join(BUNDLE_PREV_DIR, f"{stamp}-before-{ver}")
        done = []  # (dest, old bytes or None, old mode)
        rep = {"agent_version": ver, "sha256": sha, "authority": authority, "prev_dir": prev,
               "written": [p[0] for p in plan if p[3] == "write"], "kept": [p[0] for p in plan if p[3] == "kept"],
               "skipped": [p[0] for p in plan if p[3] == "skipped"], "same": counts["same"]}
        self.report["bundle"] = rep
        try:
            for dest, content, mode, action, old, old_mode in plan:
                if action != "write":
                    continue
                if old is not None:
                    self.r.put_file(prev + dest, old, 0o600)
                done.append((dest, old, old_mode))
                self.r.put_file(dest, content, mode)
                log(f"os-apply: BUNDLE WROTE {dest} ({'replaced' if old is not None else 'new'})")
            self.after_install(plan)
            rep["self_check"] = self.self_check(plan)
        except (Refused, OSError, subprocess.SubprocessError) as e:
            reason = e.reason if isinstance(e, Refused) else str(e)
            log(f"os-apply: BUNDLE FAILED — {reason}; putting {len(done)} previous file(s) back")
            rep["rolled_back"] = self.undo(done)
            rep["failed"] = reason[:300]
            self.report["failed"] = {"rc": 3, "step": "bundle", "reason": reason[:300]}
            return 3
        rep["signers_created"] = False
        if bootstrap and not self.r.lexists(TRUST_SIGNERS):
            self.r.put_file(TRUST_SIGNERS, pinned_signers_line().encode(), 0o644)
            rep["signers_created"] = True
            log(f"os-apply: BUNDLE created {TRUST_SIGNERS} with exactly the pinned key {PINNED_OPERATOR_KEY_ID}")
        record = {"format": BUNDLE_FORMAT, "agent_version": ver, "bundle_sha256": sha, "authority": authority,
                  "installed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.r.now())),
                  "files": {p[0]: (sha256_hex(p[1]) if p[3] in ("write", "same") else
                                   (sha256_hex(p[4]) if p[3] == "kept" else "skipped")) for p in plan}}
        self.r.put_file(BUNDLE_RECORD, (json.dumps(record, indent=2, sort_keys=True) + "\n").encode(), 0o644)
        self.prune()
        log(f"os-apply: BUNDLE DONE agent={ver} written={counts['write']} same={counts['same']} "
            f"self-check=ok signers-created={rep['signers_created']}")
        return 0

    def after_install(self, plan):
        written = {p[0] for p in plan if p[3] == "write"}
        dests = {p[0] for p in plan if p[3] != "skipped"}
        if any(d.startswith("/etc/systemd/system/") for d in written):
            self.must(["systemctl", "daemon-reload"], "systemctl daemon-reload")
        if "/etc/tmpfiles.d/felhom-privsep.conf" in written:
            self.must(["systemd-tmpfiles", "--create", "/etc/tmpfiles.d/felhom-privsep.conf"], "systemd-tmpfiles")
        # `enable --now` starts a unit that is not running and leaves a running one alone (no restart). For the crash
        # guard that runs its boot step once on a box that never had it: kernel.panic for THIS boot, as the boot unit
        # would set it (`11` §5.9); with no earlier state it is the "first" boot, never counted as an unclean one.
        if "/etc/systemd/system/felhom-crash-guard.service" in dests:
            self.must(["systemctl", "enable", "--now", "felhom-crash-guard.service", "felhom-crash-guard-check.timer"],
                      "enable the crash guard")
        if "/etc/systemd/system/felhom-mgmt-watchdog.timer" in dests:
            self.must(["systemctl", "enable", "--now", "felhom-mgmt-watchdog.timer"], "enable the mgmt watchdog timer")

    def must(self, argv, what):
        rc, out, err = self.r.host(argv, 120)
        if rc != 0:
            raise Refused("R18", f"{what} failed (rc={rc}): {(out + err).strip()[-200:]}")

    def self_check(self, plan):
        """After the install, before the record: the box still has a working route and working wrappers."""
        sc = {}
        self.must(["visudo", "-c"], "visudo -c (the whole sudoers)")
        sc["visudo"] = "ok"
        rc, out, err = self.r.host(["sudo", "-n", "-l", "-U", AGENT_USER], 60)
        if rc != 0 or "/usr/local/sbin/felhom-os-apply --plan" not in out:
            raise Refused("R18", f"sudo -l for {AGENT_USER} no longer lists felhom-os-apply (rc={rc})")
        sc["sudo_l"] = "felhom-os-apply listed"
        rc, out, err = self.r.host(["/usr/bin/python3", "/usr/local/sbin/felhom-os-apply", "--self-check"], 60)
        if rc != 0 or f"bundle-format={BUNDLE_FORMAT}" not in out:
            raise Refused("R18", f"the installed felhom-os-apply does not answer its self-check (rc={rc})")
        sc["os_apply"] = out.strip()[:120]
        rc, out, err = self.r.host(["/bin/sh", "/usr/local/sbin/felhom-selfupdate-guarded"], 60)
        if rc != 2 or "usage" not in (out + err):
            raise Refused("R18", f"the installed felhom-selfupdate-guarded does not answer with its usage (rc={rc})")
        sc["selfupdate"] = "usage ok"
        if "/usr/local/sbin/felhom-priv-apply" in {p[0] for p in plan if p[3] != "skipped"}:
            rc, out, err = self.r.host(["/usr/bin/python3", "/usr/local/sbin/felhom-priv-apply", "--self-check"], 60)
            if rc != 0 or "felhom-priv-apply ok" not in out:
                raise Refused("R18", f"the installed felhom-priv-apply does not answer its self-check (rc={rc})")
            sc["priv_apply"] = out.strip()[:80]
        dests = {p[0] for p in plan if p[3] != "skipped"}
        if "/usr/local/sbin/felhom-crash-guard" in dests:
            rc, out, err = self.r.host(["/usr/local/sbin/felhom-crash-guard", "status"], 60)
            try:
                st = json.loads(out)
            except ValueError:
                st = None
            if rc != 0 or not isinstance(st, dict):
                raise Refused("R18", f"the crash guard does not answer its status (rc={rc})")
            try:
                panic = int(self.r.read_file("/proc/sys/kernel/panic").strip())
            except (OSError, ValueError):
                panic = None
            if panic != st.get("kernel_panic"):
                raise Refused("R18", f"kernel.panic is {panic}, the crash guard set {st.get('kernel_panic')}")
            sc["crash_guard"] = {"armed": st.get("armed"), "kernel_panic": panic}
        return sc

    def undo(self, done):
        """Put every previous copy back (newest write first); remove files the bundle created."""
        back = []
        for dest, old, old_mode in reversed(done):
            try:
                if old is None:
                    self.r.remove(dest)
                else:
                    self.r.put_file(dest, old, old_mode if old_mode is not None else 0o644)
                back.append(dest)
            except OSError as e:
                self.r.log(f"os-apply: BUNDLE UNDO could not restore {dest}: {e}")
        if any(d.startswith("/etc/systemd/system/") for d in back):
            self.r.host(["systemctl", "daemon-reload"], 120)
        rc, _, _ = self.r.host(["visudo", "-c"], 60)
        self.r.log(f"os-apply: BUNDLE UNDO restored={len(back)} visudo-after={'ok' if rc == 0 else 'FAILED'}")
        return back

    def prune(self):
        names = [n for n in self.r.list_dir(BUNDLE_PREV_DIR)]
        for n in names[:-BUNDLE_KEEP]:
            self.r.rmtree(os.path.join(BUNDLE_PREV_DIR, n))

    # ---------- the facts mode's view ----------
    def state(self):
        """What the box runs: the record, and every bundle path's live sha256 against it (drift = changed by hand)."""
        rec = None
        try:
            rec = json.loads(self.r.read_file(BUNDLE_RECORD))
        except (OSError, ValueError):
            rec = None
        live = {}
        for dest, *_ in BUNDLE_FILES:
            try:
                data, _, _ = self.live(dest)
            except OSError:
                data = None
            live[dest] = sha256_hex(data) if data is not None else "absent"
        out = {"version": (rec or {}).get("agent_version", "none"), "live": live}
        if rec:
            out["installed_at"] = rec.get("installed_at")
            out["bundle_sha256"] = rec.get("bundle_sha256")
            want = rec.get("files") or {}
            out["drift"] = sorted(d for d, h in want.items() if h != "skipped" and live.get(d) != h)
        out["signers_present"] = self.r.lexists(TRUST_SIGNERS)
        return out


def main(argv, runner=None, environ=None):
    r = runner or Runner()
    env = os.environ if environ is None else environ
    if argv[1:] == ["--self-check"]:
        # The bundle's self-check runs the NEW wrapper this way: it parses and starts. Reads nothing, changes nothing.
        print(f"felhom-os-apply ok bundle-format={BUNDLE_FORMAT} files={len(BUNDLE_FILES)}")
        return 0
    if len(argv) == 5 and argv[1] == "--install-bundle" and argv[3] == "--sha256":
        # The INSTALLER's entry (root, a fresh box). Unreachable through sudo: the sudoers line pins argv[1] to --plan,
        # and a sudo caller is refused here as well (sudo always sets SUDO_UID).
        a = Apply(r, "")
        if "SUDO_UID" in env:
            r.log("os-apply: REFUSED: R1 --install-bundle is the installer's entry, never through sudo")
            print("OSAPPLY-REPORT " + json.dumps({"refused": {"code": "R1", "reason": "install-bundle via sudo"}}))
            return 2
        sha = argv[4]
        if not re.match(r"^[0-9a-f]{64}$", sha):
            print("OSAPPLY-REPORT " + json.dumps({"refused": {"code": "R18", "reason": "sha256 is not 64 lowercase hex"}}))
            return 2
        a.report["mode"] = "bundle"
        try:
            rc = Bundle(a).from_installer(argv[2], sha)
        except Refused as e:
            r.log(f"os-apply: REFUSED: {e.code} {e.reason}")
            a.report["refused"] = {"code": e.code, "reason": e.reason}
            rc = 2
        print("OSAPPLY-REPORT " + json.dumps(a.report, sort_keys=True))
        return rc
    if len(argv) != 3 or argv[1] != "--plan":
        r.log("os-apply: REFUSED: R1 usage: felhom-os-apply --plan /var/lib/felhom-agent/os/plan-<id>.json")
        print("OSAPPLY-REPORT " + json.dumps({"refused": {"code": "R1", "reason": "usage"}}))
        return 2
    a = Apply(r, argv[2])
    t0 = time.time()
    try:
        rc = a.run()
    except Refused as e:
        r.log(f"os-apply: REFUSED: {e.code} {e.reason}")
        a.report["refused"] = {"code": e.code, "reason": e.reason}
        rc = 2
    except subprocess.TimeoutExpired as e:
        r.log(f"os-apply: FAILED rc=124 step=timeout — {e.cmd}")
        a.report["failed"] = {"rc": 124, "timeout": str(e.cmd)[:200]}
        rc = 3
    a.report["pass_seconds"] = round(time.time() - t0, 1)
    if a.report.get("mode") == "apply":
        # R-868: BEFORE the stdout line — a killed agent never reads stdout, and this copy is how its report still
        # reaches the hub (the agent sends an unsent copy when it starts, and deletes it once sent).
        try:
            r.save_report(argv[2], a.report)
        except Exception as e:
            r.log(f"os-apply: the report copy could not be saved (the run is unaffected): {e}")
    try:
        print("OSAPPLY-REPORT " + json.dumps(a.report, sort_keys=True), flush=True)
    except OSError:
        # R-868: nobody reads stdout any more (the agent was killed); the copy above carries the report.
        sys.stdout = open(os.devnull, "w")
    return rc


if __name__ == "__main__":
    if os.geteuid() != 0:
        print("felhom-os-apply: must run as root (via sudo)", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv))
