from __future__ import annotations import argparse import base64 import cairo from datetime import datetime import json import os import re import shlex import socketserver import threading import time import uuid from dataclasses import dataclass from pathlib import Path from typing import Any import gi gi.require_version("Gdk", "4.0") gi.require_version("Gsk", "4.0") gi.require_version("Graphene", "1.0") gi.require_version("Gtk", "4.0") gi.require_version("Vte", "3.91") from gi.repository import Gdk, Gio, GLib, Graphene, Gsk, Gtk, Vte from .control import default_socket_path DEFAULT_SCROLLBACK_LINES = 10000 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[@-_]") COMMAND_START_MARKER = "__MCP_HAL9002_CMD_START__" COMMAND_END_MARKER = "__MCP_HAL9002_CMD_END__" @dataclass class TabState: tab_id: str title: str container: Gtk.Overlay scroller: Gtk.ScrolledWindow terminal: Vte.Terminal welcome_revealer: Gtk.Revealer diagnostic_layer: Gtk.DrawingArea log_path: Path command_events_path: Path history_path: Path shell_rc_path: Path cwd: str last_command: str | None last_command_token: str | None pending_submit_id: str | None pending_submit_text: str | None pending_submit_requested_at: str | None last_manual_submit_id: str | None last_manual_submit_text: str | None last_manual_submit_at: str | None current_execution_submission_id: str | None current_execution_command: str | None current_execution_started_at: str | None current_execution_after_sequence: int | None delegated_session_submission_id: str | None delegated_session_command: str | None delegated_session_started_at: str | None delegated_session_after_sequence: int | None class TerminalWindow(Gtk.ApplicationWindow): def __init__(self, app: "TerminalApp") -> None: super().__init__(application=app) self.set_title("mcp-hal9002") self.set_default_size(1100, 720) self.set_hide_on_close(False) self._app = app self._diagnostic_config: dict[str, Any] | None = None self._stack = Gtk.Stack() self._stack.set_vexpand(True) self._stack.set_hexpand(True) self._switcher = Gtk.StackSwitcher() self._switcher.set_stack(self._stack) self._switcher.set_tooltip_text("Switch terminal tabs") self._title_label = Gtk.Label(label="MCP HAL9002") self._title_label.add_css_class("heading") self._subtitle_label = Gtk.Label(label="Remote terminal cockpit") self._subtitle_label.add_css_class("caption") self._title_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) self._title_box.set_halign(Gtk.Align.CENTER) self._title_box.append(self._title_label) self._title_box.append(self._subtitle_label) header = Gtk.HeaderBar() self._header = header header.set_show_title_buttons(True) header.set_title_widget(self._title_box) add_button = Gtk.Button.new_from_icon_name("tab-new-symbolic") add_button.add_css_class("flat") add_button.set_tooltip_text("Open a new terminal tab") add_button.connect("clicked", self._on_new_tab_clicked) header.pack_end(add_button) self.set_titlebar(header) self._root_overlay = Gtk.Overlay() self._root_overlay.set_child(self._stack) self._diagnostic_layer = Gtk.DrawingArea() self._diagnostic_layer.set_hexpand(True) self._diagnostic_layer.set_vexpand(True) self._diagnostic_layer.set_can_target(False) self._diagnostic_layer.set_draw_func(self._draw_window_diagnostic_overlay) self._root_overlay.add_overlay(self._diagnostic_layer) self._root_overlay.set_measure_overlay(self._diagnostic_layer, True) self._root_overlay.set_clip_overlay(self._diagnostic_layer, False) self._diagnostic_layer.set_visible(False) self.set_child(self._root_overlay) self.sync_header_state(tab_count=0) @property def stack(self) -> Gtk.Stack: return self._stack def sync_header_state(self, tab_count: int) -> None: if tab_count > 1: self._header.set_title_widget(self._switcher) else: self._header.set_title_widget(self._title_box) def set_diagnostic_overlay(self, config: dict[str, Any] | None) -> None: self._diagnostic_config = config self._diagnostic_layer.set_visible(config is not None) self._diagnostic_layer.queue_draw() def _draw_window_diagnostic_overlay( self, _area: Gtk.DrawingArea, ctx: Any, width: int, height: int, ) -> None: self._app.draw_diagnostic_overlay( ctx, width, height, self._diagnostic_config, ) def _on_new_tab_clicked(self, _button: Gtk.Button) -> None: self._app.create_tab(title="shell") class ControlRequestHandler(socketserver.StreamRequestHandler): def handle(self) -> None: line = self.rfile.readline() if not line: return try: payload = json.loads(line.decode("utf-8")) method = payload["method"] params = payload.get("params", {}) result = self.server.app.call_on_ui_thread(method, params) response = {"id": payload.get("id"), "ok": True, "result": result} except Exception as exc: # noqa: BLE001 response = {"id": payload.get("id") if "payload" in locals() else None, "ok": False, "error": str(exc)} self.wfile.write((json.dumps(response) + "\n").encode("utf-8")) self.wfile.flush() class ControlServer(socketserver.ThreadingUnixStreamServer): allow_reuse_address = True def __init__(self, socket_path: str, app: "TerminalApp") -> None: self.app = app super().__init__(socket_path, ControlRequestHandler) class TerminalApp(Gtk.Application): def __init__(self, socket_path: Path) -> None: super().__init__(application_id="net.enne2.McpHal9002") self.socket_path = socket_path self.log_dir = self.socket_path.parent / "mcp-hal9002-logs" self.screenshot_dir = self.socket_path.parent / "mcp-hal9002-screenshots" self.window: TerminalWindow | None = None self.tabs: dict[str, TabState] = {} self.server: ControlServer | None = None self.server_thread: threading.Thread | None = None self.connect("activate", self.on_activate) def on_activate(self, _app: Gtk.Application) -> None: if self.window is None: self.window = TerminalWindow(self) self.start_control_server() self.create_tab(title="shell") self.window.present() def start_control_server(self) -> None: self.socket_path.parent.mkdir(parents=True, exist_ok=True) self.log_dir.mkdir(parents=True, exist_ok=True) self.screenshot_dir.mkdir(parents=True, exist_ok=True) if self.socket_path.exists(): self.socket_path.unlink() self.server = ControlServer(os.fspath(self.socket_path), self) self.server_thread = threading.Thread(target=self.server.serve_forever, name="gui-control-server", daemon=True) self.server_thread.start() def do_shutdown(self) -> None: if self.server is not None: self.server.shutdown() self.server.server_close() if self.socket_path.exists(): self.socket_path.unlink() Gtk.Application.do_shutdown(self) def call_on_ui_thread(self, method: str, params: dict[str, Any]) -> Any: outcome: dict[str, Any] = {} event = threading.Event() def invoke() -> bool: try: handler = getattr(self, f"rpc_{method}") except AttributeError as exc: outcome["error"] = RuntimeError(f"Unknown method: {method}") else: try: outcome["result"] = handler(**params) except Exception as exc: # noqa: BLE001 outcome["error"] = exc event.set() return False GLib.idle_add(invoke) event.wait() if "error" in outcome: raise outcome["error"] return outcome.get("result") def _format_tab_title(self, cwd: str | None = None, command: str | None = None, title: str | None = None) -> str: normalized_title = (title or "").strip() if normalized_title and normalized_title.lower() != "shell": candidate = normalized_title elif command: try: first_token = shlex.split(command)[0] except ValueError: first_token = command.strip().split()[0] if command.strip() else "shell" candidate = os.path.basename(first_token) or first_token or "shell" elif cwd: expanded = os.path.expanduser(cwd) home = os.path.expanduser("~") if expanded == home: candidate = "Home" else: candidate = os.path.basename(expanded.rstrip(os.sep)) or expanded else: candidate = "shell" candidate = candidate.strip() or "shell" if len(candidate) > 18: return candidate[:15] + "..." return candidate def _update_tab_title(self, tab: TabState, title: str) -> None: tab.title = title if self.window is None: return page = self.window.stack.get_page(tab.container) page.set_title(title) def _dismiss_onboarding(self, tab: TabState) -> None: if tab.welcome_revealer.is_visible() or tab.welcome_revealer.get_reveal_child(): tab.welcome_revealer.set_reveal_child(False) tab.welcome_revealer.set_visible(False) def _configure_tab_diagnostic_overlay(self, tab: TabState, config: dict[str, Any] | None) -> None: tab.diagnostic_layer._diagnostic_config = config # type: ignore[attr-defined] tab.diagnostic_layer.set_visible(config is not None) tab.diagnostic_layer.queue_draw() def _annotate_screenshot_with_diagnostics( self, image_path: Path, *, width: int, height: int, label: str, bounds_in_window: dict[str, Any] | None, ) -> None: surface = cairo.ImageSurface.create_from_png(os.fspath(image_path)) ctx = cairo.Context(surface) self.draw_diagnostic_overlay( ctx, width, height, { "grid_step": 48, "bounds": {"x": 0.0, "y": 0.0, "width": width, "height": height}, "label": label, }, ) if bounds_in_window is not None: alloc_text = ( f"window alloc {int(bounds_in_window.get('x', 0))},{int(bounds_in_window.get('y', 0))} " f"{int(bounds_in_window.get('width', width))}x{int(bounds_in_window.get('height', height))}" ) ctx.set_source_rgba(0.08, 0.08, 0.08, 0.88) ctx.rectangle(12.0, max(height - 34.0, 0.0), max(len(alloc_text) * 7.0, 170.0), 22.0) ctx.fill() ctx.set_source_rgba(1.0, 1.0, 1.0, 0.98) ctx.select_font_face("Sans", cairo.FONT_SLANT_NORMAL, cairo.FONT_WEIGHT_NORMAL) ctx.set_font_size(12.0) ctx.move_to(18.0, max(height - 18.0, 12.0)) ctx.show_text(alloc_text) surface.write_to_png(os.fspath(image_path)) def draw_diagnostic_overlay( self, ctx: Any, width: int, height: int, config: dict[str, Any] | None, ) -> None: if config is None: return try: grid_step = int(config.get("grid_step", 48)) bounds = config.get("bounds") or {} label = config.get("label", "") ctx.set_source_rgba(1.0, 1.0, 1.0, 0.12) ctx.set_line_width(1.0) for x in range(grid_step, width, grid_step): ctx.move_to(x + 0.5, 0) ctx.line_to(x + 0.5, height) for y in range(grid_step, height, grid_step): ctx.move_to(0, y + 0.5) ctx.line_to(width, y + 0.5) ctx.stroke() rect_x = float(bounds.get("x", 0.0)) rect_y = float(bounds.get("y", 0.0)) rect_width = float(bounds.get("width", width)) rect_height = float(bounds.get("height", height)) ctx.set_source_rgba(0.97, 0.36, 0.2, 0.18) ctx.rectangle(rect_x, rect_y, rect_width, rect_height) ctx.fill() ctx.set_source_rgba(0.97, 0.36, 0.2, 0.95) ctx.set_line_width(2.0) ctx.rectangle(rect_x + 1.0, rect_y + 1.0, max(rect_width - 2.0, 1.0), max(rect_height - 2.0, 1.0)) ctx.stroke() if label: label_text = f"{label} {int(rect_width)}x{int(rect_height)} @ {int(rect_x)},{int(rect_y)}" text_x = rect_x + 8.0 text_y = max(rect_y - 10.0, 20.0) ctx.set_source_rgba(0.08, 0.08, 0.08, 0.88) ctx.rectangle(text_x - 6.0, text_y - 16.0, max(len(label_text) * 7.2, 120.0), 20.0) ctx.fill() ctx.set_source_rgba(1.0, 1.0, 1.0, 0.98) ctx.select_font_face("Sans", 0, 0) ctx.set_font_size(12.0) ctx.move_to(text_x, text_y) ctx.show_text(label_text) except Exception: return def create_tab(self, title: str | None = None, cwd: str | None = None, command: str | None = None) -> dict[str, Any]: if self.window is None: raise RuntimeError("Window has not been created yet") working_directory = cwd or os.path.expanduser("~") terminal = Vte.Terminal() terminal.set_scrollback_lines(DEFAULT_SCROLLBACK_LINES) terminal.set_hexpand(True) terminal.set_vexpand(True) key_controller = Gtk.EventControllerKey() key_controller.connect("key-pressed", self._on_terminal_key_pressed) terminal.add_controller(key_controller) scroller = Gtk.ScrolledWindow() scroller.set_child(terminal) tab_overlay = Gtk.Overlay() tab_overlay.set_child(scroller) diagnostic_layer = Gtk.DrawingArea() diagnostic_layer.set_hexpand(True) diagnostic_layer.set_vexpand(True) diagnostic_layer.set_can_target(False) diagnostic_layer._diagnostic_config = None # type: ignore[attr-defined] diagnostic_layer.set_draw_func(self._draw_tab_diagnostic_overlay) tab_overlay.add_overlay(diagnostic_layer) tab_overlay.set_measure_overlay(diagnostic_layer, True) tab_overlay.set_clip_overlay(diagnostic_layer, False) diagnostic_layer.set_visible(False) welcome_revealer = Gtk.Revealer() welcome_revealer.set_transition_type(Gtk.RevealerTransitionType.CROSSFADE) welcome_revealer.set_halign(Gtk.Align.CENTER) welcome_revealer.set_valign(Gtk.Align.CENTER) welcome_revealer.set_can_target(False) welcome_card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10) welcome_card.add_css_class("card") welcome_card.set_margin_top(24) welcome_card.set_margin_bottom(24) welcome_card.set_margin_start(24) welcome_card.set_margin_end(24) welcome_card.set_size_request(320, -1) welcome_title = Gtk.Label(label="Ready to control this shell") welcome_title.add_css_class("heading") welcome_title.set_wrap(True) welcome_body = Gtk.Label( label="Use MCP tools to open tabs, send commands, read output, and capture screenshots. Press any key or run a command to dismiss this panel." ) welcome_body.add_css_class("caption") welcome_body.set_wrap(True) welcome_tip = Gtk.Label(label=f"Current workspace: {working_directory}") welcome_tip.add_css_class("caption") welcome_tip.set_wrap(True) welcome_card.append(welcome_title) welcome_card.append(welcome_body) welcome_card.append(welcome_tip) welcome_revealer.set_child(welcome_card) tab_overlay.add_overlay(welcome_revealer) tab_id = str(uuid.uuid4()) tab_title = self._format_tab_title(cwd=working_directory, command=command, title=title) log_path = self.log_dir / f"{tab_id}.log" command_events_path = self.log_dir / f"{tab_id}.commands.tsv" history_path = self.log_dir / f"{tab_id}.history" shell_rc_path = self.log_dir / f"{tab_id}.bashrc" log_path.write_text("", encoding="utf-8") command_events_path.write_text("", encoding="utf-8") history_path.write_text("", encoding="utf-8") page_name = tab_id self.window.stack.add_titled(tab_overlay, page_name, tab_title) self.window.stack.set_visible_child_name(page_name) self.window.stack.set_visible_child(tab_overlay) self.window.sync_header_state(tab_count=len(self.tabs) + 1) tab = TabState( tab_id=tab_id, title=tab_title, container=tab_overlay, scroller=scroller, terminal=terminal, welcome_revealer=welcome_revealer, diagnostic_layer=diagnostic_layer, log_path=log_path, command_events_path=command_events_path, history_path=history_path, shell_rc_path=shell_rc_path, cwd=working_directory, last_command=None, last_command_token=None, pending_submit_id=None, pending_submit_text=None, pending_submit_requested_at=None, last_manual_submit_id=None, last_manual_submit_text=None, last_manual_submit_at=None, current_execution_submission_id=None, current_execution_command=None, current_execution_started_at=None, current_execution_after_sequence=None, delegated_session_submission_id=None, delegated_session_command=None, delegated_session_started_at=None, delegated_session_after_sequence=None, ) self.tabs[tab_id] = tab show_onboarding = len(self.tabs) == 1 and command is None welcome_revealer.set_visible(show_onboarding) welcome_revealer.set_reveal_child(show_onboarding) terminal.connect("child-exited", self._on_child_exited, tab_id) terminal.connect("window-title-changed", self._on_terminal_window_title_changed, tab_id) self._write_shell_rc_file(tab) self._spawn_shell(tab, cwd=working_directory) if command: GLib.timeout_add(150, self._feed_terminal_input, tab_id, command + "\n") self.window.present() return self._serialize_tab(tab) def _command_start_marker(self, token: str) -> str: return f"{COMMAND_START_MARKER}:{token}" def _command_end_marker_prefix(self, token: str) -> str: return f"{COMMAND_END_MARKER}:{token}:" def _shell_hook_rc_content(self, tab: TabState) -> str: transcript_file = shlex.quote(os.fspath(tab.log_path)) events_file = shlex.quote(os.fspath(tab.command_events_path)) history_file = shlex.quote(os.fspath(tab.history_path)) return f"""# Auto-generated by mcp-hal9002 [[ -f ~/.bashrc ]] && source ~/.bashrc export HISTFILE={history_file} export HISTSIZE=50000 export HISTFILESIZE=50000 export HISTCONTROL= shopt -s histappend cmdhist lithist export __MCP_HAL9002_TRANSCRIPT_FILE={transcript_file} export __MCP_HAL9002_EVENTS_FILE={events_file} export __MCP_HAL9002_ACTIVE_TOKEN="" export __MCP_HAL9002_ACTIVE_STARTED_AT="" export __MCP_HAL9002_ACTIVE_CWD="" export __MCP_HAL9002_LAST_HISTORY_NUM="" export __MCP_HAL9002_SEQUENCE=0 export __MCP_HAL9002_IN_HOOK=0 __mcp_hal9002_b64() {{ printf '%s' "$1" | base64 | tr -d '\\n' }} __mcp_hal9002_debug_trap() {{ [[ "${{__MCP_HAL9002_IN_HOOK:-0}}" = 1 ]] && return 0 case "$BASH_COMMAND" in __mcp_hal9002_*|history*|builtin\\ history* ) return 0 ;; esac if [[ -z "${{__MCP_HAL9002_ACTIVE_TOKEN:-}}" ]]; then __MCP_HAL9002_ACTIVE_TOKEN="${{EPOCHREALTIME:-0}}-$$-$RANDOM" __MCP_HAL9002_ACTIVE_STARTED_AT="${{EPOCHREALTIME:-0}}" __MCP_HAL9002_ACTIVE_CWD="$PWD" printf '%s\\n' "{COMMAND_START_MARKER}:${{__MCP_HAL9002_ACTIVE_TOKEN}}" >> "$__MCP_HAL9002_TRANSCRIPT_FILE" fi return 0 }} __mcp_hal9002_prompt_hook() {{ local exit_code=$? local finished_at="${{EPOCHREALTIME:-0}}" local hist_line="" local hist_num="" local command="" local seq="" __MCP_HAL9002_IN_HOOK=1 hist_line=$(HISTTIMEFORMAT= builtin history 1 2>/dev/null || true) __MCP_HAL9002_IN_HOOK=0 if [[ -n "$hist_line" && "$hist_line" =~ ^[[:space:]]*([0-9]+)[[:space:]](.*)$ ]]; then hist_num="${{BASH_REMATCH[1]}}" command="${{BASH_REMATCH[2]}}" fi if [[ -n "${{__MCP_HAL9002_ACTIVE_TOKEN:-}}" && -n "$hist_num" && "$hist_num" != "${{__MCP_HAL9002_LAST_HISTORY_NUM:-}}" ]]; then __MCP_HAL9002_LAST_HISTORY_NUM="$hist_num" __MCP_HAL9002_SEQUENCE=$(( ${{__MCP_HAL9002_SEQUENCE:-0}} + 1 )) seq="${{__MCP_HAL9002_SEQUENCE}}" printf '%s\\n' "{COMMAND_END_MARKER}:${{__MCP_HAL9002_ACTIVE_TOKEN}}:$exit_code" >> "$__MCP_HAL9002_TRANSCRIPT_FILE" printf '%s\\t%s\\t%s\\t%s\\t%s\\t%s\\t%s\\t%s\\n' \ "$seq" \ "${{__MCP_HAL9002_ACTIVE_TOKEN}}" \ "$exit_code" \ "${{__MCP_HAL9002_ACTIVE_STARTED_AT:-0}}" \ "$finished_at" \ "$(__mcp_hal9002_b64 "${{__MCP_HAL9002_ACTIVE_CWD:-$PWD}}")" \ "$(__mcp_hal9002_b64 "$PWD")" \ "$(__mcp_hal9002_b64 "$command")" >> "$__MCP_HAL9002_EVENTS_FILE" fi __MCP_HAL9002_ACTIVE_TOKEN="" __MCP_HAL9002_ACTIVE_STARTED_AT="" __MCP_HAL9002_ACTIVE_CWD="" return 0 }} trap '__mcp_hal9002_debug_trap' DEBUG PROMPT_COMMAND='__mcp_hal9002_prompt_hook' """ def _write_shell_rc_file(self, tab: TabState) -> None: tab.shell_rc_path.write_text(self._shell_hook_rc_content(tab), encoding="utf-8") def _spawn_shell(self, tab: TabState, cwd: str | None) -> None: working_directory = cwd or os.path.expanduser("~") envv = [f"{key}={value}" for key, value in os.environ.items()] quoted_log = shlex.quote(os.fspath(tab.log_path)) script_prefix = f"exec > >(tee -a {quoted_log}) 2>&1; " argv = [ "/bin/bash", "-lc", f"{script_prefix}exec /bin/bash --rcfile {shlex.quote(os.fspath(tab.shell_rc_path))} -i", ] _success, _pid = tab.terminal.spawn_sync( Vte.PtyFlags.DEFAULT, working_directory, argv, envv, GLib.SpawnFlags.DEFAULT, None, None, None, ) def _feed_terminal_input(self, tab_id: str, payload: str) -> bool: tab = self.tabs.get(tab_id) if tab is None: return False tab.terminal.paste_text(payload) return False def _looks_like_interactive_handoff(self, command: str) -> bool: try: tokens = shlex.split(command) except ValueError: tokens = command.strip().split() if not tokens: return False executable = os.path.basename(tokens[0]) if executable in {"arca", "bash", "sh", "zsh", "fish", "tmux", "screen", "nu"}: return True if executable in {"python", "python3", "ipython", "bpython", "node"}: return len(tokens) == 1 if executable not in {"ssh", "mosh"}: return False options_with_value = { "-B", "-b", "-c", "-D", "-E", "-e", "-F", "-I", "-i", "-J", "-L", "-l", "-m", "-O", "-o", "-p", "-Q", "-R", "-S", "-W", "-w", } destination_seen = False index = 1 while index < len(tokens): token = tokens[index] if not destination_seen: if token == "--": destination_seen = True index += 1 continue if token.startswith("-"): if token in options_with_value: index += 2 continue if len(token) == 2 and token[0] == "-" and token in options_with_value: index += 2 continue index += 1 continue destination_seen = True index += 1 continue return False return destination_seen def _on_terminal_key_pressed( self, controller: Gtk.EventControllerKey, keyval: int, _keycode: int, _state: Gdk.ModifierType, ) -> bool: widget = controller.get_widget() if widget is None: return False tab = next((item for item in self.tabs.values() if item.terminal == widget), None) if tab is not None: 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") 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: return terminal_title = terminal.get_window_title() or "" terminal_title = terminal_title.strip() if terminal_title: self._update_tab_title(tab, self._format_tab_title(title=terminal_title)) def _draw_tab_diagnostic_overlay( self, area: Gtk.DrawingArea, ctx: Any, width: int, height: int, ) -> None: config = getattr(area, "_diagnostic_config", None) self.draw_diagnostic_overlay(ctx, width, height, config) def _on_child_exited(self, terminal: Vte.Terminal, _status: int, tab_id: str) -> None: tab = self.tabs.get(tab_id) if tab is not None: with tab.log_path.open("a", encoding="utf-8") as handle: handle.write("\n# process exited\n") self._update_tab_title(tab, self._format_tab_title(title=f"{tab.title} exited")) def _serialize_tab(self, tab: TabState) -> dict[str, Any]: active = False if self.window is not None: visible = self.window.stack.get_visible_child() active = visible == tab.container return { "tab_id": tab.tab_id, "title": tab.title, "active": active, "rows": tab.terminal.get_row_count(), "columns": tab.terminal.get_column_count(), } def _gtype_name(self, obj: object) -> str: gtype = getattr(obj, "__gtype__", None) if gtype is not None: name = getattr(gtype, "name", None) if name: return str(name) return type(obj).__name__ def _widget_bounds(self, widget: Gtk.Widget) -> dict[str, float | int] | None: width = widget.get_width() or widget.get_allocated_width() height = widget.get_height() or widget.get_allocated_height() if width <= 0 or height <= 0: return None bounds: dict[str, float | int] = { "x": 0.0, "y": 0.0, "width": width, "height": height, } if self.window is None or widget == self.window: return bounds try: success, rect = widget.compute_bounds(self.window) except TypeError: return bounds if success: bounds["x"] = round(rect.get_x(), 2) bounds["y"] = round(rect.get_y(), 2) bounds["width"] = round(rect.get_width(), 2) bounds["height"] = round(rect.get_height(), 2) return bounds def _build_screenshot_metadata( self, *, widget: Gtk.Widget, target: str, renderer: Gsk.Renderer, surface: Gdk.Surface, path: Path, tab_id: str | None, ) -> dict[str, Any]: return { "captured_at": datetime.now().isoformat(timespec="seconds"), "target": target, "tab_id": tab_id, "path": os.fspath(path), "widget": { "type": self._gtype_name(widget), "name": widget.get_name(), "visible": widget.get_visible(), "mapped": widget.get_mapped(), "focusable": widget.get_focusable(), "scale_factor": widget.get_scale_factor(), "bounds_in_window": self._widget_bounds(widget), }, "window": { "type": self._gtype_name(self.window) if self.window is not None else None, "title": self.window.get_title() if self.window is not None else None, "bounds": self._widget_bounds(self.window) if self.window is not None else None, }, "renderer": { "type": self._gtype_name(renderer), "realized": renderer.is_realized(), }, "surface": { "type": self._gtype_name(surface), "scale": surface.get_scale(), }, } def _require_tab(self, tab_id: str) -> TabState: try: return self.tabs[tab_id] except KeyError as exc: raise RuntimeError(f"Unknown tab_id: {tab_id}") from exc def _sanitize_text(self, text: str) -> str: cleaned = OSC_ESCAPE_RE.sub("", text) cleaned = ANSI_ESCAPE_RE.sub("", cleaned) cleaned = SINGLE_ESCAPE_RE.sub("", cleaned) cleaned = cleaned.replace("\r", "") return cleaned def _parse_epoch(self, value: str) -> float: return float(value.replace(",", ".")) def _parse_command_events(self, tab: TabState) -> list[dict[str, Any]]: if not tab.command_events_path.exists(): return [] events: list[dict[str, Any]] = [] for raw_line in tab.command_events_path.read_text(encoding="utf-8").splitlines(): if not raw_line.strip(): continue parts = raw_line.split("\t") if len(parts) != 8: continue try: started_epoch = self._parse_epoch(parts[3]) finished_epoch = self._parse_epoch(parts[4]) event = { "sequence": int(parts[0]), "token": parts[1], "exit_code": int(parts[2]), "started_epoch": started_epoch, "finished_epoch": finished_epoch, "started_at": datetime.fromtimestamp(started_epoch).astimezone().isoformat(timespec="seconds"), "finished_at": datetime.fromtimestamp(finished_epoch).astimezone().isoformat(timespec="seconds"), "duration_seconds": round(max(finished_epoch - started_epoch, 0.0), 6), "cwd": base64.b64decode(parts[5]).decode("utf-8"), "cwd_after": base64.b64decode(parts[6]).decode("utf-8"), "command": base64.b64decode(parts[7]).decode("utf-8"), } except Exception: continue events.append(event) return events def _latest_command_event(self, tab: TabState) -> dict[str, Any] | None: events = self._parse_command_events(tab) return events[-1] if events else None def _latest_command_sequence(self, tab: TabState) -> int: latest = self._latest_command_event(tab) return int(latest["sequence"]) if latest is not None else 0 def _manual_submit_status(self, tab: TabState, submission_id: str) -> dict[str, Any]: if tab.last_manual_submit_id == submission_id: return { "tab_id": tab.tab_id, "submission_id": submission_id, "submitted": True, "submitted_at": tab.last_manual_submit_at, "written_text": tab.last_manual_submit_text, "current_sequence": self._latest_command_sequence(tab), } if tab.pending_submit_id == submission_id: return { "tab_id": tab.tab_id, "submission_id": submission_id, "submitted": False, "requested_at": tab.pending_submit_requested_at, "written_text": tab.pending_submit_text, "current_sequence": self._latest_command_sequence(tab), } raise RuntimeError(f"Unknown manual submission id for tab_id={tab.tab_id}: {submission_id}") def _clear_current_execution(self, tab: TabState) -> None: tab.current_execution_submission_id = None tab.current_execution_command = None tab.current_execution_started_at = None tab.current_execution_after_sequence = None def _clear_delegated_session(self, tab: TabState) -> None: tab.delegated_session_submission_id = None tab.delegated_session_command = None tab.delegated_session_started_at = None tab.delegated_session_after_sequence = None def _delegated_session_status(self, tab: TabState) -> dict[str, Any]: if tab.delegated_session_submission_id is None: return { "tab_id": tab.tab_id, "state": "idle", "current_sequence": self._latest_command_sequence(tab), } after_sequence = int(tab.delegated_session_after_sequence or 0) current_sequence = self._latest_command_sequence(tab) if current_sequence > after_sequence: self._clear_delegated_session(tab) return { "tab_id": tab.tab_id, "state": "idle", "current_sequence": current_sequence, } return { "tab_id": tab.tab_id, "state": "interactive-session", "submission_id": tab.delegated_session_submission_id, "command": tab.delegated_session_command, "started_at": tab.delegated_session_started_at, "after_sequence": after_sequence, "current_sequence": current_sequence, } def _running_command_status(self, tab: TabState) -> dict[str, Any]: delegated_status = self._delegated_session_status(tab) if delegated_status["state"] == "interactive-session": return delegated_status if tab.current_execution_submission_id is None: return { "tab_id": tab.tab_id, "state": "idle", "current_sequence": self._latest_command_sequence(tab), } after_sequence = int(tab.current_execution_after_sequence or 0) current_sequence = self._latest_command_sequence(tab) status = { "tab_id": tab.tab_id, "submission_id": tab.current_execution_submission_id, "command": tab.current_execution_command, "started_at": tab.current_execution_started_at, "after_sequence": after_sequence, "current_sequence": current_sequence, } if current_sequence > after_sequence: latest = self._latest_command_event(tab) status["state"] = "completed" if latest is not None: status["sequence"] = latest["sequence"] status["finished_at"] = latest["finished_at"] status["exit_code"] = latest["exit_code"] return status status["state"] = "running" return status def _default_screenshot_path(self, target: str, tab_id: str | None = None) -> Path: timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") suffix = tab_id if tab_id else target return self.screenshot_dir / f"{timestamp}-{suffix}.png" def _extract_command_result(self, tab: TabState, token: str) -> dict[str, Any]: lines = self._sanitize_text(tab.log_path.read_text(encoding="utf-8")).splitlines() start_marker = self._command_start_marker(token) end_prefix = self._command_end_marker_prefix(token) start_index: int | None = None end_index: int | None = None exit_code: int | None = None for index, line in enumerate(lines): if line == start_marker: start_index = index end_index = None exit_code = None continue if start_index is not None and line.startswith(end_prefix): end_index = index try: exit_code = int(line[len(end_prefix):]) except ValueError: exit_code = None if start_index is None: raise RuntimeError("No tracked command result is available for this tab yet") if end_index is None: raise RuntimeError("The last tracked command is still running or has not produced a completion marker yet") text = "\n".join(lines[start_index + 1:end_index]).strip("\n") latest = next((event for event in reversed(self._parse_command_events(tab)) if event["token"] == token), None) if latest is None: raise RuntimeError("Tracked command metadata is not available for this tab yet") tab.last_command = str(latest["command"]) tab.last_command_token = token return { "tab_id": tab.tab_id, "sequence": latest["sequence"], "command": latest["command"], "cwd": latest["cwd"], "cwd_after": latest["cwd_after"], "started_at": latest["started_at"], "finished_at": latest["finished_at"], "started_epoch": latest["started_epoch"], "finished_epoch": latest["finished_epoch"], "duration_seconds": latest["duration_seconds"], "exit_code": exit_code, "text": text, } def _capture_widget_to_png(self, widget: Gtk.Widget, path: Path) -> dict[str, Any]: paintable = Gtk.WidgetPaintable.new(widget) width = widget.get_width() or widget.get_allocated_width() height = widget.get_height() or widget.get_allocated_height() if width <= 0 or height <= 0: raise RuntimeError("The widget is not ready for capture yet; try again after the window has been presented") native = widget.get_native() if native is None: raise RuntimeError("The widget is not attached to a native surface yet; try again after the window has been presented") surface = native.get_surface() if surface is None: raise RuntimeError("The widget does not have an associated surface yet; try again after the window has been presented") node = None texture = None renderer = None viewport = Graphene.Rect() viewport.init(0, 0, float(width), float(height)) main_context = GLib.MainContext.default() last_error = "Failed to snapshot the widget into a render node" for attempt in range(4): while main_context.pending(): main_context.iteration(False) snapshot = Gtk.Snapshot.new() paintable.snapshot(snapshot, float(width), float(height)) node = snapshot.to_node() if node is None: last_error = "Failed to snapshot the widget into a render node" if attempt < 3: time.sleep(0.03) continue raise RuntimeError(last_error) renderer = Gsk.Renderer.new_for_surface(surface) if renderer is None: raise RuntimeError("Failed to create a GSK renderer for the widget surface") if not renderer.is_realized() and not renderer.realize(surface): raise RuntimeError("Failed to realize the GSK renderer for screenshot capture") texture = renderer.render_texture(node, viewport) if texture is not None: break renderer.unrealize() renderer = None last_error = "Failed to render the widget snapshot into a texture" if attempt < 3: time.sleep(0.03) continue raise RuntimeError(last_error) if renderer is None or texture is None: raise RuntimeError(last_error) path.parent.mkdir(parents=True, exist_ok=True) saved = texture.save_to_png(os.fspath(path)) metadata = self._build_screenshot_metadata( widget=widget, target="", renderer=renderer, surface=surface, path=path, tab_id=None, ) renderer.unrealize() if not saved: raise RuntimeError(f"Failed to save screenshot to {path}") metadata_path = path.with_suffix(".json") metadata_path.write_text(json.dumps(metadata, indent=2), encoding="utf-8") return { "path": os.fspath(path), "metadata_path": os.fspath(metadata_path), "metadata": metadata, "width": width, "height": height, } def rpc_ping(self) -> dict[str, str]: return {"status": "ok"} def _visible_tab(self) -> TabState | None: if self.window is None: return None visible = self.window.stack.get_visible_child() if visible is None: return None return next((item for item in self.tabs.values() if item.container == visible), None) def _quit_application(self) -> bool: self.quit() return False def rpc_gui_status(self) -> dict[str, Any]: visible_tab = self._visible_tab() return { "tab_count": len(self.tabs), "active_tab_id": visible_tab.tab_id if visible_tab is not None else None, "tab_ids": list(self.tabs.keys()), "window_title": self.window.get_title() if self.window is not None else None, } def rpc_show_window(self) -> dict[str, Any]: if self.window is None: raise RuntimeError("Window has not been created yet") self.window.present() return self.rpc_gui_status() def rpc_close_window(self) -> dict[str, Any]: status = self.rpc_gui_status() GLib.idle_add(self._quit_application) status["closed"] = True return status def rpc_list_tabs(self) -> list[dict[str, Any]]: return [self._serialize_tab(tab) for tab in self.tabs.values()] def rpc_open_tab(self, title: str | None = None, cwd: str | None = None, command: str | None = None) -> dict[str, Any]: return self.create_tab(title=title, cwd=cwd, command=command) def rpc_focus_tab(self, tab_id: str) -> dict[str, Any]: tab = self._require_tab(tab_id) if self.window is None: raise RuntimeError("Window has not been created yet") self.window.stack.set_visible_child(tab.container) self.window.present() return self._serialize_tab(tab) 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": raise RuntimeError( f"tab_id={tab_id} already has a command in progress; use wait_for_running_command before sending another command" ) if running_status["state"] == "completed": self._clear_current_execution(tab) if running_status["state"] == "interactive-session": self._clear_current_execution(tab) if tab.pending_submit_id is not None: raise RuntimeError( f"tab_id={tab_id} already has a pending manual submit request; press Enter in the GUI before writing another command" ) submission_id = str(uuid.uuid4()) requested_at = datetime.now().astimezone().isoformat(timespec="milliseconds") payload = command self._dismiss_onboarding(tab) if self.window is not None: self.window.stack.set_visible_child(tab.container) self.window.present() tab.terminal.grab_focus() 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, "written_text": command, "awaiting_manual_submit": True, "after_sequence": self._latest_command_sequence(tab), "submission_id": submission_id, "requested_at": requested_at, "newline_ignored": newline, } def rpc_manual_submit_status(self, tab_id: str, submission_id: str) -> dict[str, Any]: tab = self._require_tab(tab_id) return self._manual_submit_status(tab, submission_id) def rpc_running_command_status(self, tab_id: str) -> dict[str, Any]: tab = self._require_tab(tab_id) return self._running_command_status(tab) def rpc_read_tab(self, tab_id: str, last_n_lines: int = 200) -> dict[str, Any]: tab = self._require_tab(tab_id) lines = tab.log_path.read_text(encoding="utf-8").splitlines() filtered_lines = [ line for line in lines if not line.startswith(f"{COMMAND_START_MARKER}:") and not line.startswith(f"{COMMAND_END_MARKER}:") ] tail_lines = filtered_lines[-last_n_lines:] text = self._sanitize_text("\n".join(tail_lines)) return { "tab_id": tab_id, "line_count": len(filtered_lines), "text": text, } def rpc_read_last_command_result(self, tab_id: str) -> dict[str, Any]: tab = self._require_tab(tab_id) latest = self._latest_command_event(tab) if latest is None: raise RuntimeError("No tracked command result is available for this tab yet") return self._extract_command_result(tab, str(latest["token"])) def rpc_capture_screenshot( self, target: str = "window", tab_id: str | None = None, path: str | None = None, diagnostic_overlay: bool = False, ) -> dict[str, Any]: if self.window is None: raise RuntimeError("Window has not been created yet") target_widget: Gtk.Widget selected_tab_id: str | None = tab_id normalized_target = target.strip().lower() if normalized_target == "window": self.window.present() target_widget = self.window overlay_scope = "window" elif normalized_target in {"active-tab", "active_tab", "tab", "terminal", "vte"}: if selected_tab_id is None: visible = self.window.stack.get_visible_child() if visible is None: raise RuntimeError("No active tab is available") tab = next((item for item in self.tabs.values() if item.container == visible), None) if tab is None: raise RuntimeError("Unable to resolve the active tab") else: tab = self._require_tab(selected_tab_id) self.window.stack.set_visible_child(tab.container) self.window.present() target_widget = tab.terminal selected_tab_id = tab.tab_id normalized_target = "tab" overlay_scope = "terminal" elif normalized_target in {"tab-container", "tab_container", "scroller"}: if selected_tab_id is None: visible = self.window.stack.get_visible_child() if visible is None: raise RuntimeError("No active tab is available") tab = next((item for item in self.tabs.values() if item.container == visible), None) if tab is None: raise RuntimeError("Unable to resolve the active tab") else: tab = self._require_tab(selected_tab_id) self.window.stack.set_visible_child(tab.container) self.window.present() target_widget = tab.container selected_tab_id = tab.tab_id normalized_target = "tab-container" overlay_scope = "tab-container" else: raise RuntimeError("target must be one of: window, tab, tab-container") screenshot_path = Path(path) if path else self._default_screenshot_path(normalized_target, tab_id=selected_tab_id) try: result = self._capture_widget_to_png(target_widget, screenshot_path) except RuntimeError: if normalized_target != "window": raise fallback_widget = self.window.get_child() if fallback_widget is None or fallback_widget is target_widget: raise target_widget = fallback_widget result = self._capture_widget_to_png(target_widget, screenshot_path) result["metadata"]["target"] = normalized_target result["metadata"]["tab_id"] = selected_tab_id overlay_applied = False if diagnostic_overlay: overlay_label = f"{normalized_target} {self._gtype_name(target_widget)}" self._annotate_screenshot_with_diagnostics( screenshot_path, width=result["width"], height=result["height"], label=overlay_label, bounds_in_window=result["metadata"]["widget"].get("bounds_in_window"), ) overlay_applied = True result["metadata"]["diagnostic_overlay"] = { "requested": diagnostic_overlay, "applied": overlay_applied, "scope": overlay_scope, "mode": "postprocess-cairo", } Path(result["metadata_path"]).write_text(json.dumps(result["metadata"], indent=2), encoding="utf-8") result["target"] = normalized_target if selected_tab_id is not None: result["tab_id"] = selected_tab_id return result def rpc_close_tab(self, tab_id: str) -> dict[str, Any]: tab = self._require_tab(tab_id) if self.window is None: raise RuntimeError("Window has not been created yet") self.window.stack.remove(tab.container) del self.tabs[tab_id] if tab.log_path.exists(): tab.log_path.unlink() if not self.tabs: self.create_tab(title="shell") elif self.window is not None: self.window.sync_header_state(tab_count=len(self.tabs)) return {"closed": tab_id} def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="mcp-hal9002 GTK4/VTE terminal workspace controllable through a local Unix socket") parser.add_argument("--socket", type=Path, default=default_socket_path(), help="Unix socket path for the local control plane") return parser def main() -> None: parser = build_parser() args = parser.parse_args() app = TerminalApp(socket_path=args.socket) app.run([]) if __name__ == "__main__": main()