From 73ab76bd5080983860497721d9111ebdc60467b6 Mon Sep 17 00:00:00 2001 From: enne2 Date: Sun, 15 Mar 2026 19:16:58 +0100 Subject: [PATCH] feat: Add auto-submit functionality for simple commands and new wait functions --- README.md | 9 +- src/gnome_vte_mcp/gui.py | 64 +++++++---- src/gnome_vte_mcp/server.py | 222 +++++++++++++++++++++++++++++++++++- 3 files changed, 271 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index d27b9d2..22f56fd 100644 --- a/README.md +++ b/README.md @@ -113,11 +113,13 @@ La variabile legacy `GNOME_VTE_MCP_SOCKET` resta accettata per compatibilita. - `open_tab(title=None, cwd=None, command=None)` - `list_tabs()` - `focus_tab(tab_id)` -- `exec_command(tab_id, command, newline=True, poll_interval=0.1)` +- `exec_command(tab_id, command, newline=True, auto_submit=False, poll_interval=0.1)` - `read_tab(tab_id, last_n_lines=200)` - `read_last_command_result(tab_id)` +- `wait_for_command(delay_seconds)` - `wait_for_running_command(tab_id, timeout=None, poll_interval=0.1)` - `wait_for_command_result(tab_id, after_sequence=None, timeout=None, poll_interval=0.1)` +- `wait_for_prompt(tab_id, timeout=None, poll_interval=0.1, idle_seconds=0.4, prompt_pattern=None)` - `capture_screenshot(target="window", tab_id=None, path=None, diagnostic_overlay=False)` - `close_tab(tab_id)` - `close_gui()` @@ -132,12 +134,15 @@ Lifecycle GUI: Per la lettura testuale dei comandi: - `read_tab(...)` restituisce lo scrollback recente, utile per debugging grezzo -- `exec_command(...)` scrive il testo nel terminale, poi resta bloccato indefinitamente finche l'utente non preme `Enter` manualmente nella GUI; il parametro `newline` resta ignorato per compatibilita +- `exec_command(...)` scrive il testo nel terminale e, di default, resta bloccato indefinitamente finche l'utente non preme `Enter` manualmente nella GUI; il parametro `newline` resta ignorato in questa modalita per compatibilita +- `exec_command(..., auto_submit=True)` invia anche `Enter` da solo, ma e volutamente ristretto a piccoli comandi read-only e senza shell syntax complessa - dopo eventuali modifiche manuali nella GUI, premi `Enter` tu per sbloccare davvero `exec_command(...)` e inviare il comando alla shell - per comandi che aprono una sessione interattiva delegata come `ssh`, `arca` o una subshell, la tab non viene trattata come bloccata: puoi continuare a scrivere nuovi comandi nella stessa sessione senza aspettare il ritorno della shell locale +- `wait_for_command(delay_seconds)` non osserva il terminale e non prova a capire se il comando e finito: e solo una pausa sincrona esplicita, utile quando il loop e `read_tab(...)` -> attesa arbitraria -> `read_tab(...)` - `wait_for_running_command(...)` aspetta il completamento del comando che e gia in esecuzione nella tab quando il terminale e occupato e non puoi ancora inviare un nuovo comando - se la tab e dentro una sessione interattiva delegata, `wait_for_running_command(...)` fallisce esplicitamente perche il completamento tracciato tornera disponibile solo quando esci da quella sessione - `wait_for_command_result(...)` aspetta in modo bloccante il completamento del comando dopo l'`after_sequence` restituito da `exec_command(...)` e restituisce `command`, `cwd`, `cwd_after`, `started_at`, `finished_at`, `duration_seconds`, `exit_code` e `text`; se `timeout` e omesso o `<= 0`, l'attesa e indefinita +- `wait_for_prompt(...)` serve per sessioni interattive delegate come `ssh`: aspetta che il transcript smetta di crescere per un breve intervallo e che l'ultima riga assomigli a un prompt - `read_last_command_result(tab_id)` restituisce l'ultimo comando completato con gli stessi metadati temporali e di path Per `capture_screenshot`: diff --git a/src/gnome_vte_mcp/gui.py b/src/gnome_vte_mcp/gui.py index 072892e..be4a1a4 100644 --- a/src/gnome_vte_mcp/gui.py +++ b/src/gnome_vte_mcp/gui.py @@ -684,27 +684,35 @@ PROMPT_COMMAND='__mcp_hal9002_prompt_hook' self._dismiss_onboarding(tab) if keyval in {Gdk.KEY_Return, Gdk.KEY_KP_Enter, Gdk.KEY_ISO_Enter} and tab.pending_submit_id is not None: submitted_at = datetime.now().astimezone().isoformat(timespec="milliseconds") - after_sequence = self._latest_command_sequence(tab) - delegated = self._looks_like_interactive_handoff(tab.pending_submit_text or "") - tab.last_manual_submit_id = tab.pending_submit_id - tab.last_manual_submit_text = tab.pending_submit_text - tab.last_manual_submit_at = submitted_at - if delegated: - tab.delegated_session_submission_id = tab.pending_submit_id - tab.delegated_session_command = tab.pending_submit_text - tab.delegated_session_started_at = submitted_at - tab.delegated_session_after_sequence = after_sequence - self._clear_current_execution(tab) - else: - tab.current_execution_submission_id = tab.pending_submit_id - tab.current_execution_command = tab.pending_submit_text - tab.current_execution_started_at = submitted_at - tab.current_execution_after_sequence = after_sequence - tab.pending_submit_id = None - tab.pending_submit_text = None - tab.pending_submit_requested_at = None + self._finalize_submission(tab, submitted_at=submitted_at) return False + def _finalize_submission(self, tab: TabState, *, submitted_at: str) -> int: + if tab.pending_submit_id is None: + raise RuntimeError(f"tab_id={tab.tab_id} has no pending submit to finalize") + + after_sequence = self._latest_command_sequence(tab) + delegated = self._looks_like_interactive_handoff(tab.pending_submit_text or "") + tab.last_manual_submit_id = tab.pending_submit_id + tab.last_manual_submit_text = tab.pending_submit_text + tab.last_manual_submit_at = submitted_at + if delegated: + tab.delegated_session_submission_id = tab.pending_submit_id + tab.delegated_session_command = tab.pending_submit_text + tab.delegated_session_started_at = submitted_at + tab.delegated_session_after_sequence = after_sequence + self._clear_current_execution(tab) + else: + tab.current_execution_submission_id = tab.pending_submit_id + tab.current_execution_command = tab.pending_submit_text + tab.current_execution_started_at = submitted_at + tab.current_execution_after_sequence = after_sequence + + tab.pending_submit_id = None + tab.pending_submit_text = None + tab.pending_submit_requested_at = None + return after_sequence + def _on_terminal_window_title_changed(self, terminal: Vte.Terminal, tab_id: str) -> None: tab = self.tabs.get(tab_id) if tab is None: @@ -1153,7 +1161,7 @@ PROMPT_COMMAND='__mcp_hal9002_prompt_hook' self.window.present() return self._serialize_tab(tab) - def rpc_exec(self, tab_id: str, command: str, newline: bool = True) -> dict[str, Any]: + def rpc_exec(self, tab_id: str, command: str, newline: bool = True, auto_submit: bool = False) -> dict[str, Any]: tab = self._require_tab(tab_id) running_status = self._running_command_status(tab) if running_status["state"] == "running": @@ -1180,6 +1188,22 @@ PROMPT_COMMAND='__mcp_hal9002_prompt_hook' tab.pending_submit_id = submission_id tab.pending_submit_text = command tab.pending_submit_requested_at = requested_at + if auto_submit: + tab.terminal.paste_text(payload + "\n") + submitted_at = datetime.now().astimezone().isoformat(timespec="milliseconds") + after_sequence = self._finalize_submission(tab, submitted_at=submitted_at) + return { + "tab_id": tab_id, + "written_text": command, + "awaiting_manual_submit": False, + "after_sequence": after_sequence, + "submission_id": submission_id, + "requested_at": requested_at, + "submitted_at": submitted_at, + "current_sequence": self._latest_command_sequence(tab), + "newline_ignored": False, + } + tab.terminal.paste_text(payload) return { "tab_id": tab_id, diff --git a/src/gnome_vte_mcp/server.py b/src/gnome_vte_mcp/server.py index 70cb957..68bd2fa 100644 --- a/src/gnome_vte_mcp/server.py +++ b/src/gnome_vte_mcp/server.py @@ -5,6 +5,7 @@ import fcntl import json import os import re +import shlex import shutil import subprocess import sys @@ -25,6 +26,81 @@ COMMAND_END_MARKER = "__MCP_HAL9002_CMD_END__" ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") OSC_ESCAPE_RE = re.compile(r"\x1b\].*?(?:\x07|\x1b\\)", re.DOTALL) SINGLE_ESCAPE_RE = re.compile(r"\x1b[@-_]") +CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]") +DEFAULT_PROMPT_RE = re.compile(r"^.*(?:[$#%>] |\]# )$") +AUTO_SUBMIT_BLOCKED_TOKENS = ("\n", ";", "&&", "||", "|", "`", "$(", "<(", ">(", ">", "<", "&") +AUTO_SUBMIT_SIMPLE_COMMANDS = { + "cat", + "date", + "df", + "du", + "echo", + "env", + "file", + "find", + "free", + "grep", + "head", + "hostname", + "id", + "ip", + "ls", + "netstat", + "pgrep", + "printenv", + "ps", + "pwd", + "readlink", + "rg", + "ss", + "stat", + "tail", + "uname", + "uptime", + "whoami", +} +AUTO_SUBMIT_BLOCKED_COMMANDS = { + "apt", + "apt-get", + "bash", + "chmod", + "chown", + "cp", + "curl", + "dd", + "doas", + "docker", + "docker-compose", + "install", + "kubectl", + "make", + "mkdir", + "mv", + "nano", + "node", + "npm", + "perl", + "pip", + "poetry", + "python", + "python3", + "rm", + "rsync", + "scp", + "sed", + "sh", + "ssh", + "sudo", + "su", + "systemctl", + "tee", + "touch", + "vi", + "vim", + "watch", + "wget", + "zsh", +} _LAUNCH_THREAD_LOCK = threading.Lock() @@ -58,6 +134,62 @@ def _sanitize_text(text: str) -> str: return cleaned.replace("\r", "") +def _clean_terminal_text(text: str) -> str: + return CONTROL_CHAR_RE.sub("", _sanitize_text(text)) + + +def _read_tab_text(socket_path: Path, tab_id: str) -> str: + path = _tab_log_path(socket_path, tab_id) + if not path.exists(): + raise RuntimeError(f"No transcript log exists yet for tab_id: {tab_id}") + return _clean_terminal_text(path.read_text(encoding="utf-8")) + + +def _last_visible_line(text: str) -> str: + lines = text.splitlines() + if not lines: + return "" + return lines[-1] + + +def _prompt_visible(text: str, prompt_pattern: str | None = None) -> bool: + last_line = _last_visible_line(text) + if not last_line: + return False + prompt_re = re.compile(prompt_pattern) if prompt_pattern else DEFAULT_PROMPT_RE + return prompt_re.search(last_line) is not None + + +def _auto_submit_guard(command: str) -> None: + stripped = command.strip() + if not stripped: + raise RuntimeError("auto_submit requires a non-empty command") + for token in AUTO_SUBMIT_BLOCKED_TOKENS: + if token in stripped: + raise RuntimeError( + f"auto_submit only supports simple read-only commands; found blocked shell syntax {token!r}" + ) + try: + tokens = shlex.split(stripped) + except ValueError as exc: + raise RuntimeError("auto_submit only supports commands that can be parsed as a simple shell word sequence") from exc + if not tokens: + raise RuntimeError("auto_submit requires a non-empty command") + + executable = os.path.basename(tokens[0]) + if executable in AUTO_SUBMIT_BLOCKED_COMMANDS: + raise RuntimeError( + f"auto_submit is restricted to small read-only commands; {executable!r} requires manual submit" + ) + if executable not in AUTO_SUBMIT_SIMPLE_COMMANDS: + raise RuntimeError( + f"auto_submit is only enabled for a narrow set of simple read-only commands; {executable!r} is not allowed" + ) + + if executable == "tail" and any(arg in {"-f", "--follow", "-F"} for arg in tokens[1:]): + raise RuntimeError("auto_submit does not allow follow-mode commands; use manual submit instead") + + def _parse_epoch(value: str) -> float: return float(value.replace(",", ".")) @@ -320,14 +452,32 @@ def exec_command( tab_id: str, command: str, newline: bool = True, + auto_submit: bool = False, poll_interval: float = 0.1, ) -> dict[str, Any]: """Write command text into an existing tab and block until the user manually presses Enter in the GUI. - The tool blocks indefinitely until manual submission is detected, not after command completion. + By default the tool blocks indefinitely until manual submission is detected, not after command completion. + When `auto_submit=True`, it immediately sends Enter too, but only for small read-only commands. Use wait_for_command_result() to wait for the command to finish and collect output. """ - result = call_gui("exec", tab_id=tab_id, command=command, newline=newline) + if auto_submit: + _auto_submit_guard(command) + + result = call_gui("exec", tab_id=tab_id, command=command, newline=newline, auto_submit=auto_submit) + if auto_submit: + return { + "tab_id": result["tab_id"], + "written_text": result["written_text"], + "submission_id": result["submission_id"], + "requested_at": result["requested_at"], + "submitted_at": result["submitted_at"], + "submitted_manually": False, + "after_sequence": result["after_sequence"], + "current_sequence": result["current_sequence"], + "newline_ignored": False, + } + submitted = _wait_for_manual_submit( tab_id, str(result["submission_id"]), @@ -346,6 +496,28 @@ def exec_command( } +@mcp.tool() +def wait_for_command(delay_seconds: float) -> dict[str, Any]: + """Block synchronously for a fixed amount of time. + + This is a simple sleep helper for workflows driven by repeated `read_tab()` calls. + """ + if delay_seconds < 0: + raise RuntimeError("delay_seconds must be >= 0") + + started = time.monotonic() + started_at = datetime.now().astimezone().isoformat(timespec="milliseconds") + time.sleep(delay_seconds) + finished_at = datetime.now().astimezone().isoformat(timespec="milliseconds") + return { + "state": "waited", + "delay_seconds": delay_seconds, + "started_at": started_at, + "finished_at": finished_at, + "elapsed_seconds": round(time.monotonic() - started, 6), + } + + @mcp.tool() def read_tab(tab_id: str, last_n_lines: int = 200) -> dict[str, Any]: """Read the trailing scrollback text from a tab.""" @@ -422,6 +594,52 @@ def wait_for_running_command(tab_id: str, timeout: float | None = None, poll_int raise RuntimeError(f"Timed out waiting for the running command on tab_id={tab_id} to finish") +@mcp.tool() +def wait_for_prompt( + tab_id: str, + timeout: float | None = None, + poll_interval: float = 0.1, + idle_seconds: float = 0.4, + prompt_pattern: str | None = None, +) -> dict[str, Any]: + """Block until terminal output becomes idle and the trailing line looks like a shell prompt. + + This is intended for delegated interactive sessions such as SSH, where tracked command markers + are unavailable. When `prompt_pattern` is omitted, a conservative default prompt regex is used. + """ + socket_path = ensure_gui() + log_path = _tab_log_path(socket_path, tab_id) + deadline = None if timeout is None or timeout <= 0 else time.monotonic() + timeout + last_signature: tuple[int, int] | None = None + last_changed_at: float | None = None + + while deadline is None or time.monotonic() < deadline: + if not log_path.exists(): + raise RuntimeError(f"No transcript log exists yet for tab_id: {tab_id}") + + stat = log_path.stat() + signature = (int(stat.st_size), int(stat.st_mtime_ns)) + now = time.monotonic() + if signature != last_signature: + last_signature = signature + last_changed_at = now + + if last_changed_at is not None and now - last_changed_at >= idle_seconds: + text = _read_tab_text(socket_path, tab_id) + if _prompt_visible(text, prompt_pattern=prompt_pattern): + return { + "tab_id": tab_id, + "state": "prompt", + "idle_seconds": idle_seconds, + "last_line": _last_visible_line(text), + "matched_prompt_pattern": prompt_pattern or DEFAULT_PROMPT_RE.pattern, + } + + time.sleep(poll_interval) + + raise RuntimeError(f"Timed out waiting for a visible prompt on tab_id={tab_id}") + + @mcp.tool() def close_tab(tab_id: str) -> dict[str, Any]: """Close a tab in the prototype GUI."""