forgetest: a test's implementation hash is its own function and its module's shared code

The implementation half of a domain fingerprint was the whole suite
file, so a two-line witness fix in laser.py re-required every laser
test and a rename in cloud.py every cloud test: sixty attended minutes
for changes that touched two test bodies. Now the hash is the test's
own function (its decorator included) together with the module's text
outside every @test function. A body edit moves that test alone; a
helper edit moves the tests of its module, which is what a helper does;
a file that defines no test in the @test form hashes whole. The gate
computes it the same way, from the same code.

Every recorded fingerprint moves once with this, so the next campaign is
a full one: the price of every later fix costing one test.

Also carries the re-targeted cloud replays that the previous commit
left in the working tree (the CI failure on 296fd68).

docs/ACCEPTANCE.md

@@ -65,9 +65,12 @@ each test's *details*.
 
 A test's **domain fingerprint** is the hash of the `(component, path,
 blob-id)` triples its coverage globs select in the image manifest, plus the
-platform identity, plus the hash of the test's own implementation. A PASS
+platform identity, plus the hash of the test's own implementation: its
-recorded under fingerprint F applies to any build whose recomputed
+function (decorator included) together with the code its suite module
-fingerprint is F - the same code computes it on the board and in the gate.
+shares among its tests, everything outside the module's `@test`
+functions. A PASS recorded under fingerprint F applies to any build whose
+recomputed fingerprint is F - the same code computes it on the board and
+in the gate.
 
 Consequences:
 
@@ -85,8 +88,9 @@ Consequences:
   init scripts, a third-party pin with no manifest entry) is layer content
   and invalidates everything; so does a pin written into a recipe body
   instead of its pin file (the safe direction).
-- A change to a test's implementation invalidates that test's earlier
+- A change inside a test's body invalidates that test's earlier passes
-  passes and no other.
+  and no other; a change to a helper its module shares invalidates the
+  tests of that module.
 - "Touched" is computed from content hashes carried in the image, never
   declared by hand.
 

forgetest/forgetest/catalog.py

@@ -15,10 +15,12 @@ under the runner with a Context (log, prompts, evidence, hardware
 helpers) and reports by returning normally (PASS) or raising
 runner.Failed (FAIL).
 """
+import ast
 import hashlib
 import inspect
 import os
 import re
+import threading
 
 from . import manifest as _manifest
 
@@ -59,12 +61,16 @@ class Test:
 
     @property
     def source_sha(self):
-        """sha256 of the module file that defines the test, line endings
+        """The hash of the test's own implementation: its function (the
-        normalized. Part of the fingerprint: a changed implementation
+        decorator included) together with the code its module shares
-        invalidates earlier passes of this test and no other."""
+        among its tests - everything outside the @test functions. Part
+        of the fingerprint: a change inside one test's body invalidates
+        that test's earlier passes and no other; a change to a helper
+        invalidates the tests of that module. A test the module does not
+        define in the @test form hashes its whole file."""
         if self._source_sha is None:
             path = inspect.getsourcefile(self.fn) or inspect.getfile(self.fn)
-            self._source_sha = source_file_sha(path)
+            self._source_sha = implementation_sha(path, self.id)
         return self._source_sha
 
     def fingerprint(self, manifest):
@@ -117,6 +123,55 @@ def source_file_sha(path):
     return hashlib.sha256(data).hexdigest()
 
 
+_PARTS = {}                 # path -> (shared sha, {test id: own sha}); the files never change under a run
+_PARTS_LOCK = threading.Lock()
+
+
+def module_parts(path):
+    """(shared sha, {test id: own sha}) for a suite module: the test ids
+    are read from the @test("...") decorators on the module's top-level
+    functions; a test's own text runs from its first decorator line to
+    the end of its body; the shared text is every other line of the
+    file. Line endings normalized."""
+    with _PARTS_LOCK:
+        hit = _PARTS.get(path)
+    if hit is not None:
+        return hit
+    with open(path, "rb") as f:
+        text = f.read().replace(b"\r\n", b"\n").decode("utf-8")
+    lines = text.split("\n")
+    spans = {}
+    try:
+        tree = ast.parse(text)
+    except SyntaxError:
+        tree = None
+    for node in (tree.body if tree is not None else []):
+        if not isinstance(node, ast.FunctionDef):
+            continue
+        for d in node.decorator_list:
+            if (isinstance(d, ast.Call) and isinstance(d.func, ast.Name) and d.func.id == "test"
+                    and d.args and isinstance(d.args[0], ast.Constant) and isinstance(d.args[0].value, str)):
+                start = min([dd.lineno for dd in node.decorator_list] + [node.lineno])
+                spans[d.args[0].value] = (start, node.end_lineno)
+    owned = set()
+    for a, b in spans.values():
+        owned.update(range(a, b + 1))
+    shared = "\n".join(ln for i, ln in enumerate(lines, 1) if i not in owned)
+    own = {tid: _manifest.sha256_text("\n".join(lines[a - 1:b])) for tid, (a, b) in spans.items()}
+    parts = (_manifest.sha256_text(shared), own)
+    with _PARTS_LOCK:
+        _PARTS[path] = parts
+    return parts
+
+
+def implementation_sha(path, test_id):
+    """The implementation hash of one test (see Test.source_sha)."""
+    shared, own = module_parts(path)
+    if test_id not in own:
+        return source_file_sha(path)
+    return _manifest.sha256_text("%s:%s" % (shared, own[test_id]))
+
+
 def test(id, *, title, subsystem, kind="auto", hardware="api", mode=None, covers=(),
          requires=(), always=False, est_min=1, steps=(), description="", actions=(),
          precheck=None):

forgetest/tests/test_cloud_suite.py

@@ -596,7 +596,8 @@ class CloudSuiteTests(unittest.TestCase):
     # -- a paused print cancelled by the lid, a running one by the app --------
     def cancel_parts(self):
         """(print prologue, the pause lines, the lid stop + park + cancel,
-        the same tail with the app's cancel as the trigger)."""
+        the same tail with the app's cancel as the trigger - what
+        cloud.oversize-stream's ending looks like)."""
         lines = fixture("lidabort")
         pre, rest = cut(lines, "waiting for button")
         run_pre, rest = cut(rest, "machine:_run_loop starting run")
@@ -609,46 +610,42 @@ class CloudSuiteTests(unittest.TestCase):
                                 "action cancelled mid-run; stopping motion") for l in rest]
         return pre, paused, rest, app_cancel
 
-    def test_pause_cancel_paths_on_the_bench_excerpt(self):
+    def test_paused_lid_cancel_on_the_bench_excerpt(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"])
-        pre, paused, lid_tail, app_tail = self.cancel_parts()
+        pre, paused, lid_tail, _app = self.cancel_parts()
         prints = []
 
         def next_print():
             prints.append(1)
             self.append(pre, delay=0.1)
-        run = self.run_test(cloud.pause_cancel_paths,
+        run = self.run_test(cloud.paused_lid_cancel,
                             hooks={"Click Done here": next_print,
                                    "the press pauses the print": lambda: self.append(paused, delay=0.05),
                                    "The print is paused: leave the lid open": lambda: (
                                        self.lid(False), self.append(lid_tail, delay=0.05)),
-                                   "Close the lid.": lambda: self.lid(True),
+                                   "Close the lid.": lambda: self.lid(True)},
-                                   "cancel the print from the app now": lambda: self.append(app_tail,
+                            test_id="cloud.paused-lid-cancel")
-                                                                                            delay=0.05)},
-                            test_id="cloud.pause-cancel-paths")
         ev = run.evidence
-        self.assertEqual(len(prints), 2)                      # two prints, one cue each
+        self.assertEqual(len(prints), 1)                      # one print, one cue
         self.assertEqual(ev["paused"], {"button pressed mid-run; pausing": True, "paused at": True})
         self.assertIn(":cancelled", ev["lid_from_pause"]["print finished"])
-        self.assertIn(":cancelled", ev["service_cancel"]["print finished"])
         self.assertTrue(ev["lid_from_pause"]["return home complete"])
-        self.assertTrue(ev["service_cancel"]["return home complete"])
+        self.assertEqual(ev["counters_after"], [0, 0, 3])
-        self.assertEqual(ev["counters_after_print1"], [0, 0, 3])
+        self.assertTrue(ev["latch_locked_after"])
-        self.assertEqual(ev["counters_after_print2"], [0, 0, 3])
+        self.assertFalse(ev["armed_after"])
-        self.assertTrue(ev["latch_locked_after_print1"] and ev["latch_locked_after_print2"])
+        self.assertFalse(ev["button_dark"])
-        self.assertFalse(ev["armed_after_print1"] or ev["armed_after_print2"])
         self.assertEqual(self.fc.posts, [])
         self.assertTrue(any("PASS: a paused print cancelled by the lid" in l for l in run.lines), run.lines)
 
-    def test_pause_cancel_paths_fails_when_the_paused_print_resumes_instead(self):
+    def test_paused_lid_cancel_fails_when_the_paused_print_resumes_instead(self):
         # a lid that resumed (or was ignored) leaves the print ':completed'
         self.in_cloud(pid=1522)
         self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
         pre, paused, lid_tail, _app = self.cancel_parts()
         lid_tail = [l.replace(':cancelled"', ':completed"') for l in lid_tail]
         self.assertFails(
-            cloud.pause_cancel_paths, "print 1 did not end ':cancelled'",
+            cloud.paused_lid_cancel, "the print did not end ':cancelled'",
             hooks={"Click Done here": lambda: self.append(pre, delay=0.1),
                    "the press pauses the print": lambda: self.append(paused, delay=0.05),
                    "The print is paused: leave the lid open": lambda: (

forgetest/tests/test_manifest.py

@@ -1,3 +1,4 @@
+import os
 import unittest
 
 import helpers
@@ -151,3 +152,84 @@ class CatalogTests(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class ImplementationHashTests(unittest.TestCase):
+    """A test's implementation hash is its own function plus the code its
+    module shares: a body edit moves one test, a helper edit moves them
+    all, and a file without @test functions hashes whole."""
+
+    MODULE = '''"""a suite module"""
+from forgetest.catalog import test
+
+
+def helper(x):
+    return x + 1
+
+
+@test("m.one", title="one", subsystem="m")
+def one(ctx):
+    ctx.log(helper(1))
+
+
+@test("m.two", title="two", subsystem="m",
+      steps=["a step"])
+def two(ctx):
+    ctx.log(helper(2))
+'''
+
+    def setUp(self):
+        import tempfile
+        self.tmp = tempfile.mkdtemp(prefix="forgetest-impl-")
+        self.path = os.path.join(self.tmp, "mod.py")
+        self.write(self.MODULE)
+
+    def tearDown(self):
+        import shutil
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def write(self, text):
+        catalog._PARTS.pop(self.path, None)
+        with open(self.path, "w", newline="\n") as f:
+            f.write(text)
+
+    def shas(self):
+        return catalog.implementation_sha(self.path, "m.one"), catalog.implementation_sha(self.path, "m.two")
+
+    def test_a_body_edit_moves_that_test_only(self):
+        a1, b1 = self.shas()
+        self.write(self.MODULE.replace("ctx.log(helper(2))", "ctx.log(helper(3))"))
+        a2, b2 = self.shas()
+        self.assertEqual(a1, a2)
+        self.assertNotEqual(b1, b2)
+
+    def test_a_decorator_edit_moves_that_test_only(self):
+        a1, b1 = self.shas()
+        self.write(self.MODULE.replace('steps=["a step"]', 'steps=["another step"]'))
+        a2, b2 = self.shas()
+        self.assertEqual(a1, a2)
+        self.assertNotEqual(b1, b2)
+
+    def test_a_helper_edit_moves_every_test_of_the_module(self):
+        a1, b1 = self.shas()
+        self.write(self.MODULE.replace("return x + 1", "return x + 2"))
+        a2, b2 = self.shas()
+        self.assertNotEqual(a1, a2)
+        self.assertNotEqual(b1, b2)
+
+    def test_line_endings_do_not_count(self):
+        a1, b1 = self.shas()
+        catalog._PARTS.pop(self.path, None)
+        with open(self.path, "wb") as f:
+            f.write(self.MODULE.replace("\n", "\r\n").encode())
+        self.assertEqual((a1, b1), self.shas())
+
+    def test_an_unknown_id_hashes_the_whole_file(self):
+        self.assertEqual(catalog.implementation_sha(self.path, "m.none"), catalog.source_file_sha(self.path))
+
+    def test_every_suite_test_is_spanned(self):
+        import inspect
+        reg = catalog.load_suite()
+        for t in catalog.all_tests(reg):
+            path = inspect.getsourcefile(t.fn)
+            self.assertIn(t.id, catalog.module_parts(path)[1], t.id)