Fusion - the Fusion 360 Bridge

@@ -265,6 +265,44 @@ _SECTION_BY_VERB = {     "fusion_uninstall": "bridge-lifecycle", "fusion_set_auto_update": "bridge-lifecycle", } +def stage_dynamic(target, kind):+    """Stage a dynamic fixture on the box at call time. Returns (args, None) or (None, why).++    Fixtures are real user files, staged the way a user's AI would stage them: text+    fixtures (STEP is ISO-10303 ASCII, .lbr is XML) travel via write_file to the+    scratch dir; the URN comes from a live aps_search. No test passes on canned data.+    """+    try:+        if kind == "staged_step":+            content = (REPO / "examples" / "ADS8588SIPM.step").read_text()+            dest = SCRATCH + "/ADS8588SIPM.step"+            r, _ = call(target, "write_file",+                        {"path": dest, "content": content,+                         "reason": "verb-test fixture: STEP sample for fusion_import_step"}, timeout=120)+            if not (r.get("success") or r.get("status") == "ok"):+                return None, "write_file: " + str(r.get("error"))[:80]+            return {"filePath": dest}, None+        if kind == "staged_lbr":+            content = (REPO / "examples" / "ADS8588SIPM.fusion.lbr").read_text()+            dest = SCRATCH + "/ADS8588SIPM.fusion.lbr"+            r, _ = call(target, "write_file",+                        {"path": dest, "content": content,+                         "reason": "verb-test fixture: .lbr sample for fusion_open_lbr"}, timeout=120)+            if not (r.get("success") or r.get("status") == "ok"):+                return None, "write_file: " + str(r.get("error"))[:80]+            return {"filePath": dest}, None+        if kind == "urn_from_search":+            r, _ = call(target, "fusion_aps_search", {"query": "BQ25792", "limit": 5}, timeout=90)+            for item in (r.get("results") or r.get("items") or []):+                urn = item.get("urn") or item.get("id")+                if urn and str(urn).startswith("urn:"):+                    return {"urn": urn, "wait": False}, None+            return None, "no URN in aps_search results"+        return None, "unknown fixture kind " + kind+    except Exception as e:+        return None, str(e)[:100]++ def section_of(name):     base = name.split("#")[0]     if base in _SECTION_BY_VERB:@@ -294,25 +332,34 @@ def main():     fusion_was_running = pre.get("errorCode") != "fusion_not_running"      described = {v[0] for v in describe._T}-    untriaged = described - {k.split("#")[0] for k in PLAN}+    planned = {k.split("#")[0] for k in PLAN} - {"focus_invariance"}     results, t0 = [], time.time()-    for name in sorted(untriaged):+    # Catalog drift fails the run in BOTH directions: a described verb nobody triaged,+    # AND a planned verb missing from describe (fusion_import_step shipped undescribed+    # for weeks until this reverse check was added, 2026-08-14).+    for name in sorted(described - planned):         results.append({"verb": name, "status": "FAIL", "why": "described but untriaged in tests/run_verb_tests.py"})+    for name in sorted(planned - described):+        results.append({"verb": name, "status": "FAIL", "why": "in the test plan but MISSING from describe.py - undescribed verbs are invisible to callers"})      for name, spec in PLAN.items():         if spec["tier"] == "manual" or spec["tier"] not in tiers or (only and name not in only):             if spec["tier"] == "manual" and not only:                 results.append({"verb": name, "status": "MANUAL", "why": spec.get("note", "")})             continue-        if "_dynamic" in spec["args"]:-            results.append({"verb": name, "status": "SKIP", "why": "dynamic fixture not staged this run"})-            continue+        run_args = spec["args"]+        if "_dynamic" in run_args:+            run_args, stage_err = stage_dynamic(a.target, run_args["_dynamic"])+            if run_args is None:+                results.append({"verb": name, "status": "FAIL", "why": "fixture staging failed: " + stage_err})+                print("  %-42s FAIL - fixture staging failed: %s" % (name, stage_err))+                continue         verb = name.split("#")[0]         vt = max(90, CATALOG_TIMEOUTS.get(verb, 60) + 30)-        if "_pollDoc" in spec["args"]:-            d, dt = call_polled(a.target, verb, spec["args"], spec["args"]["_pollDoc"], spec["args"]["_pollSec"])+        if "_pollDoc" in run_args:+            d, dt = call_polled(a.target, verb, run_args, run_args["_pollDoc"], run_args["_pollSec"])         else:-            d, dt = call(a.target, verb, spec["args"], timeout=vt)+            d, dt = call(a.target, verb, run_args, timeout=vt)         errs = expect(d, spec["exp"])         if name.startswith("fusion_search_cloud_files") or name.startswith("fusion_walk_cloud_tree"):             # must refuse: success:true here is the FAILURE