Catalog: exthost.motion-job leaves, after it passed on forgeext 31cdb40

exthost.motion-job stayed in 01f5991 as the bench proof of
machine_host_token() on the default token path. forgeext 31cdb40, the
revision c177aab pins, makes the token file overridable
(FORGEEXT_HOST_TOKEN_FILE) and keeps /run/forgefirm/ext-host.token when
the variable is unset or empty. The recipe's binary from that revision
(sha256 15ca6003b59f69fb9eadf686094430fbbd25e7cc177230a0417738484fe3427e)
ran on the bench reference with the variable unset, and
exthost.platform, exthost.service, motion.jog-roundtrip and
exthost.motion-job passed on it (campaign c-20261015103714-0c91): the
package's job reached the machine as job:org.forgetest.reference through
the credential read from the default file.

Its paths, now:
- forgectrl src/jobpost.*: tests/jobpost_test.c, POST /job's multipart
  reader (row added).
- forgectrl src/tokens.*: tests/tokens_test.c, the scoped tokens and
  the extension host's credential file (row added).
- forgectrl src/jobrun.*: still covered by the burn tests.
- forgeext src/api.*, src/machine.*, src/run.*: still covered by
  exthost.service's forgeext/**.
Both rows name tests forgectrl's build.yml runs at the pinned 22a2cfc.
No other catalog test required exthost.motion-job. exthost.py's hw
import went with it.

Fingerprints: every exthost test moves with the module text. The next
image owes exthost.service and exthost.armed-freeze for forgeext's
source change.

Proof: the coverage lint over the tree manifest from the pins finds 0
uncovered paths and 0 host-tested row problems (48 tests); pyflakes on
the touched files is clean; forgetest's unit suite under Linux passes
(327).

forgetest/forgetest/manifest.py

@@ -199,6 +199,14 @@ HOST_TESTED = [
     # logger in its own file at its level, a stray tag in system/, the kern
     # facility in kernel/.
     ("forgectrl", "src/fflog.*", "forgectrl:tests/fflog_e2e.sh", "forgectrl:.github/workflows/build.yml"),
+    # POST /job's multipart reader: the form's parts reach the runner's
+    # start; a missing name, a $ line and a malformed boundary are refused
+    # with nothing started.
+    ("forgectrl", "src/jobpost.*", "forgectrl:tests/jobpost_test.c", "forgectrl:.github/workflows/build.yml"),
+    # The scoped tokens: the closed capability list, the judgment, the
+    # store that holds only hashes, and the extension host's credential
+    # (mode 0600, exactly the host's capabilities).
+    ("forgectrl", "src/tokens.*", "forgectrl:tests/tokens_test.c", "forgectrl:.github/workflows/build.yml"),
 ]
 
 # This repository's root, when the package runs from a checkout of it; a

forgetest/forgetest/suite/exthost.py

@@ -13,7 +13,6 @@ import subprocess
 import sys
 import time
 
-from .. import hw
 from ..catalog import test
 from .forgectrl import lan_ip
 from .image import kernel_config
@@ -1304,142 +1303,3 @@ def armed_freeze(ctx):
     finally:
         _put_back(ctx, fc, work, prior, etag, raw, dir_mode)
     _as_found(ctx, fc, prior, raw, dir_mode, found_tree)
-
-
-@test("exthost.motion-job", title="A package runs a program, under every gate a sender is under",
-      subsystem="exthost", kind="auto", hardware="takeover", mode="grbl", est_min=9,
-      covers=[("forgeext", "src/api.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/run.*"),
-              ("forgectrl", "src/jobpost.*"), ("forgectrl", "src/jobrun.*"), ("forgectrl", "src/tokens.*")],
-      requires=["exthost.service", "motion.jog-roundtrip"],
-      steps=["Setup: lid closed, nothing in the bed."],
-      description="The reference package is installed with motion.job, which the operator grants. It "
-                  "writes a program into its own data directory and asks the host to run it. The program "
-                  "is dark: it moves and never commands the laser. The job reaches the machine as the "
-                  "package, under the machine lease, and the machine's own job record names it as the "
-                  "owner and counts the program's lines. A dark program commands no laser, so no armed "
-                  "window opens and there is nothing for a press to arm: the arm gates are in force and "
-                  "are never reached, and a job that fires is laser.emission-witness's. "
-                  "Every line is sent and acknowledged, and the runner ends the "
-                  "job as one that ended without a discharge - which is what a program that commands no "
-                  "laser is, and is the outcome this test wants. The job's own emission witnesses are all "
-                  "zero. The head accelerometer sees the motion and the "
-                  "laser's own witnesses stay at zero throughout: the latch locked before the window and "
-                  "after it, and no emission at any point. A program named with a path, or one that is not "
-                  "a file of the package's own data, is refused by the host with the machine never asked. "
-                  "Everything is put back as exthost.service puts it back.")
-def motion_job(ctx):
-    import shutil
-    import tempfile
-    fc = ctx.forgectrl
-    ev = ctx.evidence
-    ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked")
-    prior = fc.settings().get("ext_enabled") or ""
-    raw = read_file(record_path())
-    found_tree = _tree(EXT_ROOT)
-    dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777
-    st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()})
-    etag = hdrs.get("etag")
-    work = tempfile.mkdtemp(prefix="forgetest-ffx.")
-    owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub")
-
-    # A program that moves and never fires: no M3, no S above zero.
-    PROGRAM = "G21\nG91\nG1 X10 F2000\nG1 X-10 F2000\nG90\nM2\n"
-
-    try:
-        archive, pub = _pack_reference(work, lan_ip(), more_caps=["motion.job"])
-        shutil.copy(pub, owner_key)
-        os.chmod(owner_key, 0o644)
-        r = _forgeext("install", archive, "--consent-community", "--grant", "motion.job")
-        ctx.check(r.get("ok") is True, "the install with the operator's grant -> %s", r.get("error"))
-        st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE})
-        ctx.check(st == 200, "ext_enabled=1 -> %s %r", st, reply)
-        x = _until(ctx, lambda: _svc(REF_ID) if _svc(REF_ID).get("state") == "running" else None, 90, poll=0.5)
-        ctx.check(x, "the service is not running: %s", _svc(REF_ID))
-
-        data_dir = os.path.join(EXT_ROOT, "data", REF_ID)
-        ran = os.path.join(data_dir, "ran")
-
-        def ask_run(body, seconds=60):
-            if os.path.exists(ran):
-                os.remove(ran)
-            _write(os.path.join(data_dir, "run"), json.dumps(body))
-            return _until(ctx, lambda: json.loads(_read(ran)) if os.path.exists(ran) else None,
-                          seconds, poll=0.5)
-
-        # A program the host will not read: a path, and one that is not there.
-        for body_, why_ in (({"program": "../../../etc/passwd", "program_text": None}, "a program with a path"),
-                            ({"program": "nothere.gcode", "program_text": None}, "a program that is not there")):
-            got = ask_run(body_)
-            ev[why_.replace(" ", "_")] = got
-            ctx.check(got and got[0] == 400, "%s -> %s", why_, got)
-
-        # The real one: dark, and under the button like any other sender.
-        before = fc.get("/status")[1] or {}
-        ctx.check((before.get("laser") or {}).get("emission_samples") == 0 and before.get("laser_locked") is True,
-                  "the laser is not dark and locked before the job: %s", before.get("laser"))
-        accel = hw.AccelSampler()
-        ctx.check(accel.available, "no head accelerometer found: the motion witness is missing")
-        t0 = time.time()
-        with accel:
-            got = ask_run({"program": "job.gcode", "program_text": PROGRAM, "lit_within_s": 120,
-                           "timeout_s": 300})
-            ev["job"] = got
-            ctx.check(got and got[0] == 200, "the package's job was refused: %s", got)
-            # The machine's own record of the job: it took it, and it took
-            # it as this package. A dark program commands no laser, so no
-            # armed window opens and there is nothing for a press to arm -
-            # the arm gates are in force and simply never reached. A job
-            # that fires is laser.emission-witness's.
-            def job_rec():
-                st_, j_ = fc.get("/job")
-                return j_ if isinstance(j_, dict) else {}
-
-            # The lease owner is the job's, named for the package.
-            def owned():
-                r = job_rec()
-                return r if (r.get("owner") or "").endswith(REF_ID) and r.get("program") else None
-
-            took = _until(ctx, owned, 60, poll=0.5)
-            ev["job_record"] = {k: (took or {}).get(k) for k in ("state", "owner", "lines")}
-            ctx.log("the machine's job record: %s", ev["job_record"])
-            ctx.check(took, "the machine did not take the job as the package: %s", job_rec())
-            ctx.check((took or {}).get("lines", 0) > 0, "the machine read no lines of the program: %s", took)
-
-            ended = _until(ctx, lambda: job_rec() if job_rec().get("state") in ("done", "failed", "idle")
-                           else None, 180, poll=1.0)
-            r = ended or job_rec()
-            em = r.get("emission") or {}
-            ev["job_end"] = {"state": r.get("state"), "reason": r.get("reason"), "lit": r.get("lit"),
-                             "sent": r.get("sent"), "acked": r.get("acked"), "emission": em}
-            ctx.log("the job ended: %s", ev["job_end"])
-            ctx.check(ended, "the job did not end: %s", job_rec())
-            ctx.check(r.get("sent") == r.get("lines") and r.get("acked") == r.get("lines"),
-                      "the machine did not send and acknowledge every line: %s", ev["job_end"])
-            # A program that commands no laser ends without a discharge,
-            # and the runner says so rather than calling it a clean run.
-            # That is the outcome this test wants: the whole program ran
-            # and nothing fired.
-            ctx.check(r.get("state") == "failed" and "without a discharge" in (r.get("reason") or ""),
-                      "a program that never fires did not end as one: %s", ev["job_end"])
-            ctx.check(r.get("lit") is False and em.get("laser_on_samples") == 0
-                      and em.get("lit_s") == 0 and em.get("thermopile_delta") == 0,
-                      "the job's own emission witnesses are not all zero: %s", em)
-            ctx.sleep(1.0)
-        t1 = time.time()
-        p2px, p2py, n = accel.p2p(t0, t1)
-        ev["accel"] = {"p2p_x": p2px, "p2p_y": p2py, "samples": n}
-        ctx.log("the package's job moved the head: accel p2p x=%d y=%d over %d samples", p2px, p2py, n)
-        ctx.check(n > 0 and max(p2px, p2py) > 0, "the accelerometer saw no motion for the job: %s", ev["accel"])
-
-        after = fc.get("/status")[1] or {}
-        ev["laser"] = {"before": {"locked": before.get("laser_locked"),
-                                  "emission": (before.get("laser") or {}).get("emission_samples")},
-                       "after": {"locked": after.get("laser_locked"),
-                                 "emission": (after.get("laser") or {}).get("emission_samples")}}
-        ctx.log("the laser across the package's job: %s", ev["laser"])
-        ctx.check(after.get("laser_locked") is True, "the laser latch is not locked after the job: %s", ev["laser"])
-        ctx.check((after.get("laser") or {}).get("emission_samples") == 0,
-                  "the emission witness is not zero across a dark job: %s", ev["laser"])
-    finally:
-        _put_back(ctx, fc, work, prior, etag, raw, dir_mode)
-    _as_found(ctx, fc, prior, raw, dir_mode, found_tree)