Laser harness rules and catalog tests for the emission gates

The GRBL driver's emission gates change (grblHAL-glowforge: the per-tick
fire gate, the latch with an owner, the cooling verdict's two tiers, and
the arm-flow gates). This commit carries the host rules and the catalog
tests that hold them; the driver commit follows, because its CI fetches
these harnesses unpinned.

scripts/bench/laser_stream_test.py:
- Rule 24 is the verdict's pause tier: the client holds the job under the
  open window, the first deceleration runs lit to the stop (the dark lead
  before the stop is at most 1000 ticks: the producer's lead plus one
  shipper period), a resume under the standing verdict moves dark and is
  held again, the clean verdict resumes lit with no press, and the latch
  sideband carries only the arm's unlock and the program end's lock.
- Rule 25 is the fail tier: AIRFLOW mid-cut ends the job with ALARM:3,
  the stream ends dark well short of the line, the sideband ends on the
  lock, and a resume under the clean verdict that follows resumes nothing.
- Rule 26: a sender change mid-M3 holds the job with the deceleration
  dark: the gate follows the window on every tick.
- Rule 27: a verdict that goes stale holds the job at the cache's own
  expiry, lit to the stop, never a poll later; the engine's return
  resumes lit.
- The stand-in engine publishes the verdict name and has a stale mode;
  the session steps gain expect_text, reconnect and a wait_state timeout;
  every session reads the latch sideband (GFSINK_LATCH_LOG).

scripts/bench/laser_lifecycle_test.py:
- Rules 11 to 15: the pause tier resumes with no press and no prompt, the
  fail tier ends the job and nothing resumes it, a sender change during a
  re-arm cancels it, a jog does not hold the window open, and a press
  counts only after the button has been seen up. The stand-in engine
  takes a live verdict dict. start_armed_move waits for a fresh prompt
  and a fresh armed message: a press that lands before the wait has
  begun is not consent, and the old stale match let one land early.

forgetest/forgetest/suite/laser.py:
- laser.verdict-cut (kind live, one press): a 40 mm M3 line; the test
  pauses the daemon for 3.5 s so the verdict goes stale (the settings
  route is idle-gated and the engine reloads its gates at a session
  start, so no setting can trip a pause mid-cut; the crash tiers need a
  physical knock). The controller must hold with the SoC latch and the
  hardware button latch both clear in every sample, emission must read 0
  before the resume, and the clean verdict must resume the cut lit with
  no press and no prompt; M2 disarms as usual.
- laser.armed-kill asserts that the respawned controller comes up with
  the latch still locked.

Proof: both harnesses pass against the driver change on the host, and
the forgetest unit tests pass. On the bench reference, laser.verdict-cut
passed with no gap in the cut (held at +1.71 s after the pause began,
both latches clear in every sample, emission 0 at +2.06 s after the
hold, resumed lit at +4.0 s with the beam detector 678 counts over idle,
kernel drift 0.0 mm, disarmed 0.1 s after Idle) and laser.armed-kill
passed (emission 0 at +1.8 s after the supervisor's stop and +2.1 s
after the SIGKILL, latch locked, respawned with the latch locked, button
dark). Two earlier verdict-cut runs shaped the driver: a pause that
locked the latch resumed dark, because a lock sets the hardware button
latch, and a hold taken a poll after the gate closed left a several-mm
gap.

Catalog: one test added and one extended; both cover src/** of the
driver through the existing laser covers.

scripts/bench/laser_lifecycle_test.py

@@ -39,6 +39,20 @@ reported messages:
      not to where it was paused (the core restarts a held cycle through
      Idle); a job abandoned in a hold and reset ends there, so the next
      job's start is captured afresh where it begins
+ 11. the cooling verdict's pause tier (OVERTEMP) holds the job under the
+     open window without writing the latch (a lock sets the hardware
+     button latch, which only a press clears); the clean verdict then
+     resumes it with no new button press: no prompt, no second arm
+ 12. the verdict's fail tier (AIRFLOW) ends the job: disarmed, reset with
+     ALARM:3, and a ~ under the clean verdict that follows resumes nothing
+ 13. a sender change during a re-arm cancels it: the press that follows
+     resumes nothing, the job stays held, and the new sender's own ~
+     prompts afresh
+ 14. a jog does not hold the armed window open: with the spindle off the
+     grace counts down through jogs, and the window closes on time
+ 15. a press counts only after the button has been seen up: a press that
+     began before the wait is not consent (the harness's own presses
+     follow a fresh prompt for the same reason)
 
 The disarm grace is shortened via a temp config (GFHOME_CONF), the
 cooling verdict is published hermetically (GF_VERDICT_FILE), the same
@@ -141,23 +155,29 @@ def wait_idle(sock, log):
     fail("controller never returned to Idle")
 
 
-def publish_verdicts(path, stop, fire_ok, armed_ack=True, ack_after_s=0.0):
-    """Stand in for the cooling engine. "armed" is the engine's
-    acknowledgment that it has taken the controller's armed window and
-    applied the run airflow; the controller refuses to fire on a verdict
-    that lacks it, so an engine that never acknowledges (armed_ack
-    False) must produce a refused arm and no emission. ack_after_s
-    withholds the acknowledgment for that long first, which is the real
-    engine's case: it answers on its next tick, and the controller has
-    to see the refreshed verdict to get past the arm."""
+def publish_verdicts(path, stop, verdict):
+    """Stand in for the cooling engine. `verdict` is the session's live
+    dict: fire_ok, hold, resume_ok, name, armed_ack, ack_after_s. "armed"
+    is the engine's acknowledgment that it has taken the controller's
+    armed window and applied the run airflow; the controller refuses to
+    fire on a verdict that lacks it, so an engine that never
+    acknowledges (armed_ack False) must produce a refused arm and no
+    emission. ack_after_s withholds the acknowledgment for that long
+    first, which is the real engine's case: it answers on its next
+    tick, and the controller has to see the refreshed verdict to get
+    past the arm."""
     t0 = time.monotonic()
     while not stop.is_set():
-        acked = armed_ack and time.monotonic() - t0 >= ack_after_s
-        body = ('{"ts_mono":%.3f,"fire_ok":%s,"hold":false,'
-                '"resume_ok":true,"armed":%s,"reason":""}'
+        v = dict(verdict)
+        acked = v["armed_ack"] and time.monotonic() - t0 >= v["ack_after_s"]
+        body = ('{"ts_mono":%.3f,"fire_ok":%s,"verdict":"%s","hold":%s,'
+                '"resume_ok":%s,"armed":%s,"reason":"%s"}'
                 % (time.clock_gettime(time.CLOCK_MONOTONIC),
-                   "true" if fire_ok else "false",
-                   "true" if acked else "false"))
+                   "true" if v["fire_ok"] else "false", v["name"],
+                   "true" if v["hold"] else "false",
+                   "true" if v["resume_ok"] else "false",
+                   "true" if acked else "false",
+                   "" if v["fire_ok"] else "harness: %s" % v["name"]))
         tmp = path + ".tmp"
         with open(tmp, "w") as f:
             f.write(body)
@@ -165,6 +185,20 @@ def publish_verdicts(path, stop, fire_ok, armed_ack=True, ack_after_s=0.0):
         stop.wait(0.5)
 
 
+def wait_for_new(log, needle, n_before, timeout, sock=None):
+    """Wait until needle appears in the log more times than n_before: a
+    fresh occurrence, never a stale match from an earlier job."""
+    end = time.time() + timeout
+    while time.time() < end:
+        if "".join(log).count(needle) > n_before:
+            return True
+        if sock is not None:
+            read_avail(sock, log, 0.2)
+        else:
+            time.sleep(0.1)
+    return False
+
+
 class Session:
     """One controller process with the lifecycle overrides applied."""
 
@@ -189,8 +223,10 @@ class Session:
             self.set_switches(switches)
             env["GF_SWITCH_FILE"] = self.switch_file
         self.stop = threading.Event()
+        self.verdict = {"fire_ok": fire_ok, "hold": False, "resume_ok": True, "name": "OK",
+                        "armed_ack": armed_ack, "ack_after_s": ack_after_s}
         self.pub = threading.Thread(target=publish_verdicts,
-                                    args=(verdict, self.stop, fire_ok, armed_ack, ack_after_s),
+                                    args=(verdict, self.stop, self.verdict),
                                     daemon=True)
         self.pub.start()
         self.proc = subprocess.Popen([BIN, "-p", str(PORT)],
@@ -220,6 +256,16 @@ class Session:
             f.write("%d\n" % word)
         os.replace(tmp, self.switch_file)
 
+    def set_verdict(self, name="OK", fire_ok=True, hold=False, resume_ok=True):
+        """The engine's next verdicts: a pause tier (OVERTEMP: fire
+        blocked, hold, no resume), a fail tier (AIRFLOW: the same under a
+        name the client ends the job on), or clean."""
+        self.verdict.update({"name": name, "fire_ok": fire_ok, "hold": hold,
+                             "resume_ok": resume_ok})
+
+    def count(self, needle):
+        return "".join(self.log).count(needle)
+
     def send_raw(self, line):
         """Send a line without waiting for ok/error (the arm wait blocks
         the gcode stream, so the ok only comes once the button is pressed)."""
@@ -867,13 +913,17 @@ def test_lid_open_in_wait():
 
 def start_armed_move(s, tag, gcode="G1 X30 F60"):
     """Arm through the button and get a long move under way; returns once
-    the controller reports Run."""
+    the controller reports Run. The prompt and the armed message are
+    waited for as fresh occurrences: a press that lands before the
+    wait has begun is not consent (rule 15), so the press must follow
+    THIS job's prompt, not an earlier job's."""
+    prompts, armed = s.count(PROMPT), s.count(ARMED)
     s.send_raw("M4 S100")
     s.send_raw(gcode)
-    if not wait_for(s.log, PROMPT, 5, s.sock):
+    if not wait_for_new(s.log, PROMPT, prompts, 5, s.sock):
         fail("[%s] no button prompt" % tag)
     s.press_button()
-    if not wait_for(s.log, ARMED, 5, s.sock):
+    if not wait_for_new(s.log, ARMED, armed, 5, s.sock):
         fail("[%s] the button press did not arm" % tag)
     if not s.wait_state("Run", 5):
         fail("[%s] the job never reported Run" % tag)
@@ -1045,9 +1095,158 @@ def test_lid_policy_hold():
         s.close()
 
 
+def test_verdict_pause_resumes_without_press():
+    """Rule 11: the pause tier holds under the open window without a
+    latch write; the clean verdict resumes with no press. The window is
+    proven still open by the absence of any prompt and of a second arm."""
+    s = Session("verdict-pause", disarm_s=60, switches=SW_CLOSED)
+    try:
+        start_armed_move(s, "verdict-pause", gcode="G1 X30 F120")   # 15 s of motion
+        time.sleep(0.5)
+        s.set_verdict("OVERTEMP", fire_ok=False, hold=True, resume_ok=False)
+        if not s.wait_state("Hold", 5):
+            fail("[verdict-pause] the pause tier did not hold the job (state %s)" % s.state())
+        if not wait_for(s.log, "fire masked, job held", 3, s.sock):
+            fail("[verdict-pause] the pause tier did not report the masked hold")
+        if "latch locked" in "".join(s.log):
+            fail("[verdict-pause] the pause tier wrote the latch")
+        if DISARMED in "".join(s.log):
+            fail("[verdict-pause] the pause tier closed the armed window")
+        time.sleep(1.5)
+        s.set_verdict("OK", fire_ok=True, hold=False, resume_ok=True)
+        if not s.wait_state("Run", 5):
+            fail("[verdict-pause] the clean verdict did not resume the job (state %s)" % s.state())
+        if not wait_for(s.log, "resuming", 2, s.sock):
+            fail("[verdict-pause] the resume was not reported")
+        if RESUME_PROMPT in "".join(s.log) or s.count(PROMPT) != 1:
+            fail("[verdict-pause] the resume asked for a button press")
+        if s.armed_count() != 1:
+            fail("[verdict-pause] the resume armed again (armed messages: %d)" % s.armed_count())
+        if "ALARM" in "".join(s.log):
+            fail("[verdict-pause] the pause tier raised an alarm")
+        s.sock.sendall(b"\x18")
+        print("PASS [verdict-pause]: OVERTEMP held under the open window with no latch write; "
+              "the clean verdict resumed with no press")
+    finally:
+        s.close()
+
+
+def test_verdict_fail_tier_ends_job():
+    """Rule 12: the fail tier disarms, resets the job with ALARM:3, and
+    nothing resumes it."""
+    s = Session("verdict-fail", disarm_s=60, switches=SW_CLOSED)
+    try:
+        start_armed_move(s, "verdict-fail", gcode="G1 X30 F120")
+        time.sleep(0.5)
+        s.set_verdict("AIRFLOW", fire_ok=False, hold=True, resume_ok=False)
+        if not s.wait_state("Alarm", 5):
+            fail("[verdict-fail] the fail tier did not end the job in Alarm (state %s)" % s.state())
+        if not wait_for(s.log, DISARMED, 3, s.sock):
+            fail("[verdict-fail] the fail tier did not close the armed window")
+        text = "".join(s.log)
+        if "ALARM:3" not in text:
+            fail("[verdict-fail] no ALARM:3 on the fail tier")
+        if "AIRFLOW" not in text:
+            fail("[verdict-fail] the fail tier did not name the verdict")
+        s.set_verdict("OK", fire_ok=True, hold=False, resume_ok=True)
+        time.sleep(1.5)
+        s.sock.sendall(b"~")
+        read_avail(s.sock, s.log, 1.0)
+        if not s.state().startswith("Alarm"):
+            fail("[verdict-fail] a ~ after the fail tier resumed something (state %s)" % s.state())
+        if "resuming" in "".join(s.log):
+            fail("[verdict-fail] the client resumed after the fail tier")
+        send_line(s.sock, "$X", s.log)
+        if not s.wait_state("Idle", 3):
+            fail("[verdict-fail] $X did not unlock the alarm")
+        print("PASS [verdict-fail]: AIRFLOW disarmed, reset the job with ALARM:3, and nothing "
+              "resumed it")
+    finally:
+        s.close()
+
+
+def test_sender_change_during_rearm_cancels():
+    """Rule 13: a sender change while a re-arm waits for the press
+    cancels the re-arm; the press resumes nothing and the job stays
+    held; the new sender's ~ prompts afresh."""
+    s = Session("rearm-sender-change", disarm_s=60, switches=SW_CLOSED)
+    try:
+        start_armed_move(s, "rearm-sender-change", gcode="G1 X30 F120")
+        time.sleep(1.0)
+        s.sock.close()                                     # mid-move: held, disarmed
+        time.sleep(0.5)
+        s.sock = s.connect()
+        if s.wait_state("Hold", 5) is None:
+            fail("[rearm-sender-change] the job was not held on the sender change")
+        s.sock.sendall(b"~")
+        if not wait_for(s.log, RESUME_PROMPT, 5, s.sock):
+            fail("[rearm-sender-change] the resume did not prompt for the button")
+        # A second sender change, inside the re-arm wait. The cancel is
+        # reported at the disconnect, to nobody (output with no client is
+        # discarded), so the evidence is what the press does next: nothing.
+        s.sock.close()
+        time.sleep(0.5)
+        s.sock = s.connect()
+        st = read_state(s, '"connected":true')
+        if '"arming":false' not in st:
+            fail("[rearm-sender-change] the re-arm wait survived the sender change: %r" % st)
+        # A press now is the machine's resume button for the new sender:
+        # it resumes nothing and arms nothing, it opens a fresh re-arm
+        # prompt of its own (the displaced consent is gone), and only a
+        # second press, against that prompt, re-arms the held job.
+        before = s.armed_count()
+        prompts = s.count(RESUME_PROMPT)
+        s.press_button()
+        read_avail(s.sock, s.log, 1.0)
+        if s.armed_count() != before or not s.state().startswith("Hold"):
+            fail("[rearm-sender-change] a press after the canceled re-arm resumed the job "
+                 "(armed %d -> %d, state %s)" % (before, s.armed_count(), s.state()))
+        if not wait_for_new(s.log, RESUME_PROMPT, prompts, 3, s.sock):
+            fail("[rearm-sender-change] the press after the cancel did not open a fresh re-arm prompt")
+        s.press_button()
+        end = time.time() + 10
+        while time.time() < end and s.armed_count() != before + 1:
+            read_avail(s.sock, s.log, 0.2)
+        if s.armed_count() != before + 1:
+            fail("[rearm-sender-change] the second press did not re-arm the held job")
+        s.sock.sendall(b"\x18")
+        print("PASS [rearm-sender-change]: the sender change canceled the re-arm; the next press "
+              "resumed nothing and prompted afresh, and the press after it re-armed")
+    finally:
+        s.close()
+
+
+def test_jog_does_not_extend_window():
+    """Rule 14: with the spindle off, jogs do not reset the disarm grace:
+    the window closes on time through them."""
+    s = Session("jog-grace", disarm_s=2)
+    try:
+        send_line(s.sock, "M4 S100", s.log)
+        send_line(s.sock, "G1 X1 F600", s.log)
+        if not wait_for(s.log, ARMED, 5, s.sock):
+            fail("[jog-grace] job did not arm")
+        send_line(s.sock, "M5", s.log)
+        wait_idle(s.sock, s.log)
+        t0 = time.time()
+        while time.time() - t0 < 5.0 and DISARMED not in "".join(s.log):
+            send_line(s.sock, "$J=G91X1F1200", s.log)
+            read_avail(s.sock, s.log, 0.3)
+        dt = time.time() - t0
+        if DISARMED not in "".join(s.log):
+            fail("[jog-grace] the window stayed open through %.1f s of jogging (grace 2 s)" % dt)
+        wait_idle(s.sock, s.log)
+        print("PASS [jog-grace]: the window closed after %.1f s of jogging with the spindle off" % dt)
+    finally:
+        s.close()
+
+
 def main():
     if not os.path.isfile(BIN):
         fail("controller binary not found at %s" % BIN)
+    test_verdict_pause_resumes_without_press()
+    test_verdict_fail_tier_ends_job()
+    test_sender_change_during_rearm_cancels()
+    test_jog_does_not_extend_window()
     test_job_window()
     test_status_files()
     test_laser_keys_reload()