forgetest: every run starts from, and leaves, the fresh-boot idle state

A baseline pass brackets every test and bench tool: before the run the
machine is verified against the fresh-boot idle state and anything off it
is restored; after the run - pass, fail, or abort - it is restored again.
Fixed items are the resting values the boot establishes (module defaults,
forgectrl's start-up writes, the GRBL controller's init writes) and
forgectrl's idle picture (controller running with motion verified, no
diagnostic, camera and cooling engines idle); preserved items (lid lamp
level, position counters, settings map, controller mode) are captured
before and handed back after. Deviations are leftovers: in the run pane,
in the result's evidence, and on the page - attributed to the previous run
when found before, to the run itself when found after.

forgetest takes a fresh-boot reference once per boot (within ten minutes
of boot, after the supervisor settles) as the session's resting lid-lamp
level and the check on the fixed values; the values were confirmed
against a fresh boot of the dev image on the bench (step_freq rests at
28160, the controller's default tick, not the probe's 10000).

Takeover runs capture the controller-owned kernel attributes on entry and
write them back before forgectrl restarts: the bench found the kernel
tests leaving motor_lock=15 behind, which masked the supervisor's
liveness probe - no motion by construction, a false driver-wedge verdict,
the rail-off ladder, and finally motion-fault. The takeover wrapper also
waits for the supervisor to settle on both sides (moved into baseline).

Catalog consequence: none beyond the runner; the tests' own drills are
unchanged.

docs/ACCEPTANCE.md

@@ -102,6 +102,38 @@ forces a full campaign; nothing before it can be inherited.
 The raw log (`/data/forgetest/results.jsonl`, `Raw log` in the footer) is
 the bench's own record; the artifact is the release's.
 
+### Every run starts from, and leaves, the fresh-boot idle state
+
+The runner brackets every test and bench tool with a **baseline** pass
+(`baseline.py`): before the run it verifies the machine against the
+fresh-boot idle state and restores anything off it; after the run - on
+every exit path, pass, fail, or abort - it restores again. Two kinds of
+items: **fixed** resting values the boot establishes (the kernel module
+defaults, forgectrl's start-up writes, the GRBL controller's init writes:
+`motor_lock=8`, `x/y_mode=8`, `x/y_decay=1`, `step_freq=28160`,
+`ramp_rate=125000`, `streaming=0`, `state=idle`, latch locked, hold
+currents, camera lamps and button LEDs off, heater and TEC off; forgectrl:
+the controller running with motion verified, no diagnostic, the camera
+engine and cooling engine idle), and **preserved** state with no resting
+policy that a run must hand back as it found it (the lid lamp level, the
+position counters, the settings map, the controller mode). Deviations are
+**leftovers**: logged in the run pane, kept in the result's `evidence`
+(`baseline.pre` / `baseline.post`), and surfaced in the page's message
+line - a leftover found before a run is attributed to the previous run; one
+found after is the run's own defect. Takeover runs additionally capture
+the controller-owned kernel attributes on entry and write them back before
+forgectrl restarts, so the supervisor's liveness probe runs on the machine
+it expects. The runner waits for forgectrl's supervisor to settle (motion
+verified, or the ladder's verdict) before and after every takeover.
+
+**Reboot before a campaign.** forgetest takes a **fresh-boot reference**
+once per boot (`/data/forgetest/boot-<boot_id>.json`, taken only within
+the first ten minutes after boot, after the supervisor settles): the whole
+idle picture of this machine as the image boots it. It is the session's
+resting lid-lamp level and the check on the fixed values; without one the
+lamp level is unknown and the page says so. Position counters cannot be
+written back - a run that shifts them is reported and must be fixed.
+
 ## The gate
 
 `scripts/release.sh <version>` builds the release image, reads
@@ -168,6 +200,7 @@ unported tools are listed with Start disabled. Bench runs are recorded in
       campaign.py                the rules (pure functions)
       artifact.py                export + gate verification
       runner.py                  one run at a time, prompts, abort, takeover
+      baseline.py                the fresh-boot idle state around every run
       server.py / page.py        HTTP API + the page (forgectrl's access rules)
       bench.py / coverage.py     bench registry + subprocess runner; the lint
       suite/                     the catalog, one module per subsystem

forgetest/forgetest/baseline.py

@@ -0,0 +1,483 @@
+"""The fresh-boot idle state: what every test starts from and leaves behind.
+
+The runner checks the machine against this baseline before a run and
+restores it after (on every exit path), so a test cannot hand the next one
+- or the operator - a machine that only looks idle. Two kinds of items:
+
+  fixed      a resting value the boot establishes and nothing at idle
+             changes: kernel module defaults, forgectrl's start-up writes,
+             the GRBL controller's init writes. Verified against the value,
+             restored by writing it back.
+  preserved  state with no resting policy that a run must hand back as it
+             found it: the lid lamp level, the position counters, the
+             settings map, the controller mode. Captured before the run,
+             compared after, restored where the interface allows.
+
+Everything found off-baseline is a "leftover": logged, kept in the run's
+evidence, and surfaced on the page. The pre-run pass attributes leftovers
+to the previous run; the post-run pass attributes them to the run itself.
+The machine is restored either way - a leftover is a defect in the test
+that made it, not a reason to hand the dirt on.
+"""
+import json
+import os
+import struct
+import time
+
+from . import hw
+from .log import now_ts
+
+# GRBL-mode resting values (kernel attribute -> value as read back), as a
+# fresh boot of the dev image leaves them (2026-08-16 bench dump).
+# motor_lock/x_mode/y_mode/x_decay/y_decay and the hold currents are the
+# GRBL controller's init writes (glowforge_io.c), step_freq its default
+# machine tick, ramp_rate the module default; streaming is only ever 1
+# inside a live job; the head white LED is a camera lamp, off at idle; the
+# loop heater and TEC are the diagnostics' tools, off at idle.
+FIXED_SYSFS = [
+    ("cnc/motor_lock", "8"),
+    ("cnc/x_mode", "8"),
+    ("cnc/y_mode", "8"),
+    ("cnc/x_decay", "1"),
+    ("cnc/y_decay", "1"),
+    ("cnc/step_freq", "28160"),         # GFSINK_RATE_DEFAULT (grblHAL stepper_stream.c)
+    ("cnc/ramp_rate", "125000"),
+    ("cnc/streaming", "0"),
+    ("pic/x_step_current", "33"),
+    ("pic/y_step_current", "5"),
+    ("head/white_led", "0"),
+    ("thermal/heater_pwm", "0"),
+    ("thermal/tec_on", "0"),
+]
+
+# Read-only readbacks with their idle values (no direct restore: the state
+# comes right through forgectrl - see restore_forgectrl - or is fatal).
+IDLE_READBACKS = [
+    ("cnc/state", "idle"),
+    ("cnc/laser_enable", "0"),
+    ("cnc/laser_on", "0"),
+    ("cnc/faults", "0"),
+]
+
+LATCH_BIT = 0x08                    # interlock_circuit bit 3: latch locked
+
+BUTTON_LEDS = ("button_led_1", "button_led_2", "button_led_3")
+
+PRESERVED_SYSFS = ["pic/lid_led"]   # captured before, restored after
+
+BOOT_MAX_AGE_S = 600                # a boot reference is taken only this soon after boot
+SETTLE_S = 150                      # the supervisor's probe + rail-off ladder
+CAM_IDLE_S = 20                     # camera engine idle stop is 10 s
+COOL_IDLE_S = 120                   # cooldown after motion
+IDLE_S = 30                         # cnc/state back to idle after a job
+
+
+def leds_root():
+    r = os.environ.get("GF_LEDS_ROOT") or "/sys/class/leds/"
+    return r if r.endswith("/") else r + "/"
+
+
+def read_led(name):
+    try:
+        with open(leds_root() + name + "/brightness") as f:
+            return f.read().strip()
+    except OSError:
+        return None
+
+
+def write_led(name, value):
+    with open(leds_root() + name + "/target", "w") as f:
+        f.write(str(value))
+
+
+def read_position():
+    """(x, y, z) step counters, or None when unreadable."""
+    try:
+        with open(hw.sysfs_root() + "cnc/position", "rb") as f:
+            raw = f.read(32)
+        return list(struct.unpack("<3i", raw[:12]))
+    except (OSError, struct.error):
+        return None
+
+
+class Leftover:
+    def __init__(self, item, found, expected, action):
+        self.item = item
+        self.found = found
+        self.expected = expected
+        self.action = action        # "restored" | "unrestorable" | "waited" | "failed: ..."
+
+    def as_dict(self):
+        return {"item": self.item, "found": self.found, "expected": self.expected,
+                "action": self.action}
+
+    def __str__(self):
+        return "%s=%s (expected %s) -> %s" % (self.item, self.found, self.expected, self.action)
+
+
+class Baseline:
+    """One instance per run: capture() before, check() around, restore()
+    after. `log` is a callable(str); every line is prefixed 'baseline:'."""
+
+    _unreachable_until = 0.0        # class-wide: skip forgectrl for a while after a miss
+
+    def __init__(self, log, abort=None):
+        self._log = log
+        self.abort = abort or (lambda: False)
+        self.captured = None
+        self.forgectrl = None
+
+    def log(self, msg):
+        self._log("baseline: " + msg)
+
+    # -- forgectrl access ------------------------------------------------
+    def fc(self):
+        if self.forgectrl is None:
+            self.forgectrl = hw.Forgectrl()
+        return self.forgectrl
+
+    def fc_get(self, path):
+        if time.time() < Baseline._unreachable_until:
+            return None, None
+        try:
+            st, body = self.fc().get(path)
+        except hw.HwError:
+            Baseline._unreachable_until = time.time() + 30
+            return None, None
+        if st is None:
+            Baseline._unreachable_until = time.time() + 30
+        return st, body
+
+    def fc_post(self, path, **kw):
+        try:
+            return self.fc().post(path, **kw)
+        except hw.HwError as e:
+            return None, str(e)
+
+    def wait_settled(self, timeout=SETTLE_S, unreachable_s=10):
+        """Block until forgectrl reports a settled supervisor: motion
+        verified (the probe passed), motion-fault (the ladder exhausted),
+        or standby (the manual stop lever). Gives up after unreachable_s
+        without an answer. Returns the last /mode body (None if unreachable)."""
+        t0 = time.time()
+        deadline = t0 + timeout
+        last = seen = heard = None
+        while time.time() < deadline and not self.abort():
+            try:
+                st, body = self.fc().get("/mode")
+            except hw.HwError:
+                st, body = None, None
+            if st is None and heard is None and time.time() - t0 >= unreachable_s:
+                self.log("forgectrl unreachable for %d s - not waiting for it" % unreachable_s)
+                Baseline._unreachable_until = time.time() + 30
+                return None
+            if st == 200 and isinstance(body, dict):
+                heard = time.time()
+                last = body
+                key = (body.get("controller"), body.get("motion"))
+                if key != seen:
+                    seen = key
+                    self.log("/mode controller=%s motion=%s" % key)
+                if (body.get("motion") == "verified"
+                        or body.get("controller") in ("motion-fault", "standby")):
+                    if body.get("controller") == "motion-fault":
+                        self.log("WARNING - motion liveness ladder failed, controllers are "
+                                 "down (motion-fault); retry via POST /mode")
+                    return body
+            time.sleep(1.0)
+        self.log("WARNING - forgectrl did not settle within %d s (last /mode: %s)" % (timeout, last))
+        return last
+
+    # -- capture -------------------------------------------------------
+    def capture(self):
+        """Record the preserved state before a run."""
+        cap = {"sysfs": {}, "position": read_position(), "settings": None, "mode": None}
+        for attr in PRESERVED_SYSFS:
+            cap["sysfs"][attr] = hw.sysfs_read(attr)
+        st, body = self.fc_get("/settings")
+        if st == 200 and isinstance(body, dict):
+            cap["settings"] = {k: v for k, v in body.items() if isinstance(v, str)}
+        st, body = self.fc_get("/mode")
+        if st == 200 and isinstance(body, dict):
+            cap["mode"] = body.get("mode")
+        self.captured = cap
+        return cap
+
+    # -- check + restore -----------------------------------------------
+    def enforce(self, phase, captured=None):
+        """Bring the machine to the baseline; returns the list of leftovers.
+        phase is 'pre' or 'post' (log wording only). captured is the
+        preserved state to hand back (post) - None compares nothing."""
+        left = []
+        self._forgectrl_side(left)
+        self._kernel_side(left)
+        self._preserved(left, captured)
+        if left:
+            self.log("%s: %d leftover(s): %s" % (phase, len(left), "; ".join(str(x) for x in left)))
+        else:
+            self.log("%s: clean" % phase)
+        return left
+
+    def _wait(self, what, pred, timeout):
+        t0 = time.time()
+        while time.time() - t0 < timeout and not self.abort():
+            if pred():
+                return time.time() - t0
+            time.sleep(1.0)
+        return None
+
+    def _forgectrl_side(self, left):
+        st, mode = self.fc_get("/mode")
+        if st != 200 or not isinstance(mode, dict):
+            self.log("forgectrl not answering - service-side checks skipped")
+            return
+        # a diagnostic left running seizes the thermal hardware: abort it
+        st, d = self.fc_get("/diag/status")
+        if st == 200 and isinstance(d, dict) and d.get("running"):
+            self.fc_post("/diag/abort")
+            w = self._wait("diag idle", lambda: not (self.fc_get("/diag/status")[1] or {}).get("running"), 60)
+            left.append(Leftover("diag", d.get("tool"), "not running",
+                                 "aborted" if w is not None else "failed: still running"))
+        # the camera engine stops itself 10 s after the last client
+        st, c = self.fc_get("/cam/status")
+        if st == 200 and isinstance(c, dict) and c.get("running"):
+            w = self._wait("cam idle", lambda: not (self.fc_get("/cam/status")[1] or {}).get("running"),
+                           CAM_IDLE_S)
+            if w is None:
+                left.append(Leftover("cam.running", True, False, "failed: still running"))
+        # supervisor: the captured mode, controller running, motion verified
+        want = (self.captured or {}).get("mode") or mode.get("mode") or "grbl"
+        if mode.get("mode") != want:
+            st, body = self.fc_post("/mode", data={"controller": want})
+            mode = self.wait_settled() or mode
+            left.append(Leftover("mode", mode.get("mode"), want,
+                                 "restored" if mode.get("mode") == want else "failed: %s %s" % (st, body)))
+        if mode.get("controller") == "motion-fault":
+            st, body = self.fc_post("/mode", data={"controller": want})
+            mode = self.wait_settled() or mode
+            left.append(Leftover("controller", "motion-fault", "running",
+                                 "restored" if mode.get("controller") == "running"
+                                 else "failed: %s" % mode.get("controller")))
+        elif mode.get("controller") == "standby":
+            st, body = self.fc_post("/controller/start")
+            mode = self.wait_settled() or mode
+            left.append(Leftover("controller", "standby", "running",
+                                 "restored" if mode.get("controller") == "running"
+                                 else "failed: %s" % mode.get("controller")))
+        elif mode.get("controller") != "running" or mode.get("motion") != "verified":
+            before = (mode.get("controller"), mode.get("motion"))
+            mode = self.wait_settled() or mode
+            if mode.get("controller") == "running" and mode.get("motion") == "verified":
+                left.append(Leftover("controller", "%s/%s" % before, "running/verified", "waited"))
+            else:
+                left.append(Leftover("controller", "%s/%s" % before, "running/verified",
+                                     "failed: %s/%s" % (mode.get("controller"), mode.get("motion"))))
+        # machine state through /status
+        st, s = self.fc_get("/status")
+        if st == 200 and isinstance(s, dict):
+            if s.get("state") != "idle":
+                if s.get("state") == "underrun":
+                    try:
+                        hw.sysfs_write("cnc/stop", "1")     # ack
+                    except OSError:
+                        pass
+                w = self._wait("idle", lambda: (self.fc_get("/status")[1] or {}).get("state") == "idle", IDLE_S)
+                left.append(Leftover("state", s.get("state"), "idle",
+                                     "waited" if w is not None else "failed: not idle"))
+            if s.get("laser_locked") is False:
+                try:
+                    hw.sysfs_write("cnc/laser_latch", "1")
+                    act = "restored"
+                except OSError as e:
+                    act = "failed: %s" % e
+                left.append(Leftover("laser_locked", False, True, act))
+        # cooling engine idle, unarmed
+        st, c = self.fc_get("/cool/status")
+        if st == 200 and isinstance(c, dict):
+            if c.get("phase") != "idle" or c.get("armed") or c.get("hold"):
+                found = "%s/armed=%s/hold=%s" % (c.get("phase"), c.get("armed"), c.get("hold"))
+                w = self._wait("cool idle", lambda: (lambda x: x.get("phase") == "idle" and not x.get("armed")
+                                                    and not x.get("hold"))(self.fc_get("/cool/status")[1] or {}),
+                               COOL_IDLE_S)
+                left.append(Leftover("cool", found, "idle/unarmed/no hold",
+                                     "waited" if w is not None else "failed: still %s" % found))
+
+    def _kernel_side(self, left):
+        if hw.sysfs_read("cnc/state") is None:
+            self.log("kernel sysfs not present - kernel-side checks skipped")
+            return
+        for attr, want in IDLE_READBACKS:
+            got = hw.sysfs_read(attr)
+            if got is not None and got != want:
+                left.append(Leftover(attr, got, want, "unrestorable"))
+        ilk = hw.sysfs_int("cnc/interlock_circuit")
+        if ilk is not None and not (ilk & LATCH_BIT):
+            try:
+                hw.sysfs_write("cnc/laser_latch", "1")
+                act = "restored"
+            except OSError as e:
+                act = "failed: %s" % e
+            left.append(Leftover("laser_latch", "unlocked (interlock 0x%x)" % ilk, "locked", act))
+        for attr, want in FIXED_SYSFS:
+            got = hw.sysfs_read(attr)
+            if got is None or got == want:
+                continue
+            try:
+                hw.sysfs_write(attr, want)
+                back = hw.sysfs_read(attr)
+                act = "restored" if back == want else "failed: reads %s" % back
+            except OSError as e:
+                act = "failed: %s" % e
+            left.append(Leftover(attr, got, want, act))
+        for name in BUTTON_LEDS:
+            got = read_led(name)
+            if got is not None and got != "0":
+                try:
+                    write_led(name, 0)
+                    act = "restored"
+                except OSError as e:
+                    act = "failed: %s" % e
+                left.append(Leftover("leds/" + name, got, "0", act))
+
+    def _preserved(self, left, captured):
+        if not captured:
+            return
+        for attr, was in (captured.get("sysfs") or {}).items():
+            now = hw.sysfs_read(attr)
+            if was is None or now is None or now == was:
+                continue
+            try:
+                hw.sysfs_write(attr, was)
+                back = hw.sysfs_read(attr)
+                act = "restored" if back == was else "failed: reads %s" % back
+            except OSError as e:
+                act = "failed: %s" % e
+            left.append(Leftover(attr, now, was, act))
+        was = captured.get("position")
+        now = read_position()
+        if was is not None and now is not None and now != was:
+            left.append(Leftover("position", now, was, "unrestorable"))
+        was = captured.get("settings")
+        if was:
+            st, body = self.fc_get("/settings")
+            if st == 200 and isinstance(body, dict):
+                for k, v in was.items():
+                    if body.get(k) == v:
+                        continue
+                    st2, b2 = self.fc_post("/settings", data={k: v})
+                    left.append(Leftover("settings." + k, body.get(k), v,
+                                         "restored" if st2 == 200 else "failed: %s %s" % (st2, b2)))
+
+
+# ------------------------------------------------------------ boot reference
+
+def boot_id():
+    v = os.environ.get("FORGETEST_BOOT_ID")
+    if v:
+        return v
+    try:
+        with open("/proc/sys/kernel/random/boot_id") as f:
+            return f.read().strip()
+    except OSError:
+        return None
+
+
+def uptime_s():
+    try:
+        with open("/proc/uptime") as f:
+            return float(f.read().split()[0])
+    except (OSError, ValueError, IndexError):
+        return None
+
+
+def dump_sysfs():
+    """Every readable text attribute under the module's sysfs root, by
+    'group/attr'; the binary position attribute decoded to (x, y, z)."""
+    out = {}
+    root = hw.sysfs_root()
+    for group in ("cnc", "pic", "head", "thermal"):
+        d = root + group
+        try:
+            names = sorted(os.listdir(d))
+        except OSError:
+            continue
+        for n in names:
+            path = "%s/%s" % (d, n)
+            if os.path.isdir(path) or n in ("uevent",):
+                continue
+            key = "%s/%s" % (group, n)
+            if key == "cnc/position":
+                out[key] = read_position()
+                continue
+            try:
+                with open(path, "rb") as f:
+                    raw = f.read(256)
+            except OSError:
+                continue
+            try:
+                out[key] = raw.decode("ascii").strip()
+            except UnicodeDecodeError:
+                out[key] = "<binary %d bytes>" % len(raw)
+    return out
+
+
+def dump_all(bl):
+    """The whole idle picture: sysfs, LEDs, forgectrl's status endpoints."""
+    d = {"sysfs": dump_sysfs(), "leds": {}, "forgectrl": {}}
+    for name in BUTTON_LEDS + ("lid_led",):
+        d["leds"][name] = read_led(name)
+    for path in ("/mode", "/status", "/cool/status", "/cam/status", "/diag/status", "/settings"):
+        st, body = bl.fc_get(path)
+        d["forgectrl"][path] = body if st == 200 else None
+    return d
+
+
+def check_fixed_against(ref, log):
+    """Compare the fixed constants with a fresh-boot dump; log the diffs
+    (a differing constant is a fact about this machine, not a leftover)."""
+    diffs = []
+    sysfs = ref.get("sysfs") or {}
+    for attr, want in FIXED_SYSFS + IDLE_READBACKS:
+        got = sysfs.get(attr)
+        if got is not None and got != want:
+            diffs.append("%s: boot=%s constant=%s" % (attr, got, want))
+    for d in diffs:
+        log("baseline: NOTE fresh boot differs from the fixed value - " + d)
+    return diffs
+
+
+def boot_reference(log, data_dir):
+    """The fresh-boot idle state of this boot: loaded from
+    <data_dir>/boot-<boot_id>.json when forgetest already took it, taken
+    now (after the supervisor settles) when the boot is recent, None
+    otherwise. Blocks for the settle - call from a background thread."""
+    bid = boot_id()
+    if not bid:
+        return None
+    path = os.path.join(data_dir, "boot-%s.json" % bid)
+    try:
+        with open(path) as f:
+            ref = json.load(f)
+        log("baseline: fresh-boot reference loaded (%s, taken %s)" % (path, ref.get("ts")))
+        return ref
+    except (OSError, ValueError):
+        pass
+    up = uptime_s()
+    if up is None or up > BOOT_MAX_AGE_S:
+        log("baseline: no fresh-boot reference for this boot (uptime %s s > %d s) - "
+            "the lid lamp resting level is unknown; reboot to take one" % (up, BOOT_MAX_AGE_S))
+        return None
+    bl = Baseline(log)
+    bl.wait_settled()
+    ref = dump_all(bl)
+    ref.update({"ts": now_ts(), "boot_id": bid, "uptime_s": uptime_s()})
+    try:
+        os.makedirs(data_dir, exist_ok=True)
+        with open(path, "w") as f:
+            json.dump(ref, f, indent=1, sort_keys=True)
+    except OSError as e:
+        log("baseline: could not save the fresh-boot reference: %s" % e)
+    log("baseline: fresh-boot reference taken at uptime %.0f s (%d sysfs attrs)"
+        % (ref["uptime_s"], len(ref["sysfs"])))
+    check_fixed_against(ref, log)
+    return ref

forgetest/forgetest/runner.py

@@ -23,6 +23,7 @@ import time
 import traceback
 
 from . import artifact as _artifact
+from . import baseline as _baseline
 from . import campaign as _campaign
 from . import catalog as _catalog
 from . import hw
@@ -197,61 +198,46 @@ class Takeover:
     is started again on every exit path. Used by takeover tests (through
     Context.takeover()) and by takeover bench tools."""
 
+    # Controller-owned kernel attributes a takeover drill may change:
+    # captured on enter, written back on exit before forgectrl starts, so
+    # the supervisor's liveness probe runs on the machine it expects (a
+    # leftover motor_lock=15 masks the probe's steps: no motion by
+    # construction, a false driver-wedge verdict, the rail-off ladder).
+    PRESERVE = ("cnc/motor_lock", "cnc/step_freq", "cnc/ramp_rate", "cnc/streaming",
+                "cnc/x_mode", "cnc/y_mode", "cnc/x_decay", "cnc/y_decay",
+                "pic/x_step_current", "pic/y_step_current")
+
     def __init__(self, log, who):
         self.log = log            # callable(str)
         self.who = who
         self.marker = marker_path()
+        self.saved = {}
 
-    # forgectrl's supervisor probes motion liveness on every start (a
-    # small head move, verified by the accelerometer) and runs a rail-off
-    # ladder of up to ~70 s on a dead verdict; /mode reads
-    # controller=stopped/motion=unverified until that settles.
-    SETTLE_S = 150
+    def wait_settled(self):
+        return _baseline.Baseline(self.log).wait_settled()
 
-    def wait_settled(self, timeout=SETTLE_S, unreachable_s=10):
-        """Block until forgectrl reports a settled supervisor: motion
-        verified (the probe passed), motion-fault (the ladder exhausted),
-        or standby (the manual stop lever). Gives up after unreachable_s without an answer (a
-        started forgectrl listens within a second or two). Returns the
-        last /mode body (or None if unreachable)."""
-        log = self.log
-        t0 = time.time()
-        deadline = t0 + timeout
-        last = None
-        seen = None
-        heard = None
-        while time.time() < deadline:
+    def restore_attrs(self):
+        """Write the captured kernel attributes back and relock the latch."""
+        for attr, val in self.saved.items():
             try:
-                st, body = hw.Forgectrl().get("/mode")
-            except hw.HwError:
-                st, body = None, None
-            if st is None and heard is None and time.time() - t0 >= unreachable_s:
-                log("takeover: forgectrl unreachable for %d s - not waiting for it" % unreachable_s)
-                return None
-            if st == 200 and isinstance(body, dict):
-                heard = time.time()
-                last = body
-                key = (body.get("controller"), body.get("motion"))
-                if key != seen:
-                    seen = key
-                    log("takeover: /mode controller=%s motion=%s" % key)
-                # settled: the probe passed (verified) or gave its verdict
-                # (motion-fault); standby is the manual lever, nothing in flight
-                if (body.get("motion") == "verified"
-                        or body.get("controller") in ("motion-fault", "standby")):
-                    if body.get("controller") == "motion-fault":
-                        log("takeover: WARNING - motion liveness ladder failed, controllers "
-                            "are down (motion-fault); retry via POST /mode")
-                    return body
-            time.sleep(1.0)
-        log("takeover: WARNING - forgectrl did not settle within %d s (last /mode: %s)"
-            % (timeout, last))
-        return last
+                hw.sysfs_write(attr, val)
+            except OSError as e:
+                self.log("takeover: WARNING could not restore %s=%s: %s" % (attr, val, e))
+        try:
+            hw.sysfs_write("cnc/laser_latch", "1")
+        except OSError as e:
+            self.log("takeover: WARNING could not relock the latch: %s" % e)
 
     def __enter__(self):
         log = self.log
         log("takeover: waiting for forgectrl to be settled")
         self.wait_settled()
+        for attr in self.PRESERVE:
+            v = hw.sysfs_read(attr)
+            if v is not None:
+                self.saved[attr] = v
+        if self.saved:
+            log("takeover: preserving %s" % ", ".join("%s=%s" % kv for kv in self.saved.items()))
         log("takeover: stopping the controller through forgectrl")
         try:
             st, body = hw.Forgectrl().post("/controller/stop")
@@ -273,6 +259,7 @@ class Takeover:
         return self
 
     def __exit__(self, exc_type, exc, tb):
+        self.restore_attrs()
         rc, out = hw.initd("forgectrl", "start")
         self.log("takeover: forgectrl start -> rc %s" % rc)
         try:
@@ -300,7 +287,22 @@ class Runner:
         self.current = None
         self.last = None
         self.messages = []
+        self.boot_ref = None
         self.recover()
+        threading.Thread(target=self._take_boot_reference, daemon=True,
+                         name="forgetest-bootref").start()
+
+    def _take_boot_reference(self):
+        try:
+            self.boot_ref = _baseline.boot_reference(self._note, data_dir())
+        except Exception as e:  # noqa: BLE001
+            self._note("baseline: boot reference failed: %s: %s" % (type(e).__name__, e))
+
+    def _note(self, msg):
+        """A runner-level line: kept in messages for the page (bounded)."""
+        with self._lock:
+            self.messages.append(msg)
+            del self.messages[:-50]
 
     # -- startup recovery ------------------------------------------------
     def recover(self):
@@ -402,11 +404,37 @@ class Runner:
         th.start()
         return True, "started"
 
+    # -- baseline around every run -----------------------------------------
+    def _baseline_pre(self, run):
+        """Bring the machine to the fresh-boot idle state before a run and
+        record what the previous run left behind. Returns the captured
+        preserved state for the post pass."""
+        bl = _baseline.Baseline(run.log, abort=run.aborted.is_set)
+        ref = self.boot_ref
+        session = {"sysfs": {a: (ref.get("sysfs") or {}).get(a) for a in _baseline.PRESERVED_SYSFS}} if ref else None
+        left = bl.enforce("pre", captured=session)
+        if left:
+            who = self.last.id if self.last is not None else "an earlier run"
+            self.messages.append("leftovers before %s (left by %s): %s"
+                                 % (run.id, who, "; ".join(str(x) for x in left)))
+        run.evidence["baseline"] = {"pre": [x.as_dict() for x in left]}
+        return bl.capture()
+
+    def _baseline_post(self, run, captured):
+        bl = _baseline.Baseline(run.log)
+        left = bl.enforce("post", captured=captured)
+        run.evidence.setdefault("baseline", {})["post"] = [x.as_dict() for x in left]
+        if left:
+            self.messages.append("leftovers after %s: %s" % (run.id, "; ".join(str(x) for x in left)))
+        return left
+
     def _exec_test(self, t, run, campaign):
         ctx = Context(run, self, t)
         fp = t.fingerprint(self.manifest)
         result, message = _campaign.PASS, ""
+        captured = None
         try:
+            captured = self._baseline_pre(run)
             t.fn(ctx)
             if run.aborted.is_set():
                 result, message = _campaign.ABORTED, "aborted"
@@ -417,6 +445,10 @@ class Runner:
         except Exception as e:  # noqa: BLE001 - an erroring test is a failed test
             result, message = _campaign.ERROR, "%s: %s" % (type(e).__name__, e)
             run.log(traceback.format_exc().rstrip())
+        try:
+            self._baseline_post(run, captured)
+        except Exception as e:  # noqa: BLE001 - never lose the result over the cleanup
+            run.log("baseline: post pass errored: %s: %s" % (type(e).__name__, e))
         duration = int(time.time() - run.started)
         run.log("result %s%s" % (result, (": " + message) if message else ""))
         rec = {"t": "result", "campaign": campaign["id"], "test": t.id, "result": result,
@@ -455,7 +487,9 @@ class Runner:
     def _exec_bench(self, tool, run, argv, args):
         rc = None
         message = ""
+        captured = None
         try:
+            captured = self._baseline_pre(run)
             env = dict(os.environ)
             env.setdefault("PYTHONUNBUFFERED", "1")
             takeover = Takeover(run.log, "bench:" + tool["id"]) if tool.get("safety") == "takeover" else None
@@ -476,6 +510,10 @@ class Runner:
         except Exception as e:  # noqa: BLE001
             message = "%s: %s" % (type(e).__name__, e)
             run.log(message)
+        try:
+            self._baseline_post(run, captured)
+        except Exception as e:  # noqa: BLE001
+            run.log("baseline: post pass errored: %s: %s" % (type(e).__name__, e))
         duration = int(time.time() - run.started)
         result = "ABORTED" if run.aborted.is_set() else ("OK" if rc == 0 else "EXIT %s" % rc)
         run.log("bench %s finished: %s" % (tool["id"], result))

forgetest/tests/test_baseline.py

@@ -0,0 +1,169 @@
+"""The baseline: fixed resting values are restored, preserved values are
+handed back, every deviation is a recorded leftover. Runs against a fake
+sysfs tree; forgectrl is unreachable (service-side checks skip)."""
+import os
+import shutil
+import struct
+import tempfile
+import unittest
+
+from forgetest import baseline
+
+
+class BaselineTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="forgetest-bl-")
+        self.sysfs = os.path.join(self.tmp, "sysfs") + os.sep
+        self.leds = os.path.join(self.tmp, "leds") + os.sep
+        for group in ("cnc", "pic", "head", "thermal"):
+            os.makedirs(self.sysfs + group)
+        for name in baseline.BUTTON_LEDS + ("lid_led",):
+            os.makedirs(self.leds + name)
+            self._led(name, "0")
+        # a clean machine
+        for attr, val in baseline.FIXED_SYSFS + baseline.IDLE_READBACKS:
+            self._attr(attr, val)
+        self._attr("cnc/interlock_circuit", "45")
+        self._attr("pic/lid_led", "0")
+        self._pos(0, 0, 0)
+        os.environ["GF_SYSFS_ROOT"] = self.sysfs
+        os.environ["GF_LEDS_ROOT"] = self.leds
+        os.environ["FORGECTRL_URL"] = "http://127.0.0.1:1"      # nothing listens
+        baseline.Baseline._unreachable_until = 0.0
+        self.lines = []
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+        for k in ("GF_SYSFS_ROOT", "GF_LEDS_ROOT", "FORGECTRL_URL"):
+            os.environ.pop(k, None)
+
+    def _attr(self, attr, val):
+        with open(self.sysfs + attr, "w") as f:
+            f.write(str(val))
+
+    def _read(self, attr):
+        with open(self.sysfs + attr) as f:
+            return f.read().strip()
+
+    def _led(self, name, val):
+        with open(self.leds + name + "/brightness", "w") as f:
+            f.write(val)
+        # the class interface writes 'target'; the fake mirrors it into brightness
+        # only when the test asks (see _sync_leds)
+
+    def _sync_leds(self):
+        for name in baseline.BUTTON_LEDS:
+            p = self.leds + name + "/target"
+            if os.path.exists(p):
+                with open(p) as f:
+                    v = f.read().strip()
+                with open(self.leds + name + "/brightness", "w") as f:
+                    f.write(v)
+
+    def _pos(self, x, y, z):
+        with open(self.sysfs + "cnc/position", "wb") as f:
+            f.write(struct.pack("<5i", x, y, z, 0, 0))
+
+    def bl(self):
+        return baseline.Baseline(self.lines.append)
+
+    def test_clean_machine_has_no_leftovers(self):
+        left = self.bl().enforce("pre", captured=None)
+        self.assertEqual(left, [])
+        self.assertTrue(any("pre: clean" in l for l in self.lines))
+
+    def test_fixed_values_are_restored_and_recorded(self):
+        self._attr("cnc/motor_lock", "15")
+        self._attr("cnc/step_freq", "10000")
+        self._attr("cnc/streaming", "1")
+        left = self.bl().enforce("post", captured=None)
+        items = {x.item: x for x in left}
+        self.assertEqual(set(items), {"cnc/motor_lock", "cnc/step_freq", "cnc/streaming"})
+        for x in left:
+            self.assertEqual(x.action, "restored", str(x))
+        self.assertEqual(self._read("cnc/motor_lock"), "8")
+        self.assertEqual(self._read("cnc/step_freq"), "28160")
+        self.assertEqual(self._read("cnc/streaming"), "0")
+        self.assertEqual(items["cnc/motor_lock"].found, "15")
+        self.assertEqual(items["cnc/motor_lock"].expected, "8")
+
+    def test_unlocked_latch_is_relocked(self):
+        self._attr("cnc/interlock_circuit", "5")        # bit 3 clear = unlocked
+        left = self.bl().enforce("post", captured=None)
+        self.assertEqual([x.item for x in left], ["laser_latch"])
+        self.assertEqual(self._read("cnc/laser_latch"), "1")
+
+    def test_readonly_deviation_is_unrestorable(self):
+        self._attr("cnc/state", "disabled")
+        left = self.bl().enforce("post", captured=None)
+        self.assertEqual([(x.item, x.action) for x in left], [("cnc/state", "unrestorable")])
+
+    def test_button_leds_are_turned_off(self):
+        self._led("button_led_2", "255")
+        left = self.bl().enforce("post", captured=None)
+        self.assertEqual([x.item for x in left], ["leds/button_led_2"])
+        self._sync_leds()
+        self.assertEqual(baseline.read_led("button_led_2"), "0")
+
+    def test_preserved_lamp_and_position(self):
+        b = self.bl()
+        self._attr("pic/lid_led", "132")
+        cap = b.capture()
+        self.assertEqual(cap["sysfs"]["pic/lid_led"], "132")
+        self.assertEqual(cap["position"], [0, 0, 0])
+        # the run turned the lamp off and shifted the counters
+        self._attr("pic/lid_led", "0")
+        self._pos(1000, 0, 0)
+        left = b.enforce("post", captured=cap)
+        items = {x.item: x for x in left}
+        self.assertEqual(set(items), {"pic/lid_led", "position"})
+        self.assertEqual(items["pic/lid_led"].action, "restored")
+        self.assertEqual(self._read("pic/lid_led"), "132")
+        self.assertEqual(items["position"].action, "unrestorable")
+        self.assertEqual(items["position"].found, [1000, 0, 0])
+
+    def test_session_resting_lamp_from_boot_reference(self):
+        # the pre pass hands the lamp back to the boot level
+        self._attr("pic/lid_led", "77")
+        session = {"sysfs": {"pic/lid_led": "0"}}
+        left = self.bl().enforce("pre", captured=session)
+        self.assertEqual([(x.item, x.action) for x in left], [("pic/lid_led", "restored")])
+        self.assertEqual(self._read("pic/lid_led"), "0")
+
+    def test_no_sysfs_means_skip(self):
+        os.environ["GF_SYSFS_ROOT"] = os.path.join(self.tmp, "nope") + os.sep
+        left = self.bl().enforce("pre", captured=None)
+        self.assertEqual(left, [])
+        self.assertTrue(any("kernel sysfs not present" in l for l in self.lines))
+
+    def test_boot_reference_needs_a_recent_boot(self):
+        os.environ["FORGETEST_BOOT_ID"] = "test-boot"
+        try:
+            # no reference file, uptime unknown on a host without /proc/uptime,
+            # or too old: None, with the reason logged
+            ref = baseline.boot_reference(self.lines.append, self.tmp)
+            up = baseline.uptime_s()
+            if up is None or up > baseline.BOOT_MAX_AGE_S:
+                self.assertIsNone(ref)
+                self.assertTrue(any("no fresh-boot reference" in l for l in self.lines))
+            else:
+                # a young host: the reference is taken from the fake tree
+                self.assertIsNotNone(ref)
+                self.assertEqual(ref["sysfs"]["cnc/motor_lock"], "8")
+                self.assertTrue(os.path.exists(os.path.join(self.tmp, "boot-test-boot.json")))
+                # and loaded back the second time
+                self.lines[:] = []
+                ref2 = baseline.boot_reference(self.lines.append, self.tmp)
+                self.assertEqual(ref2["ts"], ref["ts"])
+                self.assertTrue(any("reference loaded" in l for l in self.lines))
+        finally:
+            os.environ.pop("FORGETEST_BOOT_ID", None)
+
+    def test_fixed_constants_checked_against_a_dump(self):
+        ref = {"sysfs": {"cnc/motor_lock": "8", "cnc/step_freq": "10000"}}
+        diffs = baseline.check_fixed_against(ref, self.lines.append)
+        self.assertEqual(diffs, ["cnc/step_freq: boot=10000 constant=28160"])
+
+
+if __name__ == "__main__":
+    unittest.main()

forgetest/tests/test_server.py

@@ -168,7 +168,9 @@ class ServerTests(unittest.TestCase):
         self.assertEqual(state["last_run"]["finished"]["result"], "PASS")
         st, rec = self.call("GET", "/result?test=fake.pass")
         self.assertEqual(st, 200)
-        self.assertEqual(rec["evidence"], {"k": 1})
+        self.assertEqual(rec["evidence"]["k"], 1)
+        # the baseline passes ran (no machine on the host: nothing to restore)
+        self.assertEqual(rec["evidence"]["baseline"], {"pre": [], "post": []})
         self.assertTrue(any("hello" in l for l in rec["log"]))
 
     def test_03_requires_and_live_gate(self):
@@ -291,7 +293,7 @@ class ServerTests(unittest.TestCase):
         log = "\n".join(state["last_run"]["log"])
         self.assertIn("takeover: pulse device free", log)
         self.assertIn("takeover: forgectrl start", log)
-        self.assertIn("takeover: forgectrl unreachable for 10 s", log)
+        self.assertIn("baseline: forgectrl unreachable for 10 s", log)
         self.assertFalse(os.path.exists(os.environ["FORGETEST_MARKER"]))
         # bench runs never touched the acceptance log
         recs = self.log.read()