Judge the cloud resume on the lines the app logs; guard every needle against the pinned app

cloud.pause-resume failed a print that paused, resumed with its laser
lead, completed and parked: the test waited for the single line "button
pressed while paused; resuming", and the app has logged that as two
lines since its feeder work ("button pressed while paused", then
"resuming (laser lead N ticks)" from _resume_retraced). The replay
fixture carried the old wording, so the host test kept passing.

The pause and resume are now judged on PAUSE_LINES + RESUME_LINES
through one checker shared by the three tests that drive a pause
(cloud.pause-resume, the streamed pause, the pause-then-lid test), which
also fails on the app's "resume refused" line with the reason. The
fixture carries the app's two lines.

So the wording cannot drift silently again: tests/test_cloud_needles.py
reads every log phrase the cloud suite greps for out of cloud.py (the
left side of each `x in ln`, every wait_log needle, the mark tuples, and
the phrases it builds) and checks each against the logger calls in the
app sources at the revisions the recipes pin, read from the manifest
cache the tree manifest builds (the sibling checkouts locally),
placeholder-aware under a rule that never lets a placeholder stand for
the phrase itself. CI now builds the tree manifest before the unit
tests so the cache is there. The old needle fails that check.

Replays added: the second press seen but no retraced restart, and a
refused resume. 159 unit tests pass; coverage lint clean. No catalog
consequence beyond the suite module's own hash.

forgetest/tests/fixtures/gfcloud-pause.log

@@ -20,7 +20,8 @@
 2026-08-17T09:42:31.288114+00:00 gfcloud[1522] INFO machine:_switch_event button released
 2026-08-17T09:42:31.612040+00:00 gfcloud[1522] INFO machine:_run_loop paused at Position(x=AxisPosition(steps=1210, mm=22.68, inch=0.89), y=AxisPosition(steps=402, mm=7.54, inch=0.30), z=AxisPosition(steps=3, mm=0.0, inch=0.0), bytes=PulsPosition(total=238043, processed=61200))
 2026-08-17T09:42:34.702311+00:00 gfcloud[1522] INFO machine:_switch_event button pushed
-2026-08-17T09:42:34.704560+00:00 gfcloud[1522] INFO machine:_run_loop button pressed while paused; resuming (laser lead 1950 ticks)
+2026-08-17T09:42:34.704560+00:00 gfcloud[1522] INFO machine:_run_loop button pressed while paused
+2026-08-17T09:42:34.706100+00:00 gfcloud[1522] INFO machine:_resume_retraced resuming (laser lead 1950 ticks)
 2026-08-17T09:42:34.901002+00:00 gfcloud[1522] INFO machine:_switch_event button released
 2026-08-17T09:42:58.130807+00:00 gfcloud[1522] INFO machine:_run_loop current state: MachineState.IDLE
 2026-08-17T09:42:58.136283+00:00 gfcloud[1522] INFO machine:_run_loop finished run

forgetest/tests/test_cloud_needles.py

@@ -0,0 +1,259 @@
+"""Every phrase the cloud suite greps for in the gfcloud log must be a
+phrase the pinned cloud app can actually log.
+
+The cloud tests judge a print on the machine's own log lines. When the
+app's wording moves (a resume that used to log one line logs two), a
+test keeps looking for the old line and fails a print that worked. The
+replay fixtures cannot catch that: they are excerpts of the old wording.
+This test reads the phrases straight out of suite/cloud.py and checks
+each against the logger calls in the app sources at the revisions the
+recipes pin (python3-gfhardware and python3-gfutilities, from the
+manifest cache the tree manifest builds, or the sibling checkouts when
+no cache is at hand), placeholder-aware: "paused at" is covered by
+'paused at %s', "warm up: holding" by '%s: holding %.1f s', and
+"authenticate_machine SUCCESS" by logger.info('SUCCESS') inside
+def authenticate_machine (the log format is module:function message).
+"""
+import ast
+import importlib.util
+import os
+import re
+import subprocess
+import unittest
+
+import helpers  # noqa: F401  (sys.path)
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+FORGETEST = os.path.dirname(HERE)
+REPO = os.path.dirname(FORGETEST)
+CLOUD_PY = os.path.join(FORGETEST, "forgetest", "suite", "cloud.py")
+APP_COMPONENTS = ("python3-gfhardware", "python3-gfutilities")
+SIBLINGS = {"python3-gfhardware": "python3-gfhardware", "python3-gfutilities": "Glowforge-Utilities"}
+
+# Phrases cloud.py builds at run time rather than writing out (a "%s: holding"
+# with the phase filled in), listed here so they are checked too.
+BUILT_PHRASES = ("warm up: holding", "cool down: holding", "warm up: skipped", "cool down: skipped",
+                 'finished with event ":cancelled"', 'finished with event ":completed"',
+                 "motion [", "print [")
+# Phrases that are not the app's: forgetest's own log, or a line prefix
+# of the log format rather than a message.
+NOT_APP = ("PASS", "gfcloud", "[")
+
+PLACEHOLDER = re.compile(r"%(?:\([^)]*\))?[-+ #0]*\d*(?:\.\d+)?[sdifrxXeEgGcu]|%%")
+
+
+# ------------------------------------------------------------ the needles
+
+def needles_in(path):
+    """String constants cloud.py tests against log lines: the left side of
+    an `x in ln`, the arguments of wait_log, endswith() arguments, and the
+    names those resolve to (module- or function-level constants)."""
+    with open(path, encoding="utf-8") as f:
+        tree = ast.parse(f.read())
+    consts = {}                                   # name -> list of str
+
+    def strings_of(node):
+        if isinstance(node, ast.Constant) and isinstance(node.value, str):
+            return [node.value]
+        if isinstance(node, (ast.List, ast.Tuple)):
+            return [s for e in node.elts for s in strings_of(e)]
+        if isinstance(node, ast.Name):
+            return consts.get(node.id, [])
+        if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
+            return strings_of(node.left) + strings_of(node.right)
+        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "list":
+            return [s for a in node.args for s in strings_of(a)]
+        return []
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
+            got = strings_of(node.value)
+            if got:
+                consts[node.targets[0].id] = got
+    found = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Compare) and any(isinstance(op, ast.In) for op in node.ops):
+            found.update(strings_of(node.left))
+        elif isinstance(node, ast.Call):
+            fn = node.func
+            if isinstance(fn, ast.Name) and fn.id == "wait_log" and len(node.args) >= 3:
+                found.update(strings_of(node.args[2]))
+            elif isinstance(fn, ast.Attribute) and fn.attr == "endswith":
+                for a in node.args:
+                    found.update(strings_of(a))
+    # the mark tuples the suite scans lines against in comprehensions
+    # (`any(m in ln for m in SESSION_MARKS)`), where the name is a loop variable
+    for name, vals in consts.items():
+        if name.endswith(("_MARKS", "_LINES")):
+            found.update(vals)
+    found.update(BUILT_PHRASES)
+    return sorted(s.strip() for s in found if s.strip() and s.strip() not in NOT_APP and " " in s or s in BUILT_PHRASES)
+
+
+# ------------------------------------------------------------ the app's lines
+
+def log_literals(source, label):
+    """(function name, fragments) for every logger call in a Python source:
+    the message's fixed text split at its placeholders."""
+    out = []
+    try:
+        tree = ast.parse(source)
+    except SyntaxError:
+        return out
+    funcs = {}
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            for sub in ast.walk(node):
+                funcs.setdefault(id(sub), node.name)
+    for node in ast.walk(tree):
+        if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
+                and node.func.attr in ("debug", "info", "warning", "warn", "error", "critical", "exception")
+                and node.args):
+            continue
+        msg = node.args[0]
+        if isinstance(msg, ast.BinOp) and isinstance(msg.op, ast.Mod):
+            msg = msg.left
+        frags = None
+        if isinstance(msg, ast.Constant) and isinstance(msg.value, str):
+            frags = [p.replace("%%", "%") for p in PLACEHOLDER.split(msg.value)]
+        elif isinstance(msg, ast.JoinedStr):
+            frags = [""]
+            for v in msg.values:
+                if isinstance(v, ast.Constant):
+                    frags[-1] += str(v.value)
+                else:
+                    frags.append("")
+        if frags is not None:
+            out.append((funcs.get(id(node), ""), frags, label))
+    return out
+
+
+def covers(frags, needle):
+    """True when `needle` is a substring of a rendering of the message whose
+    fixed text is `frags` (the gaps between fragments are placeholders; the
+    rendered line is `function message`, so frags[0] carries the function).
+    A placeholder is a value, never the phrase itself: the needle must lie
+    inside one fixed fragment, or run through the fragments in order and
+    contain at least one complete non-empty fragment."""
+    n = len(frags)
+    if any(needle in f for f in frags if f):
+        return True
+
+    def in_gap(text, i, complete):
+        """`text` starts in the gap before frags[i]."""
+        if i >= n:
+            return False
+        frag = frags[i]
+        if frag == "":
+            return complete if i == n - 1 else in_gap(text, i + 1, complete)
+        start = 0
+        while True:
+            j = text.find(frag, start)
+            if j < 0:
+                break
+            if after(text[j + len(frag):], i, True):
+                return True
+            start = j + 1
+        # the needle ends inside frags[i]: a proper prefix of it, which counts
+        # as the complete fragment the rule wants only when the gap before it
+        # was preceded by one, or when what is left of the fragment is a
+        # trailing separator (": holding " is covered by "warm up: holding")
+        for k in range(1, len(frag)):
+            if text.endswith(frag[:k]) and (complete or not frag[k:].strip()):
+                return True
+        return False
+
+    def after(text, i, complete):
+        """`text` starts right after the whole of frags[i]."""
+        if text == "":
+            return complete
+        return in_gap(text, i + 1, complete)
+
+    for i, frag in enumerate(frags):
+        for off in range(len(frag)):
+            suffix = frag[off:]
+            if needle.startswith(suffix) and after(needle[len(suffix):], i, off == 0):
+                return True
+        if i >= 1 and in_gap(needle, i, False):
+            return True
+    return False
+
+
+# ------------------------------------------------------------ sources
+
+def pinned_sources():
+    """{component: [(path, text)]} for the app components at their pinned
+    revisions, through the tree manifest's recipe parser and cache; the
+    sibling checkouts when the cache cannot be reached."""
+    spec = importlib.util.spec_from_file_location("mft", os.path.join(REPO, "scripts", "manifest-from-tree.py"))
+    mft = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mft)
+    meta = os.path.join(os.path.dirname(REPO), "meta-openglow")
+    cache = os.path.join(REPO, ".manifest-cache")
+    out = {}
+    for name, rel, layer in mft.RECIPES:
+        if name not in APP_COMPONENTS:
+            continue
+        base = REPO if layer == "forgefirm" else meta
+        files = []
+        try:
+            url, rev = mft.parse_recipe(os.path.join(base, rel))
+            repo = mft.fetch(url, rev, cache)
+            names = mft.git(["ls-tree", "-r", "--name-only", rev], cwd=repo).decode().split("\n")
+            for p in names:
+                if p.endswith(".py"):
+                    files.append((p, mft.git(["show", "%s:%s" % (rev, p)], cwd=repo).decode("utf-8", "replace")))
+            out[name] = ("pinned %s" % rev[:12], files)
+            continue
+        except (OSError, subprocess.CalledProcessError, SystemExit):
+            pass
+        sib = os.path.join(os.path.dirname(REPO), SIBLINGS[name])
+        if os.path.isdir(sib):
+            for root, _dirs, fnames in os.walk(sib):
+                if ".git" in root:
+                    continue
+                for fn in fnames:
+                    if fn.endswith(".py"):
+                        p = os.path.join(root, fn)
+                        files.append((os.path.relpath(p, sib), open(p, encoding="utf-8", errors="replace").read()))
+            out[name] = ("sibling checkout", files)
+    return out
+
+
+class CloudNeedleTests(unittest.TestCase):
+    def test_every_needle_is_a_line_the_pinned_app_can_log(self):
+        needles = needles_in(CLOUD_PY)
+        self.assertGreaterEqual(len(needles), 30, needles)
+        self.assertIn("authenticate_machine SUCCESS", needles)
+        self.assertIn("RX-EVENT: ready", needles)
+        sources = pinned_sources()
+        missing = [c for c in APP_COMPONENTS if c not in sources]
+        if missing:
+            if os.environ.get("CI"):
+                self.fail("no source for %s (run scripts/manifest-from-tree.py first)" % ", ".join(missing))
+            self.skipTest("no source for %s" % ", ".join(missing))
+        literals = []
+        for comp, (where, files) in sources.items():
+            for path, text in files:
+                literals.extend(log_literals(text, "%s:%s" % (comp, path)))
+        self.assertGreater(len(literals), 100, "too few logger calls found: is the app source complete?")
+        uncovered = []
+        for needle in needles:
+            if not any(covers([fn + " " + frags[0]] + frags[1:], needle) for fn, frags, _ in literals):
+                uncovered.append(needle)
+        self.assertEqual(uncovered, [], "cloud.py greps for lines the pinned app never logs (sources: %s)"
+                         % ", ".join("%s=%s" % (c, w) for c, (w, _) in sources.items()))
+
+    def test_cover_rules(self):
+        self.assertTrue(covers(["_run_loop paused at ", ""], "paused at"))
+        self.assertTrue(covers(["_dwell ", ": holding ", " s"], "warm up: holding"))
+        self.assertTrue(covers(["authenticate_machine SUCCESS"], "authenticate_machine SUCCESS"))
+        self.assertTrue(covers(["_resume_retraced resuming (laser lead ", " ticks)"], "resuming (laser lead"))
+        self.assertTrue(covers(["_finish_action ", " [", "]: finished with event \"", "\""],
+                               'finished with event ":completed"'))
+        self.assertFalse(covers(["_run_loop button pressed while paused"], "button pressed while paused; resuming"))
+        self.assertFalse(covers(["_run_loop paused at ", ""], "paused it"))
+
+
+if __name__ == "__main__":
+    unittest.main()

forgetest/tests/test_cloud_suite.py

@@ -11,6 +11,7 @@ What is proven here: the tests find the right lines in real log noise,
 reuse an existing cloud session and never switch back, restart the client
 for a fresh hunt, judge the print's own finish line (not another action's),
 wait the service's deferred moves out, and fail for the right reasons."""
+import contextlib
 import os
 import shutil
 import struct
@@ -311,7 +312,7 @@ class CloudSuiteTests(unittest.TestCase):
         run = self.run_test(cloud.pause_resume, hooks=hooks, test_id="cloud.pause-resume")
         ev = run.evidence
         self.assertEqual(ev["log"], {"button pressed mid-run; pausing": True, "paused at": True,
-                                     "button pressed while paused; resuming": True})
+                                     "button pressed while paused": True, "resuming (laser lead": True})
         self.assertTrue(ev["armed_after_resume"])
         self.assertIn(":completed", ev["log_end"]["print finished"])
         self.assertTrue(ev["log_end"]["return home complete"])
@@ -333,25 +334,56 @@ class CloudSuiteTests(unittest.TestCase):
                  "Press the button once NOW": lambda: self.append(rest, delay=0.05)}
         self.assertFails(cloud.pause_resume, "did not complete after the resume", hooks=hooks)
 
+    def test_pause_resume_fails_without_the_retraced_restart(self):
+        """The second press was seen but the retraced restart never logged
+        (the app's resume is its own line now): the failure names that."""
+        self.in_cloud(pid=1522)
+        self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
+        lines = [l for l in fixture("pause") if "resuming (laser lead" not in l]
+        pre, rest = cut(lines, "current state: MachineState.RUNNING")
+        pre, rest = pre + [rest[0]], rest[1:]
+        with self.fast_wait_log():
+            hooks = {"Click Done here": lambda: self.append(pre, delay=0.1),
+                     "Press the button once NOW": lambda: self.append(rest, delay=0.05)}
+            self.assertFails(cloud.pause_resume, "no retraced restart logged", hooks=hooks)
+
+    def test_pause_resume_fails_when_the_kernel_refuses_the_resume(self):
+        self.in_cloud(pid=1522)
+        self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
+        lines = [l.replace("machine:_resume_retraced resuming (laser lead 1950 ticks)",
+                           "machine:_resume_retraced resume refused ([Errno 22] Invalid argument); cancelling")
+                 for l in fixture("pause")]
+        pre, rest = cut(lines, "current state: MachineState.RUNNING")
+        pre, rest = pre + [rest[0]], rest[1:]
+        with self.fast_wait_log():
+            hooks = {"Click Done here": lambda: self.append(pre, delay=0.1),
+                     "Press the button once NOW": lambda: self.append(rest, delay=0.05)}
+            self.assertFails(cloud.pause_resume, "the resume was refused", hooks=hooks)
+
+    @contextlib.contextmanager
+    def fast_wait_log(self):
+        """The wait for the pause lines is 90 s: a replay whose line never
+        comes would sit it out. Cap it through the module's wait."""
+        saved = cloud.wait_log
+
+        def fast(ctx, offset, needles, timeout, poll=0.5):
+            return saved(ctx, offset, needles, min(timeout, 1.5), poll=0.1)
+        cloud.wait_log = fast
+        try:
+            yield
+        finally:
+            cloud.wait_log = saved
+
     def test_pause_resume_fails_without_the_pause_line(self):
         self.in_cloud(pid=1522)
         self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
         lines = [l for l in fixture("pause") if "button pressed mid-run" not in l]
         pre, rest = cut(lines, "current state: MachineState.RUNNING")
         pre, rest = pre + [rest[0]], rest[1:]
-        # the wait for the pause lines is 90 s: shorten it through the module's poll by
-        # ending the log early - the finish line arrives, but the pause never does
-        saved = cloud.wait_log
-
-        def fast_wait_log(ctx, offset, needles, timeout, poll=0.5):
-            return saved(ctx, offset, needles, min(timeout, 1.5), poll=0.1)
-        cloud.wait_log = fast_wait_log
-        try:
+        with self.fast_wait_log():
             hooks = {"Click Done here": lambda: self.append(pre, delay=0.1),
                      "Press the button once NOW": lambda: self.append(rest, delay=0.05)}
             self.assertFails(cloud.pause_resume, "did not pause the run", hooks=hooks)
-        finally:
-            cloud.wait_log = saved
 
     # -- the lid/interlock abort and the button-wait tests, on their excerpts ------
     # -- the merged lid + interlock abort test -------------------------------