Merge pull request #1228 from ispyisail/feature/mcp-macro-recorder

Add MCP tools to read macro recordings and check scripts against them

misc/qet-mcp/README.md

@@ -46,6 +46,7 @@ here read the model.
 | `qet_script_test` | **try a script** on a copy of a project: what it would change, what it logged, its errors |
 | `qet_script_install` | **make a button** — store a script (and an SVG icon) where QElectroTech shows it in Projet > Scripts and the Scripts toolbar |
 | `qet_script_list`, `qet_script_read`, `qet_script_remove` | the stored scripts: list, read one to change it, delete one |
+| `qet_recording_list`, `qet_recording_read`, `qet_recording_check`, `qet_recording_remove` | **macro recordings** — what you did by hand, and whether a script does the same |
 
 `qet_export` and `qet_edit` launch QElectroTech. Everything else parses the
 file directly, which is faster, needs no display, and cannot be confused by
@@ -301,6 +302,24 @@ becomes its file name only if it is `a-z`, `0-9`, `-` and `_`. Storing or
 removing a script needs `QET_ENABLE_SCRIPTING=1` like an edit does: a
 stored script runs with the user's rights when they click it.
 
+## Macro recordings: from something done by hand to a button
+
+In QElectroTech, Projet > Scripts > Enregistrer une macro records what you
+do on a project until you click it again. It saves the project before and
+after, and each step from the undo history with the folio after it. At Stop
+it offers to copy a ready-made request; paste that into the assistant.
+
+| | |
+|---|---|
+| `qet_recording_list` | the recordings, newest first |
+| `qet_recording_read` | one recording: each step as structured changes, and the overall change |
+| `qet_recording_check` | run a script on the "before" project, from the same folio and selection, and say whether the result **matches** the "after" project, or what differs |
+| `qet_recording_remove` | delete one |
+
+The usual round: read the recording, write a script that does the same in
+general (on the selected elements, say, not on these exact ones),
+`qet_recording_check` it until it matches, then `qet_script_install` it.
+
 ## Live mode: working in the QElectroTech you have open
 
 Every tool above works on files, with no QElectroTech window involved. The

misc/qet-mcp/qet_mcp.py

@@ -1024,8 +1024,9 @@ def _run_qet(binary: str, args: list[str], timeout: int = 180,
                 "started in -- in an MCP client that is the \"env\" block of "
                 "its entry in the client configuration. Only qet_query, "
                 "qet_continuity, qet_check, qet_project_new, qet_edit, "
-                "qet_script_api, qet_script_test, qet_script_install and "
+                "qet_script_api, qet_script_test, qet_script_install, "
-                "qet_script_remove need it; every other tool either reads "
+                "qet_script_remove and qet_recording_check need it; every "
+                "other tool either reads "
                 "the file directly "
                 "or uses a plain CLI flag.")
         return result
@@ -2688,6 +2689,10 @@ SERVER_INSTRUCTIONS = (
     "when they switched live mode on and accepted its warning at this "
     "start, each action one undo step, scripts written on the spot shown "
     "to them first.\n"
+    "MACRO RECORDINGS: the user records something by hand in QElectroTech "
+    "and pastes you a request naming it; qet_recording_read it, write a "
+    "general script, qet_recording_check it until it matches, then "
+    "qet_script_install it.\n"
     "Verify edits by reading the result (qet_diff, qet_elements), not by "
     "assuming them.")
 
@@ -3119,6 +3124,186 @@ def tool_live_screenshot() -> dict:
     return answer
 
 
+# --------------------------------------------------------------------------
+# Macro recordings: what a person did by hand, for a script to repeat
+# --------------------------------------------------------------------------
+#
+# QElectroTech's Projet > Scripts > Enregistrer une macro saves, per
+# recording, the whole project before and after, and each step from its
+# undo history with the folio as it was after the step. QElectroTech cannot
+# send these anywhere; these tools fetch them. The person pastes a request
+# QElectroTech copied for them, naming the recording.
+
+_RECORDING_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
+
+
+def recordings_dir() -> Path:
+    info = assistant_info() or {}
+    named = (info.get("folders") or {}).get("recordings")
+    if isinstance(named, str) and named and not os.environ.get("QET_MCP_SCRIPTS_DIR"):
+        return Path(named)
+    return scripts_dir().parent / "recordings"
+
+
+def _recording(recording_id: str) -> tuple:
+    if not isinstance(recording_id, str) or not _RECORDING_ID.match(recording_id):
+        raise ValueError("'id' must be a recording id as qet_recording_list gives it")
+    folder = recordings_dir() / recording_id
+    meta_path = folder / "recording.json"
+    if not meta_path.is_file():
+        raise ValueError(f"no recording {recording_id!r} in {recordings_dir()}")
+    return folder, json.loads(meta_path.read_text(encoding="utf-8"))
+
+
+def _nonempty(value):
+    """A diff with every empty list, empty object and bookkeeping field
+    dropped: what actually changed, or None for nothing."""
+    if isinstance(value, dict):
+        out = {}
+        for k, v in value.items():
+            if k in ("before", "after", "keyed_by", "moved_count", "distinct_move_deltas"):
+                continue
+            v = _nonempty(v)
+            if v not in (None, {}, [], "", 0):
+                out[k] = v
+        return out or None
+    if isinstance(value, list):
+        return value or None
+    return value
+
+
+def _folio_project(diagram: ET.Element, collection: ET.Element | None, path: Path) -> Path:
+    """One folio as the smallest project qet_diff reads."""
+    root = ET.Element("project", {"version": "0.100.0", "title": "step"})
+    if collection is not None:
+        root.append(collection)
+    root.append(diagram)
+    ET.ElementTree(root).write(path, encoding="utf-8", xml_declaration=False)
+    return path
+
+
+def tool_recording_list() -> dict:
+    folder = recordings_dir()
+    out = []
+    if folder.is_dir():
+        for meta_path in sorted(folder.glob("*/recording.json"), reverse=True):
+            try:
+                r = json.loads(meta_path.read_text(encoding="utf-8"))
+            except (OSError, ValueError):
+                continue
+            out.append({"id": meta_path.parent.name, "name": r.get("name"),
+                        "steps": len(r.get("steps") or []), "started": r.get("started"),
+                        "complete": r.get("complete"), "project": r.get("project_title"),
+                        "start_folio": (r.get("start") or {}).get("folio")})
+    return {"folder": str(folder), "recordings": out}
+
+
+def tool_recording_read(recording_id: str) -> dict:
+    """A recording as structured changes: each step, and before -> after."""
+    folder, meta = _recording(recording_id)
+    before = folder / "before.qet"
+    after = folder / "after.qet"
+    out = {"id": recording_id, "name": meta.get("name"), "project": meta.get("project_title"),
+           "start": meta.get("start"), "complete": meta.get("complete"),
+           "files": {"before": str(before), "after": str(after) if after.is_file() else None}}
+    if meta.get("note"):
+        out["note"] = meta["note"]
+    if after.is_file():
+        out["overall_change"] = _nonempty(tool_diff(str(before), str(after)))
+
+    before_root = ET.parse(before).getroot()
+    collection = before_root.find("collection")
+    folios = before_root.findall("diagram")
+    last_by_folio = {}
+    steps = []
+    with tempfile.TemporaryDirectory(prefix="qet-mcp-rec-") as tmp:
+        tmp = Path(tmp)
+        for step in meta.get("steps") or []:
+            row = {k: step.get(k) for k in ("n", "kind", "undo_text", "parts", "folio",
+                                            "folio_title", "selected_elements")}
+            row["parts"] = [p for p in (row.get("parts") or []) if p]
+            f = step.get("folio")
+            file = step.get("folio_file")
+            if isinstance(f, int) and file and (folder / file).is_file():
+                now = ET.parse(folder / file).getroot()
+                prev = last_by_folio.get(f)
+                if prev is None and 0 <= f < len(folios):
+                    prev = folios[f]
+                if prev is not None:
+                    a = _folio_project(prev, collection, tmp / f"a{step['n']}.qet")
+                    b = _folio_project(now, collection, tmp / f"b{step['n']}.qet")
+                    row["change"] = _nonempty(tool_diff(str(a), str(b)))
+                last_by_folio[f] = now
+            steps.append(row)
+    out["steps"] = steps
+    out["how_to_use"] = (
+        "Write a script that has the same effect in general (e.g. on "
+        "qet.selectedElements(qet.currentFolio()) rather than these uuids), then "
+        "qet_recording_check it against this recording until it matches, then "
+        "qet_script_install it.")
+    return out
+
+
+def tool_recording_check(binary: str, recording_id: str, source: str,
+                         elements_dir: str | None = None, timeout: int = 180) -> dict:
+    """Run a script on a copy of the recording's before.qet, from where the
+    person started (folio on screen, selection), and compare with after.qet."""
+    if not isinstance(source, str) or not source.strip():
+        raise ValueError("'source' must be the script's text")
+    folder, meta = _recording(recording_id)
+    after = folder / "after.qet"
+    if not after.is_file():
+        raise ValueError("this recording has no after.qet (the project was closed while "
+                         "recording), so there is nothing to compare with")
+    start = meta.get("start") or {}
+    folio = start.get("folio") if isinstance(start.get("folio"), int) else 0
+    selected = [e.get("uuid") for e in start.get("selected_elements") or [] if e.get("uuid")]
+    with tempfile.TemporaryDirectory(prefix="qet-mcp-check-") as tmp:
+        result = Path(tmp) / "result.qet"
+        shutil.copy2(folder / "before.qet", result)
+        # Where the person was: the folio on screen and what was selected.
+        # On the script's own first line, so its error lines stay its own.
+        # qet is a Qt object whose methods cannot be replaced, so a proxy
+        # answers currentFolio() and hands everything else to the real one.
+        preamble = ("var __qet = qet; qet = new Proxy(__qet, {get: function (t, k) { "
+                    "if (k === 'currentFolio') return function () { return %d; }; "
+                    "var v = t[k]; return typeof v === 'function' ? v.bind(t) : v; }}); "
+                    "%s.forEach(function (u) { __qet.selectElement(u); }); "
+                    % (folio, json.dumps(selected)))
+        script = (preamble + source + "\n"
+                  "qet.log(%s + JSON.stringify({kind: 'save', result: qet.save(%s)}));\n"
+                  % (json.dumps(_MARKER), json.dumps(str(result))))
+        run = _run_qet(binary, [str(result)], timeout=timeout,
+                       elements_dir=elements_dir, script=script, tail=400_000)
+        streams = run.get("stdout", "") + "\n" + run.get("stderr", "")
+        errors = [ln.strip() for ln in streams.splitlines() if "Script error:" in ln]
+        saved = any(_MARKER in ln and '"result": true' in ln.replace('":true', '": true')
+                    for ln in streams.splitlines())
+        out = {"errors": errors, "started_from": {"folio": folio, "selected": selected}}
+        if run.get("hint"):
+            out["hint"] = run["hint"]
+        if not saved:
+            out["matches"] = False
+            out["errors"] = errors or ["the script did not finish, so nothing was compared"]
+            return out
+        remaining = _nonempty(tool_diff(str(after), str(result)))
+        # The project's own fields (save path, save date) always differ.
+        if remaining:
+            remaining.pop("project", None)
+            remaining = remaining or None
+        out["matches"] = remaining is None and not errors
+        out["difference_from_recording"] = remaining
+        out["script_change"] = _nonempty(tool_diff(str(folder / "before.qet"), str(result)))
+        return out
+
+
+def tool_recording_remove(recording_id: str) -> dict:
+    _require_script_consent()
+    folder, _ = _recording(recording_id)
+    shutil.rmtree(folder)
+    return {"removed": str(folder)}
+
+
 TOOLS = [
     {
         "name": "qet_project_info",
@@ -3931,6 +4116,61 @@ TOOLS = [
         "inputSchema": {"type": "object", "properties": {}},
         "handler": lambda a: tool_about(),
     },
+    {
+        "name": "qet_recording_list",
+        "description": "Macro recordings the user made in QElectroTech (Projet > "
+                       "Scripts > Enregistrer une macro), newest first: id, name, "
+                       "steps, project. Reads files; changes nothing.",
+        "inputSchema": {"type": "object", "properties": {}},
+        "handler": lambda a: tool_recording_list(),
+    },
+    {
+        "name": "qet_recording_read",
+        "description": "One macro recording as structured changes: each step (its "
+                       "name in QElectroTech's undo history, the folio, what was "
+                       "selected, and what changed on the folio), and the overall "
+                       "change from before to after. Read this to write a script that "
+                       "repeats what the user did, in general.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {"id": {"type": "string"}},
+            "required": ["id"],
+        },
+        "handler": lambda a: tool_recording_read(a["id"]),
+    },
+    {
+        "name": "qet_recording_check",
+        "description": "Check a script against a recording: run it on a copy of the "
+                       "project as it was when recording started -- the same folio "
+                       "on screen, the same selection -- and compare the result with "
+                       "the project as it was when recording stopped. 'matches' true "
+                       "means the script does what the user did; otherwise "
+                       "'difference_from_recording' says what is off. Needs "
+                       "QET_ENABLE_SCRIPTING=1.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
+                "id": {"type": "string"},
+                "source": {"type": "string", "description": "the script's text"},
+                "elements_dir": {"type": "string"},
+                "timeout": {"type": "integer", "default": 180},
+            },
+            "required": ["id", "source"],
+        },
+        "handler": lambda a: tool_recording_check(a["binary"], a["id"], a["source"],
+                                                  a.get("elements_dir"), a.get("timeout", 180)),
+    },
+    {
+        "name": "qet_recording_remove",
+        "description": "Delete one macro recording. Needs QET_ENABLE_SCRIPTING=1.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {"id": {"type": "string"}},
+            "required": ["id"],
+        },
+        "handler": lambda a: tool_recording_remove(a["id"]),
+    },
 ]
 
 _BY_NAME = {t["name"]: t for t in TOOLS}
@@ -3988,11 +4228,13 @@ _DATA_PATHS = {
     "qet_script_api":     {},
     "qet_script_test":    {"read": ("project",)},
     "qet_script_install": {"read": ("test_project",)},
+    "qet_recording_check": {},
 }
 
 # Tools that launch QElectroTech, and so take "binary" and "elements_dir".
 _LAUNCHES_QET = {"qet_export", "qet_edit", "qet_query", "qet_continuity",
-                 "qet_check", "qet_project_new", "qet_script_api", "qet_script_test"}
+                 "qet_check", "qet_project_new", "qet_script_api", "qet_script_test",
+                 "qet_recording_check"}
 
 # Tools that launch QElectroTech only when given this argument.
 _LAUNCHES_QET_WITH = {"qet_script_install": "test_project"}

misc/qet-mcp/test_qet_mcp.py

@@ -168,7 +168,9 @@ class ToolRegistry(unittest.TestCase):
         "qet_script_install", "qet_script_list", "qet_script_read",
         "qet_script_remove", "qet_live_status", "qet_live_run_script",
         "qet_live_run_stored", "qet_live_command", "qet_live_show_folio",
-        "qet_live_undo_last", "qet_live_screenshot", "qet_about"})
+        "qet_live_undo_last", "qet_live_screenshot", "qet_about",
+        "qet_recording_list", "qet_recording_read", "qet_recording_check",
+        "qet_recording_remove"})
 
 
 class EditValidation(unittest.TestCase):
@@ -3189,6 +3191,117 @@ class AssistantInfoFile(unittest.TestCase):
                              Path("/x/QElectroTech/QElectroTech/qet-assistant.json"))
 
 
+class Recordings(unittest.TestCase):
+    """Macro recordings as QElectroTech saves them: recording.json,
+    before.qet, after.qet, steps/NNN.xml."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        root = Path(self.tmp.name)
+        self.env = mock.patch.dict(os.environ, {"QET_MCP_SCRIPTS_DIR": str(root / "scripts"),
+                                                "QET_ENABLE_SCRIPTING": "1",
+                                                "QET_MCP_INFO_FILE": str(root / "none.json")})
+        self.env.start()
+        self.rec = root / "recordings"
+
+    def tearDown(self):
+        self.env.stop()
+        self.tmp.cleanup()
+
+    def make(self, rid, steps=(), after=True, start=None):
+        d = self.rec / rid
+        (d / "steps").mkdir(parents=True)
+        project = '<project version="0.100.0" title="t"><diagram title="F1"/></project>'
+        (d / "before.qet").write_text(project)
+        if after:
+            (d / "after.qet").write_text(project)
+        (d / "recording.json").write_text(json.dumps({
+            "format": 1, "id": rid, "name": "Macro " + rid, "complete": after,
+            "start": start or {"folio": 0, "selected_elements": []}, "steps": list(steps)}))
+        return d
+
+    def test_list_newest_first(self):
+        self.make("20261002-100000")
+        self.make("20261002-110000")
+        ids = [r["id"] for r in m.tool_recording_list()["recordings"]]
+        self.assertEqual(ids, ["20261002-110000", "20261002-100000"])
+
+    def test_read_keeps_steps_and_drops_empty_parts(self):
+        self.make("r1", steps=[{"n": 1, "kind": "do", "undo_text": "Pivoter", "parts": ["", "x"],
+                                "folio": 0}])
+        r = m.tool_recording_read("r1")
+        self.assertEqual(r["steps"][0]["undo_text"], "Pivoter")
+        self.assertEqual(r["steps"][0]["parts"], ["x"])
+        self.assertIsNone(r["overall_change"])          # before and after are the same
+
+    def test_ids_are_checked(self):
+        for bad in ("../x", "", "a/b", None):
+            with self.subTest(id=bad), self.assertRaisesRegex(ValueError, "'id' must be"):
+                m.tool_recording_read(bad)
+        with self.assertRaisesRegex(ValueError, "no recording"):
+            m.tool_recording_read("missing")
+
+    def test_check_needs_an_after(self):
+        self.make("r2", after=False)
+        with self.assertRaisesRegex(ValueError, "no after.qet"):
+            m.tool_recording_check("/bin/true", "r2", "qet.log(1);")
+
+    def test_remove_needs_consent(self):
+        d = self.make("r3")
+        with mock.patch.dict(os.environ, {"QET_ENABLE_SCRIPTING": ""}):
+            with self.assertRaisesRegex(ValueError, "QET_ENABLE_SCRIPTING=1"):
+                m.tool_recording_remove("r3")
+        m.tool_recording_remove("r3")
+        self.assertFalse(d.exists())
+
+    def test_nonempty_keeps_only_changes(self):
+        self.assertIsNone(m._nonempty({"elements": {"before": 2, "after": 2, "added": [],
+                                                    "moved_count": 0}}))
+        self.assertEqual(m._nonempty({"elements": {"added": [1], "removed": []}}),
+                         {"elements": {"added": [1]}})
+
+
+@needs_examples
+class RecordingCheckIntegration(unittest.TestCase):
+    """qet_recording_check through a real QElectroTech: a recording made by
+    rotating one symbol; the right script matches, a wrong one does not."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        root = Path(self.tmp.name)
+        self.env = mock.patch.dict(os.environ, {"QET_MCP_SCRIPTS_DIR": str(root / "scripts"),
+                                                "QET_ENABLE_SCRIPTING": "1",
+                                                "QET_MCP_INFO_FILE": str(root / "none.json")})
+        self.env.start()
+        d = root / "recordings" / "rot"
+        (d / "steps").mkdir(parents=True)
+        shutil.copy2(Path(EXAMPLES) / "perceuse.qet", d / "before.qet")
+        first = m.tool_elements(str(d / "before.qet"))["elements"][0]
+        self.uuid = first["uuid"]
+        folio = first["folio"] - 1          # read tools count folios from 1, edits from 0
+        r = m.tool_edit(BINARY, str(d / "before.qet"),
+                        [{"op": "rotate_element", "folio": folio, "element": self.uuid, "angle": 90}],
+                        str(d / "after.qet"), ELEMENTS or None)
+        self.assertTrue(r.get("ok"), r)
+        (d / "recording.json").write_text(json.dumps({
+            "format": 1, "id": "rot", "complete": True, "steps": [],
+            "start": {"folio": folio, "selected_elements": [{"uuid": self.uuid}]}}))
+
+    def tearDown(self):
+        self.env.stop()
+        self.tmp.cleanup()
+
+    def test_right_script_matches_wrong_one_does_not(self):
+        right = ("var f = qet.currentFolio(); qet.selectedElements(f).forEach("
+                 "function (u) { qet.rotateElement(f, u, 90); });")
+        r = m.tool_recording_check(BINARY, "rot", right, ELEMENTS or None)
+        self.assertTrue(r["matches"], r)
+        wrong = right.replace("90", "180")
+        r = m.tool_recording_check(BINARY, "rot", wrong, ELEMENTS or None)
+        self.assertFalse(r["matches"])
+        self.assertIn("rotated", json.dumps(r["difference_from_recording"]))
+
+
 class PathPolicyOverStdio(unittest.TestCase):
     """Proves the policy is actually wired into the dispatcher.