qet-mcp: live mode tools, to act on the project open in QElectroTech

qet_live_status, qet_live_run_script and qet_live_run_stored talk to a
running QElectroTech through the channel its LiveServer opens, only when
live mode is switched on in its settings and accepted at this start. They
find it through live-session.json in QElectroTech's data folder and send
its token on every request; with no session, the error says which of the
three switches is missing. Running needs QET_ENABLE_SCRIPTING=1 as editing
does; asking for the status does not.

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

misc/qet-mcp/README.md

@@ -285,6 +285,35 @@ 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.
 
+## Live mode: working in the QElectroTech you have open
+
+Every tool above works on files, with no QElectroTech window involved. The
+three `qet_live_*` tools instead act on the project open in **your**
+QElectroTech, in front of you, so you can watch, stop or undo:
+
+| | |
+|---|---|
+| `qet_live_status` | what is on screen: project, folio, selection, last undo step, stored scripts |
+| `qet_live_run_script` | run script text on the open project: one undo step named "Assistant : …" |
+| `qet_live_run_stored` | press a stored script's button |
+
+QElectroTech only listens when three things are true:
+
+1. the server has `QET_ENABLE_SCRIPTING=1`, as for editing;
+2. in QElectroTech, Configurer > Général > "Autoriser un assistant IA à agir
+   sur le projet ouvert" is ticked (off by default);
+3. at this start, you answered *Continuer* to the warning QElectroTech shows
+   every time it starts with that setting on.
+
+While it listens, the status bar says so and shows the assistant's last
+action, with an *Arrêter* button that closes the channel for the rest of
+the session. Each action is one Ctrl+Z. A script's `qet.showMessage()` is
+logged instead of opening a box nobody asked for.
+
+The channel is a local socket only your user can open. QElectroTech writes
+its name and a random token to `live-session.json` in its data folder
+(next to `scripts`), and deletes the file when the channel closes.
+
 ## Worked examples
 
 **What did that edit change?**

misc/qet-mcp/qet_mcp.py

@@ -2895,6 +2895,90 @@ def tool_script_remove(script_id: str) -> dict:
     return {"removed": removed}
 
 
+# --------------------------------------------------------------------------
+# Live mode: act on the project open in a running QElectroTech
+# --------------------------------------------------------------------------
+#
+# Everything above is headless: it reads and writes files and launches its
+# own QElectroTech. These tools instead talk to the QElectroTech the user
+# has open, which only listens when three things are true: scripting is
+# allowed, its "mode direct" setting is on (off by default), and the user
+# accepted the warning it shows at every start. It then writes
+# live-session.json (socket name and token) in its data folder -- the
+# folder the scripts folder is in -- and removes it when the channel
+# closes. Each action is one undo step in front of the user.
+
+def _live_session() -> dict:
+    path = scripts_dir().parent / "live-session.json"
+    if not path.is_file():
+        raise ValueError(
+            "no QElectroTech is listening for an assistant. Live mode needs, in "
+            "QElectroTech: Configurer QElectroTech > Général > \"Autoriser un "
+            "assistant IA à agir sur le projet ouvert\" ticked, QElectroTech "
+            "restarted, and \"Continuer\" chosen in the warning it shows at "
+            f"start. (Looked for {path}.)")
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        raise ValueError(f"unreadable live session file {path}: {exc}") from None
+
+
+def _live_call(request: dict, timeout: float = 60.0) -> dict:
+    session = _live_session()
+    request = dict(request, token=session.get("token", ""), id=1)
+    line = (json.dumps(request) + "\n").encode("utf-8")
+    name = session.get("socket", "")
+    try:
+        if os.name == "nt":
+            # QLocalServer is a named pipe on Windows; fullServerName() is
+            # already \\.\pipe\<name>.
+            with open(name, "r+b", buffering=0) as pipe:
+                pipe.write(line)
+                data = b""
+                while not data.endswith(b"\n"):
+                    chunk = pipe.read(1)
+                    if not chunk:
+                        break
+                    data += chunk
+        else:
+            import socket
+            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+                sock.settimeout(timeout)
+                sock.connect(name)
+                sock.sendall(line)
+                data = b""
+                while not data.endswith(b"\n"):
+                    chunk = sock.recv(65536)
+                    if not chunk:
+                        break
+                    data += chunk
+    except OSError as exc:
+        raise ValueError(f"could not reach QElectroTech's live channel ({exc}); it "
+                         "may have been stopped, or QElectroTech closed") from None
+    if not data.strip():
+        raise ValueError("QElectroTech closed the live channel without answering")
+    answer = json.loads(data.decode("utf-8"))
+    answer.pop("id", None)
+    return answer
+
+
+def tool_live_status() -> dict:
+    return _live_call({"cmd": "status"})
+
+
+def tool_live_run_script(source: str, name: str = "", timeout: int = 60) -> dict:
+    _require_script_consent()
+    if not isinstance(source, str) or not source.strip():
+        raise ValueError("'source' must be the script's text")
+    return _live_call({"cmd": "run_script", "source": source, "name": name or "script"},
+                      timeout)
+
+
+def tool_live_run_stored(script_id: str, timeout: int = 60) -> dict:
+    _require_script_consent()
+    return _live_call({"cmd": "run_stored", "script": _script_id(script_id)}, timeout)
+
+
 TOOLS = [
     {
         "name": "qet_project_info",
@@ -3597,6 +3681,52 @@ TOOLS = [
         },
         "handler": lambda a: tool_script_remove(a["id"]),
     },
+    {
+        "name": "qet_live_status",
+        "description": "LIVE MODE. Ask the QElectroTech the user has open what is on "
+                       "screen: the project, the folio shown (index and title), the "
+                       "selected elements, the last undo step and the stored scripts. "
+                       "Works only if the user switched live mode on in QElectroTech "
+                       "and accepted its warning at this start; the error says which "
+                       "step is missing. Changes nothing.",
+        "inputSchema": {"type": "object", "properties": {}},
+        "handler": lambda a: tool_live_status(),
+    },
+    {
+        "name": "qet_live_run_script",
+        "description": "LIVE MODE. Run script text on the project the user has open, "
+                       "in front of them, as one undo step named after 'name'. "
+                       "qet.currentFolio() is the folio on screen. Returns what the "
+                       "script logged, its error with the line if it threw, and the "
+                       "undo step (empty if nothing changed). Try a new script with "
+                       "qet_script_test on a copy first when you can. Needs "
+                       "QET_ENABLE_SCRIPTING=1 and live mode on in QElectroTech.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "source": {"type": "string", "description": "the script's text"},
+                "name": {"type": "string", "description": "what the user sees in the undo step"},
+                "timeout": {"type": "integer", "default": 60},
+            },
+            "required": ["source"],
+        },
+        "handler": lambda a: tool_live_run_script(a["source"], a.get("name", ""),
+                                                  a.get("timeout", 60)),
+    },
+    {
+        "name": "qet_live_run_stored",
+        "description": "LIVE MODE. Press a stored script's button (see "
+                       "qet_script_list) in the QElectroTech the user has open: one "
+                       "undo step. Needs QET_ENABLE_SCRIPTING=1 and live mode on in "
+                       "QElectroTech.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {"id": {"type": "string"},
+                           "timeout": {"type": "integer", "default": 60}},
+            "required": ["id"],
+        },
+        "handler": lambda a: tool_live_run_stored(a["id"], a.get("timeout", 60)),
+    },
 ]
 
 _BY_NAME = {t["name"]: t for t in TOOLS}

misc/qet-mcp/test_qet_mcp.py

@@ -166,7 +166,8 @@ class ToolRegistry(unittest.TestCase):
             "qet_project_new", "qet_element_search", "qet_check", "qet_element_build",
             "qet_continuity", "qet_items", "qet_script_api", "qet_script_test",
         "qet_script_install", "qet_script_list", "qet_script_read",
-        "qet_script_remove"})
+        "qet_script_remove", "qet_live_status", "qet_live_run_script",
+        "qet_live_run_stored"})
 
 
 class EditValidation(unittest.TestCase):
@@ -2996,6 +2997,89 @@ class ScriptToolsIntegration(unittest.TestCase):
         self.assertTrue((self.dir / "good.js").is_file())
 
 
+
+@unittest.skipIf(os.name == "nt", "the fake QElectroTech is a Unix socket")
+class LiveClient(unittest.TestCase):
+    """The qet_live_* client against a stand-in for QElectroTech's
+    LiveServer: one JSON line in, one JSON line out, token on every line."""
+
+    def setUp(self):
+        import socket, threading
+        self.tmp = tempfile.TemporaryDirectory()
+        root = Path(self.tmp.name)
+        self.scripts = root / "data" / "scripts"
+        self.scripts.mkdir(parents=True)
+        self.env = mock.patch.dict(os.environ, {"QET_MCP_SCRIPTS_DIR": str(self.scripts),
+                                                "QET_ENABLE_SCRIPTING": "1"})
+        self.env.start()
+        self.sock_path = str(root / "s")
+        self.server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.server.bind(self.sock_path)
+        self.server.listen(4)
+        self.seen = []
+
+        def serve():
+            while True:
+                try:
+                    conn, _ = self.server.accept()
+                except OSError:
+                    return
+                with conn:
+                    data = b""
+                    while not data.endswith(b"\n"):
+                        chunk = conn.recv(4096)
+                        if not chunk:
+                            break
+                        data += chunk
+                    req = json.loads(data)
+                    self.seen.append(req)
+                    ok = req.get("token") == "T0K"
+                    conn.sendall((json.dumps({"ok": ok, "id": req.get("id"),
+                                              "echo": req.get("cmd")}) + "\n").encode())
+        threading.Thread(target=serve, daemon=True).start()
+
+    def tearDown(self):
+        self.server.close()
+        self.env.stop()
+        self.tmp.cleanup()
+
+    def session(self):
+        (self.scripts.parent / "live-session.json").write_text(
+            json.dumps({"version": 1, "socket": self.sock_path, "token": "T0K"}))
+
+    def test_no_session_says_what_to_switch_on(self):
+        with self.assertRaisesRegex(ValueError, "Continuer"):
+            m.tool_live_status()
+
+    def test_requests_carry_the_token_and_the_script_id(self):
+        self.session()
+        self.assertEqual(m.tool_live_status(), {"ok": True, "echo": "status"})
+        m.tool_live_run_stored("mark")
+        m.tool_live_run_script("qet.log(1);", "Note")
+        self.assertEqual([r["token"] for r in self.seen], ["T0K"] * 3)
+        # "id" numbers the request; the stored script travels as "script"
+        # (they once shared "id", and every stored run asked for script "1")
+        self.assertEqual(self.seen[1]["script"], "mark")
+        self.assertEqual((self.seen[2]["source"], self.seen[2]["name"]), ("qet.log(1);", "Note"))
+
+    def test_running_needs_consent_status_does_not(self):
+        self.session()
+        with mock.patch.dict(os.environ, {"QET_ENABLE_SCRIPTING": ""}):
+            m.tool_live_status()
+            with self.assertRaisesRegex(ValueError, "QET_ENABLE_SCRIPTING=1"):
+                m.tool_live_run_script("qet.log(1);")
+            with self.assertRaisesRegex(ValueError, "'id' must be"):
+                os.environ["QET_ENABLE_SCRIPTING"] = "1"
+                m.tool_live_run_stored("../x")
+        self.assertEqual(len(self.seen), 1)
+
+    def test_stale_session_file(self):
+        (self.scripts.parent / "live-session.json").write_text(
+            json.dumps({"socket": self.sock_path + "-gone", "token": "T0K"}))
+        with self.assertRaisesRegex(ValueError, "could not reach"):
+            m.tool_live_status()
+
+
 class PathPolicyOverStdio(unittest.TestCase):
     """Proves the policy is actually wired into the dispatcher.