OS updates host fast lane + true tunnel status + fast leg: wrapper host layer (R12 appliance proof from the root-owned install record, R14 kernel/boot/firmware refused), select pending-fast, one call per layer, host-side version checks, restart scan only after an install, reboot-needed for PID 1/lxc-start; the leg runs the host step after a healthy guest step; GuestTunnelProber reads the cloudflared container + its readiness check (R-841)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS

configs/felhom-os-apply

@@ -1,5 +1,5 @@
 #!/usr/bin/python3
-# felhom-os-apply — the ROOT half of the agent's operating-system update leg (`11-os-updates.md` §5.4.1).
+# 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
@@ -8,22 +8,28 @@
 #
 # 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, only the box's own customer guest.
-# Package signatures stay Debian's: apt checks every Release file against the guest's keyring, including the
-# snapshot.debian.org fallback (decision 79). Nothing here is overridable from the environment.
+# 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.
 #
-# THIS RELEASE: layer "guest", lane "fast" only. The host layer and the slow lane exist in the interface and are
-# REFUSED (R3, R12) until `11` §8 steps 3 and 5 enable them.
+# LAYERS (agent v0.141.0): "guest" (the customer LXC, entered with `pct exec`) and "host" (this Proxmox host, run
+# directly). LANE: "fast" only — the slow lane (kernel, Proxmox, Docker) is REFUSED (R3, R14) until `11` §8 steps 5–6.
 #
 # Modes (plan field "mode"):
-#   inventory   read-only for packages: `apt-get update` in the guest, then report what is installed (with origin),
-#               what is pending, restart-needed and health. Installs nothing.
-#   apply       repair first, check every refusal on an `apt-get -s` simulation of EXACTLY name=version, then
-#               install, clean, and report the same as inventory.
+#   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).
 # 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 json
 import os
 import re
@@ -46,6 +52,13 @@ 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
+# 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"
+# 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)")
+HOST_SERVICES = ["pveproxy", "pvedaemon", "pvestatd", "pve-cluster", "felhom-agent"]
 
 
 class Refused(Exception):
@@ -57,8 +70,8 @@ class Refused(Exception):
 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):
-        p = subprocess.run(argv, capture_output=True, text=True, timeout=timeout)
+    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):
@@ -75,6 +88,16 @@ class Runner:
         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 log(self, line):
         print(line, file=sys.stderr, flush=True)
         try:
@@ -117,8 +140,9 @@ class Apply:
         mode = plan.get("mode", "apply")
         if mode not in ("apply", "inventory", "health"):
             raise Refused("R11", f"unknown mode {mode!r}")
-        if plan.get("layer") != "guest":
-            raise Refused("R12", f"layer {plan.get('layer')!r} is refused in this release (guest only)")
+        layer = plan.get("layer")
+        if layer not in ("guest", "host"):
+            raise Refused("R12", f"layer {layer!r} is not guest or host")
         if plan.get("lane", "fast") != "fast":
             raise Refused("R3", "the slow lane is refused in this release")
         vmid = plan.get("vmid")
@@ -129,9 +153,16 @@ class Apply:
             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"):
+            raise Refused("R11", f"unknown select {select!r}")
         pk = plan.get("packages", [])
-        if not isinstance(pk, list) or (mode == "apply" and not pk):
-            raise Refused("R11", "packages must be a non-empty list in apply mode")
+        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 == "pending-fast" and pk:
+            raise Refused("R11", "select pending-fast takes no package list")
         seen = set()
         for e in pk:
             if not isinstance(e, dict):
@@ -146,10 +177,27 @@ class Apply:
             seen.add(n)
             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, vmid
+        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 check_guest(self, vmid):
         if vmid in RESERVED_VMIDS:
@@ -169,12 +217,24 @@ class Apply:
         if rc != 0 or "running" not in out:
             raise Refused("R10", f"vmid {vmid} is not running")
 
-    # ---------- guest helpers ----------
-    def g(self, argv, timeout=1800):
+    # ---------- target helpers ----------
+    def x(self, argv, timeout=1800):
+        """Run in the TARGET layer: the guest via pct exec, or the host directly."""
+        if self.layer == "host":
+            return self.r.host(argv, timeout)
         return self.r.guest(self.vmid, argv, timeout)
 
+    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.g(["dpkg-query", "-W", "-f", "${Package}\t${Version}\t${db:Status-Abbrev}\n"])
+        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")
@@ -182,21 +242,20 @@ class Apply:
                 res[parts[0]] = parts[1]
         return res
 
-    def dpkg_cmp(self, a, op, b):
-        rc, _, _ = self.g(["dpkg", "--compare-versions", a, op, b])
-        return rc == 0
-
-    def madison(self, name):
-        rc, out, _ = self.g(["apt-cache", "madison", name])
-        vs = set()
+    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] == name:
-                vs.add(f[1])
-        return vs
+            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.g(APT_ENV + ["apt-get", "-s", "-q"] + 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)
@@ -213,17 +272,17 @@ class Apply:
         return {o.strip().split(":")[0] for o in origin.split(",") if o.strip()}
 
     def free_bytes(self):
-        rc, out, _ = self.g(["df", "-B1", "--output=avail", "/"])
+        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.g(["fuser", "/var/lib/dpkg/lock-frontend", "/var/lib/dpkg/lock"])
+        rc, out, _ = self.x(["fuser", "/var/lib/dpkg/lock-frontend", "/var/lib/dpkg/lock"])
         return rc == 0 and out.strip() != ""
 
-    def health(self):
+    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", "--format", "{{.Names}}\t{{.State}}\t{{.Status}}"], timeout=60)
         cont = {}
@@ -238,21 +297,34 @@ class Apply:
                 "controller": cont.get("felhom-controller", {}).get("health", "absent"),
                 "network_ok": nrc == 0}
 
-    def restart_needed(self):
-        """Processes still mapping deleted files, OUTSIDE docker containers (C11)."""
-        script = ('for p in /proc/[0-9]*; do grep -q "(deleted)" $p/maps 2>/dev/null || continue; '
-                  'grep -q "docker" $p/cgroup 2>/dev/null && continue; echo "${p#/proc/} $(cat $p/comm 2>/dev/null)"; done')
-        rc, out, _ = self.g(["sh", "-c", script], timeout=120)
-        procs = sorted({l.split(" ", 1)[1] for l in out.splitlines() if " " in l})
-        pid1 = any(l.split(" ", 1)[0] == "1" for l in out.splitlines())
-        return procs, pid1
+    def health(self):
+        if self.layer == "guest":
+            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 inventory(self):
-        inst = self.installed()
+    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 = "docker" if self.layer == "guest" else "lxc"
+        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 = {}
-        for i in range(0, len(names), 200):
-            rc, out, _ = self.g(["apt-cache", "policy"] + names[i:i + 200])
+        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(" "):
@@ -270,10 +342,12 @@ class Apply:
                     continue
                 if star and not re.match(r"^[0-9-]+ ", s):
                     star = False
-        # Map an index URL to an origin name the hub understands.
+
         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:
@@ -282,6 +356,7 @@ class Apply:
                 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"],
@@ -291,51 +366,70 @@ class Apply:
     # ---------- the run ----------
     def run(self):
         plan = self.load_plan()
-        self.mode, self.vmid = self.check_plan(plan)
-        self.report.update(mode=self.mode, release_id=plan.get("release_id"), vmid=self.vmid)
+        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)
+        if self.layer == "host":
+            self.check_appliance()
         self.check_guest(self.vmid)
         log = self.r.log
         if self.mode == "health":
             self.report["health"] = self.health()
             return 0
-        log(f"os-apply: START release={plan.get('release_id')} layer=guest:{self.vmid} lane=fast mode={self.mode} packages={len(plan.get('packages', []))}")
+        log(f"os-apply: START release={plan.get('release_id')} layer={self.layer}" +
+            (f":{self.vmid}" if self.layer == "guest" else "") +
+            f" lane=fast mode={self.mode} select={self.select} packages={len(plan.get('packages', []))}")
         if self.apt_lock_held():
-            raise Refused("R9", "another apt/dpkg holds the lock in the guest")
+            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.g(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
+        rc, out, err = self.x(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
         if rc != 0:
-            raise Refused("R7", f"apt-get update failed in the guest: {(out + err).strip().splitlines()[-1:]}")
+            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 = self.apply(plan)
+            rc, installed_after = self.apply(plan)
             if rc:
                 return rc
-        procs, pid1 = self.restart_needed()
-        self.report.update(self.inventory())
-        self.report["restart_needed"] = procs
-        self.report["docker_restart_needed"] = any(p in ("dockerd", "containerd") for p in procs)
-        self.report["reboot_needed"] = pid1
+        self.report.update(self.inventory(installed_after))
         self.report["health_after"] = self.health()
         return 0
 
     def repair(self):
-        rc, before, _ = self.g(["dpkg", "--audit"])
-        self.g(APT_ENV + ["dpkg", "--configure", "-a", "--force-confold"])
-        rc2, out, err = self.g(APT_ENV + ["apt-get", "-f", "install", "-y", "-q"] + DPKG_OPTS)
-        _, after, _ = self.g(["dpkg", "--audit"])
+        rc, before, _ = self.x(["dpkg", "--audit"])
         configured = len([l for l in before.splitlines() if l.startswith(" ")])
-        fixed = len(re.findall(r"^Setting up ", out, re.M))
+        fixed = 0
+        after = ""
+        if before.strip():  # nothing half-done → nothing to run (R-845: two calls saved on every clean pass)
+            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, _ = self.x(["dpkg", "--audit"])
+            fixed = len(re.findall(r"^Setting up ", out, re.M))
         self.report["repair"] = {"half_configured_before": configured, "fixed": fixed, "clean_after": after.strip() == ""}
         self.r.log(f"os-apply: REPAIR configured={configured} fixed={fixed}")
         if after.strip():
             raise Refused("R13", "dpkg is still broken after the repair: " + after.strip().splitlines()[0])
 
+    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 apply(self, plan):
         log = self.r.log
+        packages = plan["packages"] if self.select == "listed" else self.pending_fast()
         inst = self.installed()
         upgrade, already, notinst = [], 0, 0
-        for e in plan["packages"]:
+        for e in packages:
             n, v = e["name"], e["version"]
             if n not in inst:
                 notinst += 1
@@ -345,13 +439,18 @@ class Apply:
                 continue
             upgrade.append((n, v))
         from_snap = 0
-        missing = [(n, v) for n, v in upgrade if v not in self.madison(n)]
+        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)
-            still = [(n, v) for n, v in missing if v not in self.madison(n)]
+            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}")
@@ -362,7 +461,7 @@ class Apply:
             if not upgrade:
                 self.report["upgraded"] = []
                 log("os-apply: DONE rc=0 seconds=0 upgraded=0 (nothing to do)")
-                return 0
+                return 0, inst
             args = ["install", "--only-upgrade", "--no-install-recommends"] + [f"{n}={v}" for n, v in upgrade]
             rc, sim, remv, text = self.simulate(args)
             if rc != 0:
@@ -382,12 +481,14 @@ class Apply:
                     raise Refused("R5", f"{p['name']} would be downgraded {p['from']} -> {p['to']}")
                 if not self.origin_name(p["origin"]) & set(FAST_ORIGINS):
                     raise Refused("R2", f"{p['name']} would come from {p['origin']}, not Debian")
+                if self.layer == "host" 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.g(APT_ENV + ["apt-get", "-y", "-q"] + DPKG_OPTS + args)
+            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
@@ -404,23 +505,27 @@ class Apply:
                     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.g(["apt-get", "clean"])
+            self.x(["apt-get", "clean"])
             if rc != 0:
-                _, aud, _ = self.g(["dpkg", "--audit"])
+                _, 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
+                return 3, None
             self.report["upgraded"] = [{"name": n, "version": v} for n, v in upgrade]
             self.report["seconds"] = round(secs, 1)
-            log(f"os-apply: DONE rc=0 seconds={secs:.1f} upgraded={len(upgrade)}")
-            return 0
+            procs, reboot = self.restart_needed()  # only after an install (R-845)
+            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):
-        rc, out, _ = self.g(APT_ENV + ["apt-get", "-s", "-o", "Debug::NoLocking=1", "--print-uris", "-q"] + args)
+        rc, out, _ = self.x(APT_ENV + ["apt-get", "-s", "-o", "Debug::NoLocking=1", "--print-uris", "-q"] + args)
         total = 0
         for l in out.splitlines():
             m = re.match(r"^'[^']+' \S+ ([0-9]+) ", l)
@@ -429,33 +534,22 @@ class Apply:
         return total
 
     def add_snapshot_sources(self, snap):
-        rc, out, _ = self.g(["sh", "-c", ". /etc/os-release && echo $VERSION_CODENAME"])
+        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 guest's Debian codename ({code!r})")
+            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.guest_write(self.vmid, SNAPSHOT_LIST, body)
+        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.g(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
+        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.g(["rm", "-f", SNAPSHOT_LIST])
-        self.g(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
-
-
-def guest_write(self, vmid, path, body):
-    """Write a small text file inside the guest via `pct exec … tee` (stdin), never via a shell string."""
-    p = subprocess.run(["/usr/sbin/pct", "exec", str(vmid), "--", "tee", path], input=body,
-                       capture_output=True, text=True, timeout=60)
-    if p.returncode != 0:
-        raise Refused("R7", f"could not write {path} in the guest")
-
-
-Runner.guest_write = guest_write
+        self.x(["rm", "-f", SNAPSHOT_LIST])
+        self.x(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
 
 
 def main(argv, runner=None):
@@ -465,6 +559,7 @@ def main(argv, runner=None):
         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:
@@ -475,6 +570,7 @@ def main(argv, runner=None):
         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)
     print("OSAPPLY-REPORT " + json.dumps(a.report, sort_keys=True))
     return rc
 

configs/test_felhom_os_apply.py

@@ -61,6 +61,9 @@ class Fake:
         self.written = {}
         self.status = "status: running"
         self.lock_held = False
+        self.services = {}
+        self.files[osapply.INSTALL_STATE] = json.dumps({"mode": "appliance"})
+        self.stats[osapply.INSTALL_STATE] = St(mode=statmod.S_IFREG | 0o644, uid=0)
 
     # Runner interface
     def read_file(self, p):
@@ -81,14 +84,19 @@ class Fake:
     def log(self, line):
         self.logs.append(line)
 
-    def host(self, argv, timeout=600):
+    def host(self, argv, timeout=600, stdin=None):
         self.calls.append(("host", argv))
-        if argv[1] == "status":
+        if argv[0] == "/usr/sbin/pct" and argv[1] == "status":
             return 0, self.status + "\n", ""
-        return 1, "", "unexpected host call"
+        if argv[0] == "dpkg" and argv[1] == "--compare-versions":
+            return (0 if dpkg_cmp(argv[2], argv[3], argv[4]) else 1), "", ""
+        if argv[0] == "systemctl" and argv[1] == "is-active":
+            return 0, "\n".join(self.services.get(s, "active") for s in argv[2:]) + "\n", ""
+        return self.emulate(argv)
 
-    def guest_write(self, vmid, path, body):
+    def write_file(self, layer, vmid, path, body):
         self.written[path] = body
+        self.write_layer = layer
         if path == osapply.SNAPSHOT_LIST:
             self.snap_active = True
 
@@ -100,6 +108,9 @@ class Fake:
 
     def guest(self, vmid, argv, timeout=1800):
         self.calls.append(("guest", vmid, argv))
+        return self.emulate(argv)
+
+    def emulate(self, argv):
         a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"]
         cmd = a[0]
         if cmd == "dpkg-query":
@@ -113,7 +124,8 @@ class Fake:
         if cmd == "fuser":
             return (0, " 123", "") if self.lock_held else (1, "", "")
         if cmd == "apt-cache" and a[1] == "madison":
-            return 0, "".join(f" {a[2]} | {v} | http://deb.debian.org trixie/main amd64 Packages\n" for v in self.avail(a[2])), ""
+            self.madison_calls = getattr(self, "madison_calls", 0) + 1
+            return 0, "".join(f" {n} | {v} | http://deb.debian.org trixie/main amd64 Packages\n" for n in a[2:] for v in self.avail(n)), ""
         if cmd == "apt-cache" and a[1] == "policy":
             out = ""
             for n in a[2:]:
@@ -146,6 +158,8 @@ class Fake:
         if cmd == "sh":
             if "os-release" in a[2]:
                 return 0, "trixie\n", ""
+            if "(deleted)" in a[2]:
+                return 0, getattr(self, "restart_out", ""), ""
             return 0, "", ""
         if cmd == "rm":
             self.snap_active = False
@@ -156,6 +170,9 @@ class Fake:
         if "--print-uris" in a:
             return 0, "'http://x/libc6.deb' libc6.deb 4000000 SHA256:x\n", ""
         if "dist-upgrade" in a:
+            if getattr(self, "pending_sim", None) is not None and not getattr(self, "_pending_used", False):
+                self._pending_used = True
+                return 0, "\n".join(self.pending_sim) + "\n", ""
             return 0, "Inst bash [5.2.37-2+b9] (5.2.37-2+b10 Debian:13.7/stable [amd64])\n", ""
         out = ""
         for x in a:
@@ -209,7 +226,7 @@ class Happy(unittest.TestCase):
         self.assertEqual(rc, 0, rep)
         self.assertEqual(f.installed["libc6"], "2.41-12+deb13u3")
         self.assertIn("installed", rep)
-        self.assertIn("restart_needed", rep)
+        self.assertNotIn("restart_needed", rep, "the restart scan runs only after an install (R-845)")
 
     def test_health_mode(self):
         f = Fake()
@@ -419,11 +436,44 @@ class Refusals(unittest.TestCase):
         f.plan["packages"][0]["name"] = "--purge"
         self.refused(f, "R11")
 
-    def test_R12_host_layer(self):
+    def test_R12_unknown_layer(self):
+        f = Fake()
+        f.plan["layer"] = "vm"
+        self.refused(f, "R12")
+
+    def test_R12_host_on_a_byo_box(self):
         f = Fake()
         f.plan["layer"] = "host"
+        f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"})
         self.refused(f, "R12")
 
+    def test_R12_host_without_an_install_record(self):
+        f = Fake()
+        f.plan["layer"] = "host"
+        del f.stats[osapply.INSTALL_STATE]
+        self.refused(f, "R12")
+
+    def test_R12_host_record_not_root_owned(self):
+        # the agent can write agent.json's deployment_mode; only a ROOT-owned record proves anything
+        f = Fake()
+        f.plan["layer"] = "host"
+        f.stats[osapply.INSTALL_STATE] = St(mode=statmod.S_IFREG | 0o644, uid=999)
+        self.refused(f, "R12")
+
+    def test_R14_kernel_package_in_a_host_plan(self):
+        f = Fake()
+        f.plan["layer"] = "host"
+        f.plan["packages"].append({"name": "linux-image-amd64", "version": "6.12.1-1", "origin": "Debian"})
+        self.refused(f, "R14")
+
+    def test_R14_kernel_package_pulled_by_the_simulation(self):
+        f = Fake()
+        f.plan["layer"] = "host"
+        f.extra_sim = ["Inst grub-common [2.12-9] (2.12-10 Debian:13.7/stable [amd64])"]
+        f.installed["grub-common"] = "2.12-9"
+        f.plan["packages"].append({"name": "grub-common", "version": "2.12-10", "origin": "Debian"})
+        self.refused(f, "R14")
+
     def test_R13_repair_does_not_fix_it(self):
         f = Fake()
         f.dpkg_audit = "The following packages are broken\n perl\n"
@@ -453,6 +503,79 @@ class Conffiles(unittest.TestCase):
         self.assertFalse(any("kept /etc/debian_version" in l for l in f.logs), "an updated file must not be reported as kept")
 
 
+class HostLayer(unittest.TestCase):
+    def test_host_runs_on_the_host_not_in_the_guest(self):
+        f = Fake()
+        f.plan["layer"] = "host"
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        inst = [c for c in f.calls if c[0] == "host" and "install" in c[1] and "-s" not in c[1] and "-f" not in c[1]]
+        self.assertTrue(inst, "the host install must run on the host")
+        self.assertFalse([c for c in f.calls if c[0] == "guest" and "install" in c[2]], "nothing installed in the guest")
+        self.assertEqual(sorted(rep["health_after"]["host_services"]), sorted(osapply.HOST_SERVICES))
+        self.assertTrue(rep["health_after"]["guest_running"])
+
+    def test_pending_fast_skips_proxmox_docker_and_kernel(self):
+        f = Fake()
+        f.plan["layer"] = "host"
+        f.plan["select"] = "pending-fast"
+        f.plan["packages"] = []
+        f.installed.update({"pve-manager": "9.2.2", "linux-image-amd64": "6.12.1", "docker-ce": "29.7"})
+        f.pending_sim = [
+            "Inst libc6 [2.41-12+deb13u3] (2.41-12+deb13u4 Debian:13.7/stable [amd64])",
+            "Inst openssl [3.5.6-1~deb13u1] (3.5.7-1~deb13u3 Debian:13.7/stable, Debian-Security:13/stable-security [amd64])",
+            "Inst pve-manager [9.2.2] (9.2.21 Proxmox Debian Repository:stable [amd64])",
+            "Inst linux-image-amd64 [6.12.1] (6.12.9 Debian:13.7/stable [amd64])",
+            "Inst docker-ce [29.7] (29.8 Docker CE:trixie [amd64])",
+            "Inst brand-new (1.0 Debian:13.7/stable [amd64])",
+        ]
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        got = sorted(u["name"] for u in rep["upgraded"])
+        self.assertEqual(got, ["libc6", "openssl"], "pending-fast must take only installed, Debian-origin, non-kernel packages")
+
+    def test_reboot_needed_when_pid1_or_lxc_start(self):
+        f = Fake()
+        f.plan["layer"] = "host"
+        f.restart_out = "1 systemd\n2101 lxc-start\n530 sshd\n"
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        self.assertTrue(rep["reboot_needed"])
+        self.assertIn("lxc-start", rep["restart_needed"])
+
+    def test_reboot_needed_for_lxc_start_alone(self):
+        # lxc-start runs the guest; only a guest restart (or a host reboot) replaces it
+        f = Fake()
+        f.plan["layer"] = "host"
+        f.restart_out = "2101 lxc-start\n530 sshd\n"
+        rc, rep = run(f)
+        self.assertTrue(rep["reboot_needed"], rep)
+
+    def test_no_reboot_for_ordinary_daemons(self):
+        f = Fake()
+        f.plan["layer"] = "host"
+        f.restart_out = "530 sshd\n611 cron\n"
+        rc, rep = run(f)
+        self.assertFalse(rep["reboot_needed"], rep)
+
+
+class Speed(unittest.TestCase):
+    # R-845: no `pct exec` per package — version checks on the host, madison once, the restart scan only after an install.
+    def test_no_per_package_guest_calls(self):
+        f = Fake()
+        for i in range(40):
+            f.installed[f"pkg{i}"] = "1.0-1"
+            f.live[f"pkg{i}"] = {"1.0-2"}
+            f.plan["packages"].append({"name": f"pkg{i}", "version": "1.0-2", "origin": "Debian"})
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        guest_cmp = [c for c in f.calls if c[0] == "guest" and "--compare-versions" in c[2]]
+        self.assertEqual(guest_cmp, [], "version comparisons must run on the host")
+        self.assertEqual(f.madison_calls, 1, "madison must run once for all packages")
+        guest_calls = len([c for c in f.calls if c[0] == "guest"])
+        self.assertLess(guest_calls, 30, f"{guest_calls} guest calls for 42 packages — something is per-package again")
+
+
 class Failure(unittest.TestCase):
     def test_install_failure_is_rc3_with_dpkg_state(self):
         f = Fake()