diff --git a/misc/qet-mcp/README.md b/misc/qet-mcp/README.md index 546014f1e..5504af40d 100644 --- a/misc/qet-mcp/README.md +++ b/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?** diff --git a/misc/qet-mcp/qet_mcp.py b/misc/qet-mcp/qet_mcp.py index 0e994560a..ccbbf488b 100755 --- a/misc/qet-mcp/qet_mcp.py +++ b/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\. + 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} diff --git a/misc/qet-mcp/test_qet_mcp.py b/misc/qet-mcp/test_qet_mcp.py index ebbe49362..7f0a50862 100644 --- a/misc/qet-mcp/test_qet_mcp.py +++ b/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.