From 9e9d03ba94919d916c7f858c7d512587e106b69d Mon Sep 17 00:00:00 2001
From: ScottW514 <s.e.wiederhold@gmail.com>
Date: Fri, 14 Aug 2026 19:07:29 -0400
Subject: [PATCH] bench: armed-window lifecycle harness (null-sink, host-side)

Four sessions against the reported messages: arm once per job with
M5/M3 persistence and the M2 close (grace pushed beyond the horizon so
only the program-end path can disarm); a displaced sender must re-arm;
the disarm grace counts down in Hold; a blocking cooling verdict
refuses the arm. Test-the-test proven: a build with the job-based
window reverted fails at the first discriminating assertion. Wired
into the controller repo CI alongside the emission harness.
---
 scripts/bench/laser_lifecycle_test.py | 284 ++++++++++++++++++++++++++
 1 file changed, 284 insertions(+)
 create mode 100644 scripts/bench/laser_lifecycle_test.py

diff --git a/scripts/bench/laser_lifecycle_test.py b/scripts/bench/laser_lifecycle_test.py
new file mode 100644
index 0000000..1802fc3
--- /dev/null
+++ b/scripts/bench/laser_lifecycle_test.py
@@ -0,0 +1,284 @@
+#!/usr/bin/env python3
+"""Host-side verification of the operator-armed window lifecycle.
+
+Runs the native grblHAL_glowforge binary in null-sink mode (no
+hardware, no root; arming is automatic without the physical button)
+and drives jobs over TCP, asserting the window's state machine on the
+reported messages:
+
+  1. the first laser-on of a job arms exactly once, and the window
+     persists across M5/M3 toggles within the job (no re-prompt)
+  2. program end (M2) closes the window: the disarm lands promptly,
+     and the next job arms afresh
+  3. a sender change closes the window: a reconnected client must
+     re-arm - the displaced session's consent does not carry over
+  4. the spindle-off grace counts down in HOLD, not only in Idle - a
+     job abandoned in feed hold disarms after laser_disarm_s
+  5. arming is refused while the cooling verdict blocks fire, and no
+     armed message appears
+
+The disarm grace is shortened via a temp config (GFHOME_CONF) and the
+cooling verdict is published hermetically (GF_VERDICT_FILE), the same
+overrides the emission harness uses.
+
+Usage: laser_lifecycle_test.py [path-to-binary]
+       (default ./build-native/grblHAL_glowforge)
+"""
+import os
+import re
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+BIN = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "build-native/grblHAL_glowforge")
+PORT = 2398
+
+ARMED = "laser armed"
+DISARMED = "laser disarmed - latch locked"
+BLOCKED = "laser fire blocked"
+
+
+def fail(msg):
+    print("FAIL: %s" % msg)
+    sys.exit(1)
+
+
+def read_avail(sock, log, timeout, until=None):
+    end = time.time() + timeout
+    buf = b""
+    while time.time() < end:
+        sock.settimeout(max(0.05, end - time.time()))
+        try:
+            data = sock.recv(4096)
+        except socket.timeout:
+            data = b""
+        except OSError:
+            return None
+        if data:
+            buf += data
+            log.append(data.decode(errors="replace"))
+            if until:
+                text = buf.decode(errors="replace")
+                for token in until:
+                    if token in text or re.search(r"^%s\b" % re.escape(token), text, re.M):
+                        return token
+        elif until is None:
+            return None
+    return None
+
+
+def send_line(sock, line, log, tolerate_error=False):
+    sock.sendall((line + "\n").encode())
+    r = read_avail(sock, log, 5.0, until=("ok", "error"))
+    if r is None:
+        fail("no ok/error for %r" % line)
+    if r == "error" and not tolerate_error:
+        fail("error response to %r" % line)
+    return r
+
+
+def wait_for(log, needle, timeout, sock=None):
+    """Wait until needle appears in the accumulated log (draining the
+    socket while waiting when one is given)."""
+    end = time.time() + timeout
+    while time.time() < end:
+        if needle in "".join(log):
+            return True
+        if sock is not None:
+            read_avail(sock, log, 0.2)
+        else:
+            time.sleep(0.1)
+    return False
+
+
+def wait_idle(sock, log):
+    for _ in range(100):
+        sock.sendall(b"?")
+        read_avail(sock, log, 0.3)
+        if re.search(r"<Idle", "".join(log[-3:])):
+            return
+        time.sleep(0.2)
+    fail("controller never returned to Idle")
+
+
+def publish_verdicts(path, stop, fire_ok):
+    while not stop.is_set():
+        body = ('{"ts_mono":%.3f,"fire_ok":%s,"hold":false,'
+                '"resume_ok":true,"reason":""}'
+                % (time.clock_gettime(time.CLOCK_MONOTONIC),
+                   "true" if fire_ok else "false"))
+        tmp = path + ".tmp"
+        with open(tmp, "w") as f:
+            f.write(body)
+        os.replace(tmp, path)
+        stop.wait(0.5)
+
+
+class Session:
+    """One controller process with the lifecycle overrides applied."""
+
+    def __init__(self, name, fire_ok=True, disarm_s=2):
+        self.name = name
+        self.workdir = tempfile.mkdtemp(prefix="laser-lifecycle-")
+        conf = os.path.join(self.workdir, "forgefirm.conf")
+        with open(conf, "w") as f:
+            f.write("laser_disarm_s = %d\n" % disarm_s)
+        verdict = os.path.join(self.workdir, "cooling.state")
+        env = dict(os.environ, GF_VERDICT_FILE=verdict, GFHOME_CONF=conf)
+        env.pop("GFSINK", None)
+        self.stop = threading.Event()
+        self.pub = threading.Thread(target=publish_verdicts,
+                                    args=(verdict, self.stop, fire_ok),
+                                    daemon=True)
+        self.pub.start()
+        self.proc = subprocess.Popen([BIN, "-p", str(PORT)],
+                                     cwd=self.workdir, env=env,
+                                     stdout=subprocess.DEVNULL,
+                                     stderr=subprocess.PIPE)
+        self.log = []
+        self.sock = self.connect()
+
+    def connect(self):
+        for _ in range(50):
+            try:
+                s = socket.create_connection(("127.0.0.1", PORT), timeout=1)
+                read_avail(s, self.log, 0.5)    # banner
+                return s
+            except OSError:
+                time.sleep(0.1)
+        fail("[%s] cannot connect to the controller" % self.name)
+
+    def armed_count(self):
+        return "".join(self.log).count(ARMED)
+
+    def disarmed_count(self):
+        return "".join(self.log).count(DISARMED)
+
+    def close(self):
+        try:
+            self.sock.close()
+        except OSError:
+            pass
+        self.proc.send_signal(signal.SIGINT)
+        try:
+            self.proc.wait(5)
+        except subprocess.TimeoutExpired:
+            self.proc.kill()
+        self.stop.set()
+        self.pub.join(2)
+        shutil.rmtree(self.workdir, ignore_errors=True)
+
+
+def test_job_window():
+    """Rules 1 + 2: arm once per job, persist across M5/M3, close at M2,
+    fresh consent for the next job. The grace is set far beyond the test
+    horizon so only the program-end close can produce the disarm."""
+    s = Session("job-window", disarm_s=60)
+    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("[job-window] first laser-on did not arm")
+        send_line(s.sock, "M5", s.log)
+        send_line(s.sock, "M3 S200", s.log)
+        send_line(s.sock, "G1 X0 F600", s.log)
+        wait_idle(s.sock, s.log)
+        if s.armed_count() != 1:
+            fail("[job-window] M5/M3 inside one job re-prompted the arm "
+                 "(armed messages: %d)" % s.armed_count())
+        send_line(s.sock, "M2", s.log)
+        if not wait_for(s.log, DISARMED, 5, s.sock):
+            fail("[job-window] program end (M2) did not close the armed window")
+        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) or s.armed_count() != 2:
+            fail("[job-window] next job after M2 did not arm afresh")
+        wait_idle(s.sock, s.log)
+        print("PASS [job-window]: armed once per job, M2 closed the window, "
+              "next job re-armed")
+    finally:
+        s.close()
+
+
+def test_sender_change():
+    """Rule 3: a reconnected sender must re-arm. The grace is set far
+    beyond the test horizon so only the sender change can close the
+    window, and the first job ends in M5 so the next M4 is a fresh
+    laser-on edge (the arm prompt fires on the off->on edge)."""
+    s = Session("sender-change", disarm_s=60)
+    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("[sender-change] first laser-on did not arm")
+        send_line(s.sock, "M5", s.log)
+        wait_idle(s.sock, s.log)
+        s.sock.close()
+        time.sleep(0.5)
+        s.sock = s.connect()
+        before = s.armed_count()
+        send_line(s.sock, "M4 S100", s.log)
+        send_line(s.sock, "G1 X0 F600", s.log)
+        end = time.time() + 5
+        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("[sender-change] the reconnected sender inherited the "
+                 "displaced session's armed window")
+        wait_idle(s.sock, s.log)
+        print("PASS [sender-change]: displaced consent did not survive; "
+              "the new sender re-armed")
+    finally:
+        s.close()
+
+
+def test_hold_grace():
+    """Rule 4: the disarm grace counts down in HOLD."""
+    s = Session("hold-grace", disarm_s=2)
+    try:
+        send_line(s.sock, "M4 S100", s.log)
+        send_line(s.sock, "G1 X30 F60", s.log)     # ~30 s move
+        if not wait_for(s.log, ARMED, 5, s.sock):
+            fail("[hold-grace] job did not arm")
+        time.sleep(1.0)
+        s.sock.sendall(b"!")                        # feed hold
+        if not wait_for(s.log, DISARMED, 10, s.sock):
+            fail("[hold-grace] a job abandoned in Hold stayed armed past "
+                 "the disarm grace")
+        print("PASS [hold-grace]: the armed window closed while parked in Hold")
+    finally:
+        s.close()
+
+
+def test_verdict_blocks_arm():
+    """Rule 5: a blocking cooling verdict refuses the arm."""
+    s = Session("verdict-blocked", fire_ok=False)
+    try:
+        send_line(s.sock, "M4 S100", s.log, tolerate_error=True)
+        send_line(s.sock, "G1 X1 F600", s.log, tolerate_error=True)
+        if not wait_for(s.log, BLOCKED, 5, s.sock):
+            fail("[verdict-blocked] no refusal message for a blocking verdict")
+        if ARMED in "".join(s.log):
+            fail("[verdict-blocked] armed despite a blocking cooling verdict")
+        print("PASS [verdict-blocked]: arming refused while fire_ok=false")
+    finally:
+        s.close()
+
+
+def main():
+    if not os.path.isfile(BIN):
+        fail("controller binary not found at %s" % BIN)
+    test_job_window()
+    test_sender_change()
+    test_hold_grace()
+    test_verdict_blocks_arm()
+    print("PASS: the armed-window lifecycle holds")
+
+
+if __name__ == "__main__":
+    main()