forgetest: the lid lamp idles at forgectrl's lid_lamp_idle; the liveness test masks and restarts

The lid lamp now has a resting policy in forgectrl (lid_lamp_idle,
default 236, asserted at start and at every spawn), so the baseline
expects it there instead of preserving whatever level a boot left, and
forgectrl.settings-bounds proves it: resting at the setting, 256 / -1 /
'bright' refused, a new level applied to the lamp at once, the cleared
key back to the default. Clearing a key goes through the query-string
form (an empty JSON value reads as no setting).

motion.liveness-probe adds the regression the bench needed: with every
axis masked (cnc/motor_lock=15, as a bench tool may leave it) forgectrl
is restarted and its fresh probe must read MOTION OK on the first try -
the probe unmasks the axes itself - and the controller comes up with the
mask cleared. Bench 2026-08-16: MOTION OK at p2p 2047/1341 against the
800 threshold, no ladder.

The baseline's settle no longer counts 'probe verified, spawn pending' as
settled (the post pass ran between the probe's own writes and the
controller's init writes and mis-flagged motor_lock/step_freq): it waits
for the controller to be running, with a bounded allowance for a respawn
backoff.

forgetest/forgetest/suite/forgectrl.py

@@ -1,6 +1,7 @@
 """forgectrl.* - the machine-services daemon's API, access control, and panel."""
 import json
 import socket
+import time
 
 from ..catalog import test
 from .. import hw
@@ -107,9 +108,11 @@ def auth(ctx):
 
 @test("forgectrl.settings-bounds", title="Settings validation and restore", subsystem="forgectrl",
       kind="auto", est_min=1,
-      covers=[("forgectrl", "src/settings.*"), ("forgectrl", "src/main.c")],
+      covers=[("forgectrl", "src/settings.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/cam.c")],
       description="An over-length value and an out-of-range value are refused (400) and leave the "
-                  "settings byte-identical; an in-range value is accepted (200).")
+                  "settings byte-identical; an in-range value is accepted (200). The lid lamp "
+                  "idles at lid_lamp_idle (unset = 236), an out-of-range level is refused, a new "
+                  "level applies to the lamp at once, and clearing it returns the default.")
 def settings_bounds(ctx):
     fc = ctx.forgectrl
     ev = ctx.evidence
@@ -158,6 +161,44 @@ def settings_bounds(ctx):
     ctx.check(others_before == others_after, "other settings changed by the write")
     ctx.check(final.get(key) == val, "%s reads back %r, wrote %r", key, final.get(key), val)
 
+    # the lid lamp's idle level: resting at the setting, bounded, applied live
+    lamp_was = (before.get("lid_lamp_idle") or "").strip()
+    want = lamp_was or "236"
+    got = ctx.sysfs("pic/lid_led")
+    ev["lid_lamp"] = {"setting": lamp_was, "resting": got}
+    ctx.log("lid lamp: setting %r, pic/lid_led=%s (expected %s)", lamp_was, got, want)
+    ctx.check(got == want, "lid lamp rests at %s, lid_lamp_idle is %s", got, want)
+    for bad in ("256", "-1", "bright"):
+        st, body = fc.post("/settings", data={"lid_lamp_idle": bad})
+        ctx.check(st == 400, "lid_lamp_idle=%s -> %s, expected 400", bad, st)
+    ctx.log("lid_lamp_idle 256 / -1 / bright refused")
+    try_level = "100" if want != "100" else "120"
+    st, body = fc.post("/settings", data={"lid_lamp_idle": try_level})
+    ctx.check(st == 200, "lid_lamp_idle=%s -> %s, expected 200", try_level, st)
+    applied = None
+    t0 = time.time()
+    while time.time() - t0 < 5:
+        applied = ctx.sysfs("pic/lid_led")
+        if applied == try_level:
+            break
+        ctx.sleep(0.2)
+    ctx.log("lid_lamp_idle=%s -> pic/lid_led=%s after %.1f s", try_level, applied, time.time() - t0)
+    # an empty value clears the key: the query-string form carries it
+    st, body = (fc.post("/settings", params={"lid_lamp_idle": ""}) if not lamp_was
+                else fc.post("/settings", data={"lid_lamp_idle": lamp_was}))
+    ctx.check(st == 200, "restoring lid_lamp_idle=%r -> %s", lamp_was, st)
+    t0 = time.time()
+    back = None
+    while time.time() - t0 < 5:
+        back = ctx.sysfs("pic/lid_led")
+        if back == want:
+            break
+        ctx.sleep(0.2)
+    ev["lid_lamp"].update({"applied": applied, "restored": back})
+    ctx.check(applied == try_level, "lamp did not follow lid_lamp_idle=%s (reads %s)", try_level, applied)
+    ctx.check(back == want, "lamp did not return to %s after the restore (reads %s)", want, back)
+    ctx.log("lid lamp follows the setting live and returns to %s", want)
+
 
 @test("forgectrl.panel-serves", title="Control panel and status endpoints", subsystem="forgectrl",
       kind="auto", est_min=1,

forgetest/forgetest/suite/motion.py

@@ -5,6 +5,7 @@ Ported from `scripts/bench/pacing_test.py` (protocol-loop pacing) and
 and round-trip; the laser stays latched (the tests never touch it); the
 suite is the only Grbl client while a test runs.
 """
+import os
 import time
 
 from ..catalog import test
@@ -230,13 +231,23 @@ def jog_roundtrip(ctx):
       kind="auto", est_min=1,
       covers=[("forgectrl", "src/super.c"), ("forgectrl", "src/liveness.c"), ("kernel-module-glowforge", "**")],
       requires=["kernel.latch-locked-idle"],
-      steps=["Bed clear, lid closed: the probe jogs the head a few mm (+X first)."],
+      steps=["Bed clear, lid closed: the probe jogs the head 15 mm out and back (+X first); "
+             "forgectrl is restarted once for a fresh probe."],
       description="forgectrl's supervisor reports the head-accelerometer liveness probe as "
                   "verified for the running controller (the DRV8825s are not wedged); when the "
-                  "probe was skipped at spawn, the controller is respawned once so it runs.")
+                  "probe was skipped at spawn, the controller is respawned once so it runs. Then "
+                  "the regression: with every axis masked (cnc/motor_lock=15, as a bench tool "
+                  "may leave it) forgectrl is restarted and its fresh probe must still read "
+                  "MOTION OK - the probe unmasks the axes itself - with the head-accel p2p at "
+                  "or above the moving threshold.")
 def liveness_probe(ctx):
     fc = ctx.forgectrl
     ev = ctx.evidence
+    _liveness_verdict(ctx, fc, ev)
+    _liveness_masked_restart(ctx, fc, ev)
+
+
+def _liveness_verdict(ctx, fc, ev):
     st, m = fc.get("/mode")
     ctx.check(st == 200 and isinstance(m, dict), "GET /mode -> %s", st)
     ev["mode_before"] = m
@@ -263,6 +274,64 @@ def liveness_probe(ctx):
     ctx.check(m.get("motion") == "verified", "liveness is %r, expected verified", m.get("motion"))
 
 
+FORGECTRL_LOG = "/data/log/forgefirm/forgectrl/forgectrl.log"
+
+
+def _log_offset(path):
+    try:
+        return os.path.getsize(path)
+    except OSError:
+        return 0
+
+
+def _probe_lines(path, offset):
+    try:
+        with open(path, "rb") as f:
+            f.seek(offset)
+            data = f.read().decode("utf-8", "replace")
+    except OSError:
+        return []
+    return [ln.strip() for ln in data.splitlines() if "liveness probe:" in ln]
+
+
+def _liveness_masked_restart(ctx, fc, ev):
+    """The regression: a leftover motor_lock must not read as a wedge."""
+    ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle before the masked restart")
+    x0 = _kernel_x_mm(ctx)
+    hw.sysfs_write("cnc/motor_lock", "15")
+    ctx.log("masked every axis (cnc/motor_lock=15); restarting forgectrl for a fresh probe")
+    off = _log_offset(FORGECTRL_LOG)
+    rc, out = hw.initd("forgectrl", "restart")
+    ctx.check(rc == 0, "forgectrl restart -> rc %s", rc)
+    m = None
+    t0 = time.time()
+    while time.time() - t0 < 150:
+        ctx.checkpoint()
+        try:
+            st, m = fc.get("/mode")
+        except hw.HwError:
+            m = None                # the daemon is still coming up
+        if isinstance(m, dict) and ((m.get("controller") == "running" and m.get("motion") == "verified")
+                                    or m.get("controller") == "motion-fault"):
+            break
+        ctx.sleep(1)
+    lines = _probe_lines(FORGECTRL_LOG, off)
+    for ln in lines:
+        ctx.log("  %s", ln.split(" INFO ", 1)[-1] if " INFO " in ln else ln[-160:])
+    ev["masked_restart"] = {"mode": m, "probe_lines": lines[-4:], "motor_lock_after": ctx.sysfs("cnc/motor_lock")}
+    ctx.check(m and m.get("controller") == "running" and m.get("motion") == "verified",
+              "fresh probe under a leftover mask did not verify motion: %s", m)
+    ctx.check(lines and "MOTION OK" in lines[0],
+              "the first probe after the restart was not MOTION OK: %s", lines[:1])
+    ctx.check(len(lines) == 1, "the probe needed the recovery ladder (%d probes) - a false dead verdict", len(lines))
+    ctx.check(ctx.sysfs("cnc/motor_lock") == "8", "motor_lock reads %s after the controller start (expected 8)",
+              ctx.sysfs("cnc/motor_lock"))
+    ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle after the probe")
+    x1 = _kernel_x_mm(ctx)
+    ctx.log("kernel X %s -> %s mm across the probe (out and back)", x0, x1)
+    ctx.log("PASS: masked restart probed MOTION OK on the first try, mask cleared, controller up")
+
+
 # ---------------------------------------------------------------- cancel / abort
 
 @test("motion.cancel-abort", title="Jog cancel and controlled abort recover cleanly", subsystem="motion",