Add cooling.flow-under-load: the flow check with the tube lit through its window

A live catalog case: two 30 x 4 mm fills at full power on the press, the
window held open until the engine's flow verdict lands in the forgectrl
log, then M2 and the head brought back. PASS needs a verified line with
the laser's share on it (at least 0.3 C: the window and the fire
overlapped) and the judged rise at least 1 C under cool_flow_rise; an arm
refused by a gate names the gate. Passed on image 20260829190323
(heater rise 11.9 C judged, laser 0.6 off 12.5).

CAMPAIGN-LOG records the run and the day's two air-assist gate holds (a
bench connection, reseated); BRINGUP item 21 names the case.

Catalog: the case covers forgectrl src/cool.* and its neighbors; a new
test, so the manifest's coverage widens rather than moves.

docs/BRINGUP.md

@@ -1463,11 +1463,11 @@ Open items only. Anything closed is in `CAMPAIGN-LOG.md`.
     and toggles between two levels in between. Together they put an
     ordinary job's check within a few tenths of the limit. The engine now
     reads means and takes the tube's share off (`cool_laser_heat_cw`,
-    `cool_laser_heat_density`). Owed: a `cooling.*` catalog case with an
+    `cool_laser_heat_density`), and `cooling.flow-under-load` is the
-    armed CW load; a re-measure of the two coefficients once a second
+    catalog's case. Owed: a re-measure of the two coefficients once a
-    machine is on the bench (one tube, one supply so far); and if a lit
+    second machine is on the bench (one tube, one supply so far); and if a
-    check still trips, the void-on-emission design with the tube as its
+    lit check still trips, the void-on-emission design with the tube as
-    own flow tracer.
+    its own flow tracer.
     Open question, the offset's source: the timing points at the airflow
     drive (the step lands one sample after the fans go to run duty, before
     any HV, and lifts when they go idle, long after the tube is dark), but

docs/CAMPAIGN-LOG.md

@@ -4388,6 +4388,38 @@ and the reset came twice, 0.1 s apart: the tail of the blast arrived after
 the first reset had flushed the ring and overran it again. Harmless, the
 job was already stopped.
 
+## 2026-08-29: the air-assist fan and the airflow gate, twice
+
+Twice in the day the airflow gate held a job at the end of its 15 s grace
+with the air-assist tach still at its idle reading (1895 at 17:09 in a
+`flowload t2 40` run, 2207 at 20:19 in the first bench run of
+`cooling.flow-under-load`; the floor is 6000 and every other session of the
+day read about 10,700 within 5 s of the run profile). The head probed clean
+and the kernel log carried no I2C error; the fan read normal at idle both
+times. The gate did what it is for: the hold came before the arm, the
+operator's press was refused, the session closed with the fans, and no job
+ran without air. The operator's reading is a bench hardware glitch, the
+head's pogo-pin connection to the air assist; the machine was powered down
+and the head reseated, and the case was run again. Not a project item.
+
+## 2026-08-29: `cooling.flow-under-load`, the catalog's case for the lit check
+
+The catalog case for BRINGUP item 21 (`forgetest/suite/cooling.py`, kind
+live, mode grbl, one press): two 30 x 4 mm fills at full power on the press,
+the window held open until the engine's verdict lands in the forgectrl log,
+then M2. PASS needs `coolant flow verified` with the laser's share on the
+line (at least 0.3 C, the proof the window and the fire overlapped) and the
+judged rise at least 1 C under `cool_flow_rise`; an arm refused by a gate
+names the gate. First run on image 20260829190323 (the module staged over
+the installed suite, forgetest restarted; the prerequisites overridden, no
+campaign results on this image yet): the airflow gate held before the arm
+(the entry above). Second run after the head reseat: **PASS**, the engine
+line `coolant flow verified (heater rise 11.9 C, dT 9.5 C; laser 0.6 off
+12.5)`, 66 s from job to verdict, the window closed 0.0 s after M2, the
+head jogged back by the baseline; the case now brings the head back
+itself. The case rides the next image; its campaign standing comes with
+that image's campaign.
+
 ## Superseded status notes
 
 ### Shared machine services — remaining polish, as listed 2026-08-13

forgetest/forgetest/suite/cooling.py

@@ -622,3 +622,137 @@ def fan_gate_trips(ctx):
     ctx.check(all(after.get(k, "") == orig[k] for k in FAN_KEYS),
               "settings not restored: %s", {k: after.get(k) for k in FAN_KEYS})
     ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle")
+
+
+# The flow check under a lit tube. With a prompt press the tube is lit for
+# most of the arm-time window, which adds about 1.5 C to the heater rise at
+# full power; the engine takes that share off from the tube current before
+# the limit (cool_laser_heat_*), reads its baseline and end as means, and
+# says so on the verdict line: "verified (heater rise 11.3 C, dT 9.6 C;
+# laser 0.8 off 12.0)". The share on the line is the proof the window and
+# the fire overlapped; the judged rise inside the dark band is the proof
+# the share was right.
+LOAD_FEED = 1500                    # mm/min: 25 mm/s, the engrave dose that does not char
+LOAD_PITCH = 0.3                    # mm between lines
+LOAD_WIDTH = 30.0                   # mm, the line length (+X)
+LOAD_LINES = 27                     # two 30 x 4 mm fills back to back, about 35 s lit
+LOAD_SHARE_MIN_C = 0.3              # the tube was lit inside the window
+LOAD_MARGIN_C = 1.0                 # the judged rise this far under the limit
+LOAD_VERDICT_WAIT_S = 120           # the window opens ~15 s into the session and runs 50 s
+LOAD_LOG_LINES = "300"
+_LOAD_LINE_RX = None
+
+
+def _load_verdict_rx():
+    global _LOAD_LINE_RX
+    if _LOAD_LINE_RX is None:
+        import re
+        _LOAD_LINE_RX = re.compile(
+            r"(COOLANT FLOW SUSPECT|COOLANT FLOW FAULT|coolant flow verified|"
+            r"coolant flow suspicion cleared|coolant flow recovered)[^\n]*?heater rise ([0-9.]+) C"
+            r"(?: \(limit ([0-9.]+))?, dT ([0-9.]+)(?: C)?(?:; laser ([0-9.]+) off ([0-9.]+))?\)?")
+    return _LOAD_LINE_RX
+
+
+def _load_fill():
+    lines = []
+    for i in range(LOAD_LINES):
+        lines.append("G1 X%g F%d" % (LOAD_WIDTH if i % 2 == 0 else -LOAD_WIDTH, LOAD_FEED))
+        if i < LOAD_LINES - 1:
+            lines.append("G1 Y%g F%d" % (LOAD_PITCH, LOAD_FEED))
+    return lines
+
+
+def _forgectrl_tail(fc):
+    st, body = fc.get("/logs/tail", params={"name": "forgectrl", "lines": LOAD_LOG_LINES})
+    return body.get("text", "") if st == 200 and isinstance(body, dict) else ""
+
+
+@test("cooling.flow-under-load", title="The flow check reads true with the tube lit through its window",
+      subsystem="cooling", kind="live", mode="grbl", est_min=4,
+      covers=_COOL_COVERS + [("forgectrl", "src/main.c")],
+      requires=["laser.emission-witness", "cooling.flow-verify"], actions=["button"],
+      steps=["Scrap under the head with 35 mm of free +X and 10 mm of +Y travel; lid closed; exhaust on.",
+             "Press the physical button AS SOON AS it lights white: the check must run while the "
+             "tube is lit, and the share on the verdict line is the proof it did."],
+      description="Two 30 x 4 mm fills at full power on the press (about 35 s lit), the window "
+                  "held open until the engine's flow verdict lands. The verdict must be "
+                  "'verified' with the laser's share on the line (at least 0.3 C: the tube was "
+                  "lit inside the window) and the judged rise at least 1 C under the limit, "
+                  "where the same check read within tenths of the limit before the engine took "
+                  "the tube's share off and read its baseline as a mean.")
+def flow_under_load(ctx):
+    from .laser import sample, prepare, LiveJob, ARM_CUE, run_and_sample, wait_disarm
+    fc = ctx.forgectrl
+    ev = ctx.evidence
+    state, check_s = _gate_state(fc, "cool_flow_check_s")
+    _st, limit = _gate_state(fc, "cool_flow_rise")
+    ctx.check(state != "off" and check_s, "the flow check is off (cool_flow_check_s=%s)", check_s)
+    ctx.check(limit, "cool_flow_rise unreadable")
+    ev.update({"check_s": check_s, "limit_c": limit})
+    before = _forgectrl_tail(fc).count("heater rise")
+    # The fill ends one line length out and (LOAD_LINES - 1) pitches up;
+    # the job brings the head back so the baseline finds it where it began.
+    back = "G0 X%g Y%g" % (-LOAD_WIDTH if LOAD_LINES % 2 else 0.0, -LOAD_PITCH * (LOAD_LINES - 1))
+    job = ["G91", "G21", "M3 S1000"] + _load_fill() + ["M5", back]
+    line = None
+    heater_seen = False
+    with ctx.grbl() as g, LiveJob(ctx, g):
+        prepare(ctx, g)
+        base = sample(ctx)
+        ctx.check(base, "forgectrl /status or /cool/status unavailable")
+        ctx.check(not base["emission"], "emission_samples nonzero before the job (%s)", base["emission"])
+        ctx.ready(ARM_CUE % "35 mm +X and 10 mm +Y")
+        ctx.arm_press()
+        t_job = time.time()
+        try:
+            samples = run_and_sample(ctx, g, job, overall_timeout=240)
+        finally:
+            ctx.clear_notice()
+        emis = [s["emission"] for s in samples if s["emission"] is not None]
+        ev["emission_peak"] = max(emis) if emis else 0
+        # An arm refused by a gate names itself: the engine's verdict left OK
+        # during the wait (AIRFLOW, OVERTEMP, a flow FAULT).
+        gates = sorted({s["verdict"] for s in samples if s.get("verdict") not in (None, "OK")})
+        ev["verdicts_seen"] = gates
+        ctx.check(ev["emission_peak"] > 0,
+                  "no emission witnessed: %s" % ("the engine held the run (%s) before the arm"
+                                                  % ", ".join(gates) if gates
+                                                  else "arm refused, or no button press"))
+        # The window stays open (no M2) until the check has judged.
+        t0 = time.time()
+        while time.time() - t0 < LOAD_VERDICT_WAIT_S:
+            ctx.checkpoint()
+            if hw.sysfs_int("thermal/heater_pwm", 0) > 0:
+                heater_seen = True
+            text = _forgectrl_tail(fc)
+            if text.count("heater rise") > before:
+                m = None
+                for m in _load_verdict_rx().finditer(text):
+                    pass
+                line = m.group(0) if m else None
+                break
+            ctx.sleep(1)
+        ev["heater_seen"] = heater_seen
+        ctx.log("M2: %s", g.command("G90\nM2", timeout=10))
+        dt = wait_disarm(ctx, 30)
+        ev["disarm_after_m2_s"] = round(dt, 1) if dt is not None else None
+    ctx.check(line, "the engine published no flow verdict within %d s of the job (heater seen: %s)",
+              LOAD_VERDICT_WAIT_S, heater_seen)
+    m = _load_verdict_rx().search(line)
+    kind, rise, _lim, dtc, share, raw = m.groups()
+    rise = float(rise)
+    share = float(share) if share else 0.0
+    raw = float(raw) if raw else rise
+    ev.update({"verdict_line": line, "kind": kind, "rise_c": rise, "dt_c": float(dtc),
+               "laser_share_c": share, "raw_rise_c": raw, "job_to_verdict_s": round(time.time() - t_job, 1)})
+    ctx.log("engine: %s", line)
+    ctx.check(kind == "coolant flow verified", "the check did not verify flow under load: %s", line)
+    ctx.check(share >= LOAD_SHARE_MIN_C,
+              "the tube's share on the line is %.1f C (under %.1f): the window and the fire did not "
+              "overlap - press the button as soon as it lights", share, LOAD_SHARE_MIN_C)
+    ctx.check(rise <= float(limit) - LOAD_MARGIN_C,
+              "judged rise %.1f C is within %.1f C of the %.1f C limit", rise, LOAD_MARGIN_C, float(limit))
+    ctx.check(dt is not None, "the window did not close after M2")
+    ctx.log("PASS: lit through the window, raw rise %.1f C, laser %.1f C off, judged %.1f C against %.1f "
+            "(dT %.1f), window closed %.1f s after M2", raw, share, rise, float(limit), float(dtc), dt)