qet-mcp: clean up .qetz copies, never keep an unfinished one, bound scans

Review of #1472 (scorpio810, points 4-7):

- The .qet made from each .qetz lived in its own temporary folder that was
  never removed. Only the newest 8 are kept now, an older version of the
  same file is dropped when a newer one is read, and all of them are
  removed when the server exits.
- A --resave that failed or timed out could leave a partial .qet behind,
  which was kept and read. Only a run that succeeded is used now.
- qet_scan read every .qetz in the folder, each through QElectroTech with
  up to 900 s. It reads the first max_qetz (default 20) and lists the rest
  as skipped.
- test_edit_writes_a_qetz checks the rotated symbol is the one reported.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

misc/qet-mcp/qet_mcp.py

@@ -56,6 +56,7 @@ workspace policy applies exactly as it does over stdio.
 
 from __future__ import annotations
 
+import atexit
 import datetime
 import json
 import re
@@ -110,7 +111,21 @@ def _root(path: str) -> ET.Element:
 # folio. The tools that read a project's XML read the .qet QElectroTech
 # saves from it, so they read what QElectroTech reads, whatever the zipped
 # format holds; one copy per version of the file (path, size, time).
+# Only the newest few are kept, and the server removes them when it exits.
 _QETZ_COPIES: dict = {}
+_QETZ_COPIES_MAX = 8
+
+
+def _forget_qetz_copy(key) -> None:
+    out = _QETZ_COPIES.pop(key, None)
+    if out is not None:
+        shutil.rmtree(out.parent, ignore_errors=True)
+
+
+@atexit.register
+def _forget_qetz_copies() -> None:
+    for key in list(_QETZ_COPIES):
+        _forget_qetz_copy(key)
 
 
 def _is_qetz(path) -> bool:
@@ -129,6 +144,7 @@ def _as_qet(path) -> Path:
     key = (str(p.resolve()), st.st_size, st.st_mtime_ns)
     cached = _QETZ_COPIES.get(key)
     if cached is not None and cached.is_file():
+        _QETZ_COPIES[key] = _QETZ_COPIES.pop(key)   # newest last
         return cached
     binary = resolve_binary()
     if not binary:
@@ -138,12 +154,21 @@ def _as_qet(path) -> Path:
     out = folder / (p.stem + ".qet")
     # Saving as .qet builds every folio: allow a big project its time
     result = _run_qet(str(binary), ["--resave", str(p), str(out)], timeout=900)
-    if not out.is_file() or out.stat().st_size == 0:
+    # A run that timed out or failed may have left part of a .qet behind
+    if not result.get("ok") or not out.is_file() or out.stat().st_size == 0:
         shutil.rmtree(folder, ignore_errors=True)
-        detail = (result.get("stderr") or result.get("stdout") or "").strip()[-400:]
+        if result.get("timed_out"):
+            detail = f"timed out after {result.get('timeout_s')} s"
+        else:
+            detail = (result.get("stderr") or result.get("stdout") or "").strip()[-400:]
         raise ValueError(f"QElectroTech could not read {p.name}"
                          + (f": {detail}" if detail else ""))
+    # An older version of the same file is not read again
+    for old in [k for k in _QETZ_COPIES if k[0] == key[0]]:
+        _forget_qetz_copy(old)
     _QETZ_COPIES[key] = out
+    while len(_QETZ_COPIES) > _QETZ_COPIES_MAX:
+        _forget_qetz_copy(next(iter(_QETZ_COPIES)))
     return out
 
 
@@ -840,12 +865,17 @@ def _num(v) -> float:
 
 
 def tool_scan(directory: str, tag: str = "conductor",
-              attribute: str = "cable", recursive: bool = True) -> dict:
+              attribute: str = "cable", recursive: bool = True,
+              max_qetz: int = 20) -> dict:
     """Sweep every .qet and .qetz in a directory, counting how many <tag> carry a
     non-empty `attribute`.
 
     This is the corpus question: "3190 conductors, 0 cable values across
     25 projects" is exactly one call to this tool.
+
+    Each .qetz is read by QElectroTech saving it as a .qet, which can take
+    minutes for a big project, so only the first `max_qetz` are read; the
+    others are listed as skipped.
     """
     d = Path(directory).expanduser()
     if not d.is_dir():
@@ -853,8 +883,14 @@ def tool_scan(directory: str, tag: str = "conductor",
     files = sorted([f for pattern in ("*.qet", "*.qetz")
                     for f in (d.rglob(pattern) if recursive else d.glob(pattern))])
     total = non_empty = 0
-    per_file, values, unreadable = [], {}, []
+    per_file, values, unreadable, skipped = [], {}, [], []
+    qetz_read = 0
     for f in files:
+        if _is_qetz(f):
+            if qetz_read >= max_qetz:
+                skipped.append(f.name)
+                continue
+            qetz_read += 1
         try:
             root = ET.parse(_as_qet(f)).getroot()
         except (ET.ParseError, ValueError) as exc:
@@ -872,6 +908,9 @@ def tool_scan(directory: str, tag: str = "conductor",
         per_file.append({"file": f.name, tag: n, "non_empty": ne})
     return {
         "directory": str(d), "files": len(files), "unreadable": unreadable,
+        **({"skipped_qetz": skipped,
+            "note": f"only the first {max_qetz} .qetz were read (max_qetz)"}
+           if skipped else {}),
         "tag": tag, "attribute": attribute,
         "total": total, "non_empty": non_empty,
         "distinct_values": sorted(values.items(), key=lambda kv: -kv[1])[:25],
@@ -4902,10 +4941,12 @@ TOOLS = [
     },
     {
         "name": "qet_scan",
-        "description": "Sweep every .qet in a directory and count how many nodes of "
-                       "a given tag carry a non-empty attribute, with the distinct "
-                       "values found. For corpus questions such as how many "
-                       "conductors in the shipped examples have a cable value.",
+        "description": "Sweep every .qet and .qetz in a directory and count how many "
+                       "nodes of a given tag carry a non-empty attribute, with the "
+                       "distinct values found. For corpus questions such as how many "
+                       "conductors in the shipped examples have a cable value. Each "
+                       ".qetz is read through QElectroTech, which can take minutes "
+                       "for a big project: only the first max_qetz are read.",
         "inputSchema": {
             "type": "object",
             "properties": {
@@ -4913,12 +4954,14 @@ TOOLS = [
                 "tag": {"type": "string", "default": "conductor"},
                 "attribute": {"type": "string", "default": "cable"},
                 "recursive": {"type": "boolean", "default": True},
+                "max_qetz": {"type": "integer", "default": 20},
             },
             "required": ["directory"],
         },
         "handler": lambda a: tool_scan(a["directory"], a.get("tag", "conductor"),
                                        a.get("attribute", "cable"),
-                                       a.get("recursive", True)),
+                                       a.get("recursive", True),
+                                       a.get("max_qetz", 20)),
     },
     {
         "name": "qet_element_info",

misc/qet-mcp/test_qet_mcp.py

@@ -6489,6 +6489,66 @@ class ZippedProjectWithoutBinary(unittest.TestCase):
                     m.tool_project_info(str(z))
             self.assertIn("QET_BINARY", str(caught.exception))
 
+    QET = '<project><diagram><conductors><conductor cable="W1"/></conductors></diagram></project>'
+
+    def fake_resave(self, ok=True, timed_out=False):
+        """A _run_qet that writes a small .qet where --resave was asked to"""
+        def run(binary, args, timeout=180, **kw):
+            Path(args[2]).write_text(self.QET)
+            if timed_out:
+                return {"ok": False, "timed_out": True, "timeout_s": timeout}
+            return {"ok": ok, "exit_code": 0 if ok else 1, "stderr": "", "stdout": ""}
+        return run
+
+    def qetz_files(self, tmp, n):
+        files = []
+        for i in range(n):
+            z = Path(tmp) / f"p{i}.qetz"
+            z.write_bytes(b"PK")
+            files.append(z)
+        return files
+
+    def test_a_failed_or_timed_out_conversion_is_not_kept(self):
+        """A .qet left behind by a run that did not finish is not read"""
+        for kind in ({"ok": False}, {"timed_out": True}):
+            with self.subTest(**kind), tempfile.TemporaryDirectory() as tmp:
+                m._forget_qetz_copies()
+                z, = self.qetz_files(tmp, 1)
+                with mock.patch.object(m, "resolve_binary", return_value="qet"), \
+                     mock.patch.object(m, "_run_qet", side_effect=self.fake_resave(**kind)):
+                    with self.assertRaises(ValueError):
+                        m._as_qet(z)
+                self.assertEqual(m._QETZ_COPIES, {})
+
+    def test_only_the_newest_conversions_are_kept_on_disk(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            m._forget_qetz_copies()
+            files = self.qetz_files(tmp, m._QETZ_COPIES_MAX + 2)
+            with mock.patch.object(m, "resolve_binary", return_value="qet"), \
+                 mock.patch.object(m, "_run_qet", side_effect=self.fake_resave()):
+                copies = [m._as_qet(z) for z in files]
+                self.assertEqual(len(m._QETZ_COPIES), m._QETZ_COPIES_MAX)
+                self.assertFalse(copies[0].parent.exists())
+                self.assertTrue(copies[-1].is_file())
+                    # A new version of a file replaces its old copy
+                os.utime(files[-1], ns=(1, 1))
+                newer = m._as_qet(files[-1])
+                self.assertFalse(copies[-1].parent.exists())
+            m._forget_qetz_copies()
+            self.assertFalse(newer.parent.exists())
+
+    def test_scan_reads_at_most_max_qetz(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            m._forget_qetz_copies()
+            self.qetz_files(tmp, 3)
+            with mock.patch.object(m, "resolve_binary", return_value="qet"), \
+                 mock.patch.object(m, "_run_qet", side_effect=self.fake_resave()) as run:
+                scan = m.tool_scan(tmp, max_qetz=2)
+            m._forget_qetz_copies()
+        self.assertEqual(run.call_count, 2)
+        self.assertEqual((scan["total"], scan["non_empty"]), (2, 2))
+        self.assertEqual(scan["skipped_qetz"], ["p2.qetz"])
+
 
 @needs_examples
 class ZippedProjects(unittest.TestCase):
@@ -6561,8 +6621,9 @@ class ZippedProjects(unittest.TestCase):
         self.assertTrue(r.get("ok"), r)
         self.assertEqual(out.read_bytes()[:2], b"PK")
             # The rotation is what the diff of the two .qetz finds
-        diff = json.dumps(m.tool_diff(str(self.qetz), str(out)))
-        self.assertIn(m._uuid_key(first["uuid"]) if hasattr(m, "_uuid_key") else first["uuid"], diff.lower())
+        rotated = m.tool_diff(str(self.qetz), str(out))["elements"]["rotated"]
+        self.assertEqual([r["uuid"] for r in rotated], [first["uuid"]])
+        self.assertNotEqual(*rotated[0]["orientation"])
 
 
 if __name__ == "__main__":