from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone import json import os from pathlib import Path, PurePosixPath import re import shlex import shutil import subprocess import sys import time from typing import Any, Callable from urllib.parse import quote from .config import Config, DeviceConfig from .runner import ( CommandResult, remote_command, run, scp_from_argv, scp_to_argv, ssh_argv, truncate, user_bus_env, ) from .vendor import build_sailfishos ToolHandler = Callable[[dict[str, Any]], dict[str, Any]] @dataclass(frozen=True) class Tool: spec: dict[str, Any] handler: ToolHandler def build_registry(config: Config) -> dict[str, Tool]: tools = [ Tool(_spec_devices(), lambda args: handle_devices(config, args)), Tool(_spec_device_journal(), lambda args: handle_device_journal(config, args)), Tool(_spec_device_topmost_pid(), lambda args: handle_device_topmost_pid(config, args)), Tool(_spec_device_proc_maps(), lambda args: handle_device_proc_maps(config, args)), Tool( _spec_device_lipstick_screenshot(), lambda args: handle_device_lipstick_screenshot(config, args), ), Tool( _spec_device_touch(), lambda args: handle_device_touch(config, args), ), Tool( _spec_device_touch_workflow(), lambda args: handle_device_touch_workflow(config, args), ), Tool( _spec_device_user_bus_call(), lambda args: handle_device_user_bus_call(config, args), ), Tool( _spec_device_user_session_command(), lambda args: handle_device_user_session_command(config, args), ), Tool(_spec_device_install_rpm(), lambda args: handle_device_install_rpm(config, args)), Tool( _spec_device_restart_service(), lambda args: handle_device_restart_service(config, args), ), Tool( _spec_device_browser_launch(), lambda args: handle_device_browser_launch(config, args), ), Tool(_spec_build_rpm(), lambda args: handle_build_rpm(config, args)), Tool(_spec_build_status(), lambda args: handle_build_status(config, args)), Tool( _spec_sdk_refresh_metadata(), lambda args: handle_sdk_refresh_metadata(config, args), ), Tool(_spec_obs_results(), lambda args: handle_obs_results(config, args)), Tool(_spec_obs_buildlog(), lambda args: handle_obs_buildlog(config, args)), Tool(_spec_repo_status(), lambda args: handle_repo_status(config, args)), Tool(_spec_repo_find(), lambda args: handle_repo_find(config, args)), Tool(_spec_spec_summary(), lambda args: handle_spec_summary(config, args)), Tool( _spec_qml_find_translations(), lambda args: handle_qml_find_translations(config, args), ), Tool( _spec_qml_check_translator_ternaries(), lambda args: handle_qml_check_translator_ternaries(config, args), ), ] return {tool.spec["name"]: tool for tool in tools} def handle_devices(config: Config, args: dict[str, Any]) -> dict[str, Any]: structured = config.public_dict() text = "\n".join( ( f"{name}: {device.ssh_target}" f" user={device.username}" f"{f' arch={device.architecture}' if device.architecture else ''}" f"{f' release={device.release}' if device.release else ''}" ) for name, device in sorted(config.devices.items(), key=lambda item: item[0]) ) return ok_text(text or "No configured devices", structured) def handle_device_journal(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) lines = _int_arg(args, "lines", default=200, minimum=1, maximum=5000) timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=600) command = ["journalctl", "--no-pager", "-n", str(lines)] unit = _optional_str(args, "unit") since = _optional_str(args, "since") grep = _optional_str(args, "grep") if unit: command += ["-u", unit] if since: command += ["--since", since] result = _run_ssh(config, device, command, timeout=timeout) stdout = result.stdout if grep: stdout = "\n".join(line for line in stdout.splitlines() if grep in line) result = CommandResult(result.argv, result.returncode, stdout, result.stderr) return command_result("device journal", result) def handle_device_topmost_pid(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) timeout = _int_arg(args, "timeout", default=20, minimum=1, maximum=120) command = user_bus_env(device) + [ "dbus-send", "--session", "--print-reply", "--dest=org.nemomobile.lipstick", "/", "org.freedesktop.DBus.Properties.Get", "string:org.nemomobile.compositor", "string:privateTopmostWindowProcessId", ] result = _run_ssh(config, device, command, timeout=timeout) pid = _parse_dbus_integer(result.stdout) structured = result.public_dict() structured["pid"] = pid text = f"topmost PID: {pid}" if pid is not None else _command_text("topmost PID", result) return { "content": [{"type": "text", "text": text}], "structuredContent": structured, "isError": not result.ok or pid is None, } def handle_device_proc_maps(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) pid = _int_arg(args, "pid", minimum=1) contains = _optional_str(args, "contains") max_lines = _int_arg(args, "max_lines", default=200, minimum=1, maximum=5000) timeout = _int_arg(args, "timeout", default=20, minimum=1, maximum=120) result = _run_ssh(config, device, ["cat", f"/proc/{pid}/maps"], timeout=timeout) lines = result.stdout.splitlines() if contains: lines = [line for line in lines if contains in line] limited = lines[:max_lines] stdout = "\n".join(limited) if len(lines) > max_lines: stdout += f"\n[truncated after {max_lines} lines]" filtered = CommandResult(result.argv, result.returncode, stdout, result.stderr) return command_result("process maps", filtered, {"matched_lines": len(lines), "pid": pid}) def handle_device_lipstick_screenshot(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") home_path = _device_home_path(device) remote_path = _optional_str(args, "remote_path") or ( f"{home_path}/Pictures/Screenshots/lipstick-{timestamp}.png" ) local_path_arg = _optional_str(args, "local_path") privileged = _bool_arg(args, "privileged", default=True) timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=120) inner = remote_command( user_bus_env(device) + [ "dbus-send", "--session", "--print-reply", "--dest=org.nemomobile.lipstick", "/org/nemomobile/lipstick/screenshot", "org.nemomobile.lipstick.saveScreenshot", f"string:{remote_path}", ] ) prepare_dir = _screenshot_prepare_command(device, remote_path) dbus_call = f"sg privileged -c {shlex.quote(inner)}" if privileged else inner remote = f"{prepare_dir} && {dbus_call}" result = run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout) structured: dict[str, Any] = result.public_dict() structured["remote_path"] = remote_path if result.ok and local_path_arg: local_path = _safe_output_path(config, local_path_arg) local_path.parent.mkdir(parents=True, exist_ok=True) pull = run( scp_from_argv(device, config.paths.ssh_config, remote_path, local_path), timeout=timeout, ) structured["pull"] = pull.public_dict() structured["local_path"] = str(local_path) if not pull.ok: return command_result("pull screenshot", pull, structured) return command_result("lipstick screenshot", result, structured) def handle_device_touch(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) action = _enum_arg(args, "action", ["discover", "tap", "swipe"]) timeout = _int_arg(args, "timeout", default=10, minimum=1, maximum=120) input_device = _optional_str(args, "input_device") if input_device and not re.fullmatch(r"/dev/input/event\d+", input_device): return tool_error("input_device must look like /dev/input/event") if action == "discover": include_evdev_trace = _bool_arg(args, "include_evdev_trace", default=False) remote = _touch_discover_command(include_evdev_trace) return command_result( "touchscreen discovery", run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout), ) values: dict[str, str | int] = {"ACTION": action} if input_device: values["INPUT_DEVICE"] = input_device if action == "tap": values["X"] = _int_arg(args, "x", minimum=0, maximum=10000) values["Y"] = _int_arg(args, "y", minimum=0, maximum=10000) values["HOLD_MS"] = _int_arg(args, "hold_ms", default=80, minimum=1, maximum=5000) else: values["START_X"] = _int_arg(args, "start_x", minimum=0, maximum=10000) values["START_Y"] = _int_arg(args, "start_y", minimum=0, maximum=10000) values["END_X"] = _int_arg(args, "end_x", minimum=0, maximum=10000) values["END_Y"] = _int_arg(args, "end_y", minimum=0, maximum=10000) values["DURATION_MS"] = _int_arg(args, "duration_ms", default=300, minimum=1, maximum=10000) values["STEPS"] = _int_arg(args, "steps", default=12, minimum=1, maximum=200) remote = _touch_inject_command(values) structured = { "action": action, "input_device": input_device, } return command_result( "device touch", run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout), structured, ) def handle_device_touch_workflow(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) action = _enum_arg(args, "action", ["tap", "swipe"]) timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) screenshot_before = _bool_arg(args, "screenshot_before", default=True) screenshot_after = _bool_arg(args, "screenshot_after", default=False) discover_input = _bool_arg(args, "discover_input", default=True) privileged = _bool_arg(args, "privileged", default=True) steps: list[tuple[str, dict[str, Any]]] = [] base_args = {"device": _optional_str(args, "device") or device.name, "timeout": timeout} def append_step(name: str, result: dict[str, Any]) -> bool: steps.append((name, result)) return bool(result.get("isError", False)) if screenshot_before: before_args: dict[str, Any] = { **base_args, "privileged": privileged, "remote_path": _optional_str(args, "before_remote_path") or _default_screenshot_path(device, "touch-before"), } before_local_path = _optional_str(args, "before_local_path") or _optional_str(args, "local_path") if before_local_path: before_args["local_path"] = before_local_path if append_step("screenshot_before", handle_device_lipstick_screenshot(config, before_args)): return combined_result("touch workflow", steps) input_device = _optional_str(args, "input_device") if discover_input and not input_device: discover_args = { **base_args, "action": "discover", "include_evdev_trace": _bool_arg(args, "include_evdev_trace", default=False), } if append_step("touchscreen_discovery", handle_device_touch(config, discover_args)): return combined_result("touch workflow", steps) touch_args = dict(args) touch_args["action"] = action touch_args["timeout"] = timeout if append_step("touch", handle_device_touch(config, touch_args)): return combined_result("touch workflow", steps) if screenshot_after: after_args: dict[str, Any] = { **base_args, "privileged": privileged, "remote_path": _optional_str(args, "after_remote_path") or _default_screenshot_path(device, "touch-after"), } after_local_path = _optional_str(args, "after_local_path") if after_local_path: after_args["local_path"] = after_local_path append_step("screenshot_after", handle_device_lipstick_screenshot(config, after_args)) return combined_result("touch workflow", steps) def handle_device_user_bus_call(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) destination = _str_arg(args, "destination") path = _str_arg(args, "path") interface = _str_arg(args, "interface") member = _str_arg(args, "member") dbus_args = args.get("arguments") or [] if not isinstance(dbus_args, list) or not all(isinstance(item, str) for item in dbus_args): return tool_error("arguments must be a list of dbus-send argument strings") timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) command = user_bus_env(device) + [ "dbus-send", "--session", "--print-reply", f"--dest={destination}", path, f"{interface}.{member}", *dbus_args, ] return command_result("user bus call", _run_ssh(config, device, command, timeout=timeout)) def handle_device_user_session_command(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) command = _command_list_arg(args, "command") timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) run_as_user = _bool_arg(args, "run_as_user", default=False) session_command = user_bus_env(device) + command if run_as_user: remote = _run_as_user_command(device, remote_command(session_command)) result = run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout) else: result = _run_ssh(config, device, session_command, timeout=timeout) return command_result( "user session command", result, {"run_as_user": run_as_user, "username": device.username}, ) def handle_device_install_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) rpm_path = _safe_input_path(config, _str_arg(args, "rpm_path"), allow_tmp=True) remote_path = _optional_str(args, "remote_path") or f"/tmp/{rpm_path.name}" installer = _enum_arg(args, "installer", ["pkcon", "rpm"], default="pkcon") timeout = _int_arg(args, "timeout", default=180, minimum=1, maximum=1200) copy_result = run( scp_to_argv(device, config.paths.ssh_config, rpm_path, remote_path), timeout=timeout, ) if not copy_result.ok: return command_result("copy RPM to device", copy_result) if installer == "pkcon": remote = ["pkcon", "install-local", "-y", remote_path] else: remote = ["rpm", "-Uvh", "--replacepkgs", remote_path] install_result = _run_ssh(config, device, remote, timeout=timeout) structured = { "copy": copy_result.public_dict(), "install": install_result.public_dict(), "remote_path": remote_path, "installer": installer, } return command_result("install RPM", install_result, structured) def handle_device_restart_service(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) unit = _service_unit_arg(args, "unit") action = _enum_arg(args, "action", ["restart", "start", "stop", "status"], default="restart") mode = _enum_arg(args, "mode", ["system", "user"], default="system") timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=300) if mode == "user": command = user_bus_env(device) + ["systemctl", "--user", action, unit] else: command = ["systemctl", action, unit] return command_result( f"{mode} service {action}", _run_ssh(config, device, command, timeout=timeout), ) def handle_device_browser_launch(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) url = _str_arg(args, "url") stop_stale = _bool_arg(args, "stop_stale", default=True) wait_seconds = _int_arg(args, "wait_seconds", default=3, minimum=0, maximum=60) timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=300) remote = _browser_launch_command(device, url, stop_stale, wait_seconds) result = run(ssh_argv(device, config.paths.ssh_config, remote), timeout=timeout) return command_result( "browser launch", result, {"url": url, "stop_stale": stop_stale, "wait_seconds": wait_seconds}, ) def handle_build_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: project_path = _safe_input_path(config, _str_arg(args, "project_path"), allow_tmp=False) device = config.device(_optional_str(args, "device")) if args.get("device") else None script = config.paths.build_sailfishos if not script.exists(): return tool_error(f"build helper not found: {script}") command = ["python3", str(script), "--project-dir", str(project_path)] release = _optional_str(args, "release") or (device.release if device else None) arches = args.get("arch") artifacts_dir = _optional_str(args, "artifacts_dir") if config.paths.local_sdk: command += ["--local-sdk", str(config.paths.local_sdk)] if not release: release = "live" if release: command += ["--release", release] if isinstance(arches, str): command += ["--arch", arches] elif isinstance(arches, list): for arch in arches: if not isinstance(arch, str): return tool_error("arch must be a string or list of strings") command += ["--arch", arch] elif arches is not None: return tool_error("arch must be a string or list of strings") elif device and device.architecture: command += ["--arch", device.architecture] if artifacts_dir: output = _safe_output_path(config, artifacts_dir) command += ["--artifacts-dir", str(output)] if _bool_arg(args, "all_arches", default=False): command.append("--all") if _bool_arg(args, "clean", default=False): command.append("--clean") if _bool_arg(args, "debug", default=False): command.append("--debug") if _bool_arg(args, "no_pull", default=False): command.append("--no-pull") for local_dir in _string_list_arg(args, "local_rpms_dir"): command += ["--local-rpms-dir", str(_safe_input_path(config, local_dir, allow_tmp=True))] timeout = _int_arg(args, "timeout", default=3600, minimum=1, maximum=21600) if _bool_arg(args, "wait", default=False): return command_result("build Sailfish RPM", run(command, timeout=timeout)) job = _start_background_command("build Sailfish RPM", command, timeout=timeout) text = "\n".join( [ f"started build job {job['job_id']}", f"status: {job['status_path']}", f"log: {job['log_path']}", "poll with sailfish_build_status", ] ) return ok_text(text, job) def handle_build_status(config: Config, args: dict[str, Any]) -> dict[str, Any]: job_id = _optional_str(args, "job_id") lines = _int_arg(args, "lines", default=80, minimum=0, maximum=1000) jobs_dir = _build_jobs_dir() if not job_id: jobs = [] if jobs_dir.exists(): for status_path in sorted(jobs_dir.glob("*/status.json"), key=lambda p: p.stat().st_mtime): status = _read_job_status(status_path) if status: jobs.append(status) jobs = jobs[-20:] text = "\n".join( f"{job.get('job_id')}: {job.get('state')} {job.get('started_at', '')}" for job in jobs ) return ok_text(text or "no build jobs found", {"jobs_dir": str(jobs_dir), "jobs": jobs}) job_dir = jobs_dir / job_id status_path = job_dir / "status.json" if not status_path.exists(): return tool_error(f"unknown build job: {job_id}", {"jobs_dir": str(jobs_dir)}) status = _read_job_status(status_path) if not status: return tool_error(f"could not read build job status: {job_id}") log_path = Path(str(status.get("log_path") or job_dir / "build.log")) log_tail = _tail_file(log_path, lines) state = str(status.get("state") or "unknown") returncode = status.get("returncode") text_lines = [ f"job_id: {status.get('job_id')}", f"state: {state}", f"returncode: {returncode}", f"log: {log_path}", ] if log_tail: text_lines += ["", log_tail] structured = dict(status) structured["log_tail"] = log_tail return { "content": [{"type": "text", "text": "\n".join(text_lines)}], "structuredContent": structured, "isError": state == "finished" and returncode not in (0, None), } def _mcp_state_dir() -> Path: value = os.environ.get("SAILFISH_DEVEL_MCP_STATE_DIR") if value: return Path(value).expanduser() value = os.environ.get("SAILFISH_DEVEL_MCP_LOG_DIR") if value: return Path(value).expanduser() value = os.environ.get("XDG_STATE_HOME") if value: return Path(value).expanduser() / "sailfish-devel-mcp" value = os.environ.get("HOME") if value: return Path(value).expanduser() / ".local" / "state" / "sailfish-devel-mcp" return Path("/tmp") / "sailfish-devel-mcp" def _build_jobs_dir() -> Path: return _mcp_state_dir() / "builds" def _write_json_atomic(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") tmp.replace(path) def _start_background_command(label: str, command: list[str], *, timeout: int) -> dict[str, Any]: now = datetime.now(timezone.utc) job_id = f"build-{now.strftime('%Y%m%dT%H%M%S')}-{os.getpid()}-{int(time.time() * 1000) % 100000}" job_dir = _build_jobs_dir() / job_id log_path = job_dir / "build.log" status_path = job_dir / "status.json" status = { "job_id": job_id, "label": label, "state": "starting", "argv": command, "timeout": timeout, "created_at": now.isoformat(), "status_path": str(status_path), "log_path": str(log_path), } _write_json_atomic(status_path, status) supervisor = r""" from __future__ import annotations from datetime import datetime, timezone import json import os from pathlib import Path import signal import subprocess import sys import time def now() -> str: return datetime.now(timezone.utc).isoformat() def write_status(path: Path, data: dict[str, object]) -> None: tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") tmp.replace(path) status_path = Path(sys.argv[1]) log_path = Path(sys.argv[2]) timeout = int(sys.argv[3]) command = sys.argv[4:] status = json.loads(status_path.read_text(encoding="utf-8")) status["supervisor_pid"] = os.getpid() status["started_at"] = now() log_path.parent.mkdir(parents=True, exist_ok=True) with log_path.open("a", encoding="utf-8", errors="replace") as log: log.write(f"[{now()}] starting {' '.join(command)}\n") log.flush() process = subprocess.Popen( command, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, text=True, start_new_session=True, close_fds=True, ) status["pid"] = process.pid status["state"] = "running" write_status(status_path, status) deadline = time.monotonic() + timeout timed_out = False returncode = None while True: returncode = process.poll() if returncode is not None: break if time.monotonic() >= deadline: timed_out = True log.write(f"[{now()}] timeout after {timeout}s; terminating process group {process.pid}\n") log.flush() try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: pass try: returncode = process.wait(timeout=30) except subprocess.TimeoutExpired: log.write(f"[{now()}] process group did not exit; killing {process.pid}\n") log.flush() try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass returncode = process.wait() break time.sleep(1) status["state"] = "finished" status["returncode"] = returncode status["timed_out"] = timed_out status["finished_at"] = now() write_status(status_path, status) log.write(f"[{now()}] finished returncode={returncode} timed_out={timed_out}\n") """ process = subprocess.Popen( [sys.executable, "-c", supervisor, str(status_path), str(log_path), str(timeout), *command], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, close_fds=True, ) status["supervisor_pid"] = process.pid _write_json_atomic(status_path, status) return status def _read_job_status(path: Path) -> dict[str, Any] | None: try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None return data if isinstance(data, dict) else None def _tail_file(path: Path, lines: int) -> str: if lines <= 0: return "" try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: return "" return "\n".join(text.splitlines()[-lines:]) def handle_sdk_refresh_metadata(config: Config, args: dict[str, Any]) -> dict[str, Any]: local_sdk = _local_sdk_path(config, args) if local_sdk is None: return tool_error("paths.local_sdk or local_sdk argument is required") target = _local_sdk_target(config, args, local_sdk) if target is None: return tool_error("target, arch, or a device with architecture is required") main_target = _main_sdk_target(target) timeout = _int_arg(args, "timeout", default=600, minimum=1, maximum=3600) command = ["sb2", "-t", main_target, "-m", "sdk-install", "-R", "zypper", "ref"] if _bool_arg(args, "force", default=False): command.append("-f") result = run(_local_sdk_docker_argv(local_sdk, command), timeout=timeout) return command_result( "refresh local SDK metadata", result, { "local_sdk": str(local_sdk), "target": target, "main_target": main_target, }, ) def handle_obs_results(config: Config, args: dict[str, Any]) -> dict[str, Any]: project = _str_arg(args, "project") package = _optional_str(args, "package") api_alias = _optional_str(args, "api_alias") or config.paths.osc_api_alias command = ["osc"] if api_alias: command += ["-A", api_alias] command += ["results", project] if package: command.append(package) timeout = _int_arg(args, "timeout", default=60, minimum=1, maximum=600) return command_result("OBS results", run(command, timeout=timeout)) def handle_obs_buildlog(config: Config, args: dict[str, Any]) -> dict[str, Any]: project = _str_arg(args, "project") package = _str_arg(args, "package") repository = _str_arg(args, "repository") arch = _str_arg(args, "arch") api_alias = _optional_str(args, "api_alias") or config.paths.osc_api_alias timeout = _int_arg(args, "timeout", default=90, minimum=1, maximum=1800) nostream = _bool_arg(args, "nostream", default=True) command = ["osc"] if api_alias: command += ["-A", api_alias] if nostream: path = "/build/{}/{}/{}/{}/_log?nostream=1".format( quote(project, safe=""), quote(repository, safe=""), quote(arch, safe=""), quote(package, safe=""), ) command += ["api", path] else: command += ["remotebuildlog", project, package, repository, arch] return command_result("OBS build log", run(command, timeout=timeout)) def handle_repo_status(config: Config, args: dict[str, Any]) -> dict[str, Any]: path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=120) result = run(["git", "status", "--short", "--branch"], cwd=path, timeout=timeout) return command_result("git status", result, {"path": str(path)}) def handle_repo_find(config: Config, args: dict[str, Any]) -> dict[str, Any]: path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) query = _str_arg(args, "query") max_count = _int_arg(args, "max_count", default=100, minimum=1, maximum=1000) fixed_strings = _bool_arg(args, "fixed_strings", default=True) timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) if shutil.which("rg"): command = ["rg", "--line-number", "--hidden", "--glob", "!.git"] if fixed_strings: command.append("--fixed-strings") command += ["--max-count", str(max_count), query, str(path)] else: command = ["grep", "-R", "-n", query, str(path)] result = run(command, timeout=timeout) if result.returncode == 1 and not result.stdout: return ok_text("no matches", {"path": str(path), "query": query, "matches": 0}) return command_result("repo find", result, {"path": str(path), "query": query}) def handle_spec_summary(config: Config, args: dict[str, Any]) -> dict[str, Any]: path_arg = _optional_str(args, "spec_path") if path_arg: spec_path = _safe_input_path(config, path_arg, allow_tmp=False) else: repo = _safe_input_path(config, _optional_str(args, "repo_path") or ".", allow_tmp=False) specs = sorted((repo / "rpm").glob("*.spec")) if not specs: return tool_error(f"no rpm/*.spec file found under {repo}") spec_path = specs[0] if not spec_path.exists(): return tool_error(f"spec file does not exist: {spec_path}") data = _parse_spec(spec_path) text_lines = [f"{key}: {value}" for key, value in data.items() if value] return ok_text("\n".join(text_lines), {"spec_path": str(spec_path), "summary": data}) def handle_qml_find_translations(config: Config, args: dict[str, Any]) -> dict[str, Any]: path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) timeout = _int_arg(args, "timeout", default=30, minimum=1, maximum=300) if not shutil.which("rg"): return tool_error("rg is required for qml_find_translations") command = [ "rg", "--line-number", "--glob", "*.qml", r"qsTrId|//%|//:", str(path), ] result = run(command, timeout=timeout) if result.returncode == 1 and not result.stdout: return ok_text("no QML translation markers found", {"path": str(path), "matches": 0}) return command_result("QML translations", result, {"path": str(path)}) def handle_qml_check_translator_ternaries(config: Config, args: dict[str, Any]) -> dict[str, Any]: path = _safe_input_path(config, _optional_str(args, "path") or ".", allow_tmp=False) files = [path] if path.is_file() and path.suffix == ".qml" else sorted(path.rglob("*.qml")) findings: list[dict[str, Any]] = [] for qml in files: if ".git" in qml.parts: continue try: lines = qml.read_text(encoding="utf-8").splitlines() except UnicodeDecodeError: continue for index, line in enumerate(lines, start=1): if "qsTrId(" in line and "?" in line and ":" in line: findings.append( { "path": str(qml), "line": index, "text": line.strip(), "message": ( "ternary qsTrId expression should give each branch " "its own translator comment and source text" ), } ) if not findings: return ok_text("no ternary translation issues found", {"path": str(path), "findings": []}) text = "\n".join( f"{item['path']}:{item['line']}: {item['message']}\n {item['text']}" for item in findings ) return { "content": [{"type": "text", "text": text}], "structuredContent": {"path": str(path), "findings": findings}, "isError": True, } def _run_ssh( config: Config, device: DeviceConfig, command: list[str], *, timeout: int, ) -> CommandResult: return run( ssh_argv(device, config.paths.ssh_config, remote_command(command)), timeout=timeout, ) def _default_screenshot_path(device: DeviceConfig, label: str) -> str: timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%f") return f"{_device_home_path(device)}/Pictures/Screenshots/{label}-{timestamp}.png" def _screenshot_prepare_command(device: DeviceConfig, remote_path: str) -> str: remote_dir = str(PurePosixPath(remote_path).parent) home_path = _device_home_path(device) script = f""" set -e home={shlex.quote(home_path)} remote_dir={shlex.quote(remote_dir)} owner=$(stat -Lc %U "$home") group=$(stat -Lc %G "$home") if [ "$remote_dir" = "$home/Pictures/Screenshots" ]; then if [ ! -d "$home/Pictures" ]; then install -d -m 775 -o "$owner" -g "$group" "$home/Pictures" fi if getent group privileged >/dev/null 2>&1; then screenshot_group=privileged else screenshot_group=$group fi if [ ! -d "$remote_dir" ]; then install -d -m 755 -o "$owner" -g "$screenshot_group" "$remote_dir" fi else if [ ! -d "$remote_dir" ]; then install -d -m 755 -o "$owner" -g "$group" "$remote_dir" fi fi """.strip() return remote_command(["sh", "-lc", script]) def _run_as_user_command(device: DeviceConfig, inner: str) -> str: if not re.fullmatch(r"[A-Za-z0-9_.-]+", device.username): raise ValueError("device username contains unsupported characters") username = shlex.quote(device.username) quoted_inner = shlex.quote(inner) script = f""" if command -v runuser >/dev/null 2>&1; then runuser -u {username} -- {inner} else su -s /bin/sh {username} -c {quoted_inner} fi """.strip() return remote_command(["sh", "-lc", script]) def _browser_launch_command( device: DeviceConfig, url: str, stop_stale: bool, wait_seconds: int, ) -> str: script = f""" set -u export XDG_RUNTIME_DIR={shlex.quote(device.user_bus_runtime_dir)} export DBUS_SESSION_BUS_ADDRESS={shlex.quote(device.user_bus_address)} if [ {shlex.quote("1" if stop_stale else "0")} = 1 ]; then systemctl --user stop booster-browser@sailfish-browser.service >/dev/null 2>&1 || true browser_name=sailfish-browser for pid in $(pidof "$browser_name" 2>/dev/null || true); do kill "$pid" >/dev/null 2>&1 || true done if command -v ps >/dev/null 2>&1; then firejail_name=firejail ps -eo pid=,comm=,args= 2>/dev/null | awk -v c="$firejail_name" -v b="$browser_name" '$2 == c && index($0, b) {{ print $1 }}' | while read -r pid; do [ -n "$pid" ] || continue kill "$pid" >/dev/null 2>&1 || true done fi fi WAYLAND_DISPLAY=../../display/wayland-0 QT_QPA_PLATFORM=hwcomposer \\ /usr/bin/invoker --type=browser,silica-qt5 -A -- \\ /usr/bin/sailfish-browser {shlex.quote(url)} \\ >/tmp/sailfish-browser-mcp.log 2>&1 & launcher_pid=$! echo "launcher_pid=$launcher_pid" echo "launch_log=/tmp/sailfish-browser-mcp.log" if [ {wait_seconds} -gt 0 ]; then sleep {wait_seconds} fi reply=$(dbus-send --session --print-reply \\ --dest=org.nemomobile.lipstick \\ / \\ org.freedesktop.DBus.Properties.Get \\ string:org.nemomobile.compositor \\ string:privateTopmostWindowProcessId 2>/dev/null || true) topmost_pid=$(printf '%s\\n' "$reply" | awk '/\\b(int32|uint32|int64|uint64)\\b/ {{ print $2; exit }}') if [ -n "$topmost_pid" ]; then echo "topmost_pid=$topmost_pid" if [ -r "/proc/$topmost_pid/maps" ]; then if grep -q 'libxul\\.so' "/proc/$topmost_pid/maps"; then echo "libxul=present" else echo "libxul=absent" fi fi else echo "topmost_pid=" fi """.strip() return remote_command(["sh", "-lc", script]) def _local_sdk_path(config: Config, args: dict[str, Any]) -> Path | None: value = _optional_str(args, "local_sdk") if value: return Path(value).expanduser().resolve(strict=False) if config.paths.local_sdk: return config.paths.local_sdk.expanduser().resolve(strict=False) return None def _local_sdk_target(config: Config, args: dict[str, Any], local_sdk: Path) -> str | None: explicit_target = _optional_str(args, "target") if explicit_target: return build_sailfishos.canonical_local_target_name(explicit_target) device = config.device(_optional_str(args, "device")) if args.get("device") else None arch = _optional_str(args, "arch") or (device.architecture if device else "") if not arch: return None release = _optional_str(args, "release") or (device.release if device else "") or "live" normalized_release = build_sailfishos.normalize_local_release(release) or "live" matching = [ target for target in build_sailfishos.list_local_sdk_targets(local_sdk) if target.arch == arch and build_sailfishos.local_target_matches_release(target, normalized_release) ] if matching: return matching[0].target if normalized_release == "live": return arch return f"{arch}-{normalized_release}" def _main_sdk_target(target: str) -> str: if target.endswith(".default"): return target for suffix in (".build-sailfishos-skill", ".build"): if target.endswith(suffix): target = target[: -len(suffix)] break return f"{target}.default" def _local_sdk_docker_argv(local_sdk: Path, command: list[str]) -> list[str]: user = build_sailfishos.host_user() uid = os.getuid() gid = os.getgid() home = str(Path.home().resolve()) image = build_sailfishos.local_sdk_build_engine_image(user) sdk_mount_root = build_sailfishos.local_sdk_mount_root(local_sdk) inner = remote_command(command) wrapper_command = f""" set -euo pipefail if [ ! -x "$LOCAL_SDK" ]; then echo "Installed Sailfish SDK chroot not found or not executable at $LOCAL_SDK" >&2 exit 1 fi if getent passwd mersdk >/dev/null 2>&1; then sed -i 's#^mersdk:[^:]*:[0-9]*:[0-9]*:[^:]*:[^:]*:#{user}:x:{uid}:{gid}::{home}:#' /etc/passwd elif ! getent passwd {shlex.quote(user)} >/dev/null 2>&1; then printf '%s:x:%s:%s::%s:/bin/bash\\n' {shlex.quote(user)} {uid} {gid} {shlex.quote(home)} >> /etc/passwd fi "$LOCAL_SDK" -u {shlex.quote(user)} {inner} """.strip() return [ "docker", "run", "--rm", "--privileged", "-v", f"{sdk_mount_root}:{sdk_mount_root}", "-e", f"LOCAL_SDK={local_sdk}", image, "bash", "-lc", wrapper_command, ] def _touch_discover_command(include_evdev_trace: bool) -> str: evdev_trace = """ if command -v evdev_trace >/dev/null 2>&1; then echo "### evdev_trace -i" if command -v timeout >/dev/null 2>&1; then timeout 5 evdev_trace -i || true else evdev_trace -i & pid=$! sleep 5 kill "$pid" >/dev/null 2>&1 || true wait "$pid" >/dev/null 2>&1 || true fi echo fi """.strip() script = f""" set -e {evdev_trace if include_evdev_trace else ""} echo "### /proc/bus/input/devices" cat /proc/bus/input/devices """.strip() return remote_command(["sh", "-lc", script]) def _touch_inject_command(values: dict[str, str | int]) -> str: env = " ".join( f"{key}={shlex.quote(str(value))}" for key, value in sorted(values.items()) ) script = f""" set -e {env} python3 - <<'PY' {_touch_inject_python()} PY """.strip() return remote_command(["sh", "-lc", script]) def _touch_inject_python() -> str: return r''' import os import re import struct import sys import time EV_SYN = 0 EV_KEY = 1 EV_ABS = 3 SYN_REPORT = 0 BTN_TOUCH = 0x14a ABS_X = 0x00 ABS_Y = 0x01 ABS_MT_SLOT = 0x2f ABS_MT_POSITION_X = 0x35 ABS_MT_POSITION_Y = 0x36 ABS_MT_TRACKING_ID = 0x39 TOUCH_KEYWORDS = ( "touch", "touchscreen", "digitizer", "fts", "ft5x", "goodix", "synaptics", "cyttsp", "atmel", "elan", "himax", "novatek", "nvt", "silead", "gt9", "spi3", ) def die(message): print(message, file=sys.stderr) raise SystemExit(2) def input_blocks(): try: text = open("/proc/bus/input/devices", encoding="utf-8").read() except OSError as exc: die(f"failed to read /proc/bus/input/devices: {exc}") return [block for block in text.split("\n\n") if block.strip()] def field(block, prefix): for line in block.splitlines(): if line.startswith(prefix): return line[len(prefix):].strip() return "" def event_handler(block): match = re.search(r"\bevent\d+\b", field(block, "H: Handlers=")) return match.group(0) if match else "" def name(block): value = field(block, "N: Name=") return value.strip('"') def score(block): lower = block.lower() value = 0 if any(keyword in lower for keyword in TOUCH_KEYWORDS): value += 100 if "b: abs=" in lower: value += 20 if "b: prop=" in lower: value += 5 if "mouse" in lower or "keyboard" in lower or "keypad" in lower: value -= 100 return value def discover(): devices = [] for block in input_blocks(): event = event_handler(block) if not event: continue devices.append( { "path": f"/dev/input/{event}", "name": name(block), "score": score(block), } ) devices.sort(key=lambda item: item["score"], reverse=True) return devices def choose_device(): explicit = os.environ.get("INPUT_DEVICE", "") if explicit: return explicit devices = discover() if not devices or devices[0]["score"] <= 0: summary = ", ".join(f"{item['path']}:{item['name']}" for item in devices) die(f"could not identify touchscreen input device; candidates: {summary}") return devices[0]["path"] def event(fileobj, event_type, code, value): fileobj.write(struct.pack("@llHHi", 0, 0, event_type, code, value)) def sync(fileobj): event(fileobj, EV_SYN, SYN_REPORT, 0) def move(fileobj, x, y): event(fileobj, EV_ABS, ABS_MT_POSITION_X, x) event(fileobj, EV_ABS, ABS_MT_POSITION_Y, y) event(fileobj, EV_ABS, ABS_X, x) event(fileobj, EV_ABS, ABS_Y, y) def down(fileobj, x, y): event(fileobj, EV_ABS, ABS_MT_SLOT, 0) event(fileobj, EV_ABS, ABS_MT_TRACKING_ID, int(time.time() * 1000) & 0x7fffffff) move(fileobj, x, y) event(fileobj, EV_KEY, BTN_TOUCH, 1) sync(fileobj) def up(fileobj): event(fileobj, EV_ABS, ABS_MT_SLOT, 0) event(fileobj, EV_ABS, ABS_MT_TRACKING_ID, -1) event(fileobj, EV_KEY, BTN_TOUCH, 0) sync(fileobj) def integer(name): try: return int(os.environ[name]) except KeyError: die(f"missing {name}") except ValueError: die(f"{name} must be an integer") action = os.environ.get("ACTION", "") device = choose_device() print(f"input_device={device}") with open(device, "wb", buffering=0) as fileobj: if action == "tap": down(fileobj, integer("X"), integer("Y")) time.sleep(integer("HOLD_MS") / 1000) up(fileobj) elif action == "swipe": start_x = integer("START_X") start_y = integer("START_Y") end_x = integer("END_X") end_y = integer("END_Y") steps = integer("STEPS") duration = integer("DURATION_MS") / 1000 down(fileobj, start_x, start_y) for step in range(1, steps + 1): fraction = step / steps x = round(start_x + ((end_x - start_x) * fraction)) y = round(start_y + ((end_y - start_y) * fraction)) move(fileobj, x, y) sync(fileobj) time.sleep(duration / steps) up(fileobj) else: die(f"unsupported ACTION {action!r}") print(f"{action}=ok") '''.strip() def _device_home_path(device: DeviceConfig) -> str: return f"/home/{device.username}" def command_result( title: str, result: CommandResult, structured: dict[str, Any] | None = None, ) -> dict[str, Any]: data = result.public_dict() if structured: data.update(structured) return { "content": [{"type": "text", "text": _command_text(title, result)}], "structuredContent": data, "isError": not result.ok, } def combined_result(title: str, steps: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]: text_parts: list[str] = [] structured_steps: dict[str, Any] = {} is_error = False for name, result in steps: content = result.get("content") or [] step_text = content[0].get("text", "") if content else "" text_parts.append(f"{name}:\n{step_text}".rstrip()) structured_steps[name] = result.get("structuredContent", {}) is_error = is_error or bool(result.get("isError", False)) return { "content": [{"type": "text", "text": "\n\n".join(text_parts)}], "structuredContent": { "title": title, "step_order": [name for name, _ in steps], "steps": structured_steps, }, "isError": is_error, } def ok_text(text: str, structured: dict[str, Any] | None = None) -> dict[str, Any]: return { "content": [{"type": "text", "text": text}], "structuredContent": structured or {}, "isError": False, } def tool_error(message: str, structured: dict[str, Any] | None = None) -> dict[str, Any]: return { "content": [{"type": "text", "text": message}], "structuredContent": structured or {"error": message}, "isError": True, } def _command_text(title: str, result: CommandResult) -> str: stdout, _ = truncate(result.stdout, 12000) stderr, _ = truncate(result.stderr, 8000) parts = [f"{title}: exit {result.returncode}"] if stdout: parts += ["", stdout.rstrip()] if stderr: parts += ["", "stderr:", stderr.rstrip()] return "\n".join(parts) def _device(config: Config, args: dict[str, Any]) -> DeviceConfig: return config.device(_optional_str(args, "device")) def _str_arg(args: dict[str, Any], name: str) -> str: value = args.get(name) if not isinstance(value, str) or not value: raise ValueError(f"{name} must be a non-empty string") return value def _optional_str(args: dict[str, Any], name: str) -> str | None: value = args.get(name) if value is None or value == "": return None if not isinstance(value, str): raise ValueError(f"{name} must be a string") return value def _int_arg( args: dict[str, Any], name: str, default: int | None = None, minimum: int | None = None, maximum: int | None = None, ) -> int: value = args.get(name, default) if not isinstance(value, int): raise ValueError(f"{name} must be an integer") if minimum is not None and value < minimum: raise ValueError(f"{name} must be >= {minimum}") if maximum is not None and value > maximum: raise ValueError(f"{name} must be <= {maximum}") return value def _bool_arg(args: dict[str, Any], name: str, default: bool = False) -> bool: value = args.get(name, default) if not isinstance(value, bool): raise ValueError(f"{name} must be a boolean") return value def _enum_arg( args: dict[str, Any], name: str, values: list[str], default: str | None = None, ) -> str: value = args.get(name, default) if not isinstance(value, str) or value not in values: raise ValueError(f"{name} must be one of: {', '.join(values)}") return value def _string_list_arg(args: dict[str, Any], name: str) -> list[str]: value = args.get(name) if value is None: return [] if not isinstance(value, list) or not all(isinstance(item, str) for item in value): raise ValueError(f"{name} must be a list of strings") return value def _command_list_arg(args: dict[str, Any], name: str) -> list[str]: value = args.get(name) if ( not isinstance(value, list) or not value or not all(isinstance(item, str) and item for item in value) ): raise ValueError(f"{name} must be a non-empty list of strings") return value def _service_unit_arg(args: dict[str, Any], name: str) -> str: unit = _str_arg(args, name) if not re.fullmatch(r"[A-Za-z0-9@_.:\-]+", unit): raise ValueError(f"{name} contains unsupported characters") return unit def _safe_input_path(config: Config, value: str, *, allow_tmp: bool) -> Path: path = Path(value).expanduser() if not path.is_absolute(): path = config.paths.git_root / path path = path.resolve(strict=False) roots = _host_path_roots(config) if allow_tmp: roots.append(Path("/tmp").resolve(strict=False)) if not _is_relative_to_any(path, roots): raise ValueError(f"path is outside allowed roots: {path}") return path def _safe_output_path(config: Config, value: str) -> Path: path = Path(value).expanduser() if not path.is_absolute(): path = config.paths.git_root / path path = path.resolve(strict=False) roots = _host_path_roots(config) + [Path("/tmp").resolve(strict=False)] if not _is_relative_to_any(path, roots): raise ValueError(f"output path is outside allowed roots: {path}") return path def _host_path_roots(config: Config) -> list[Path]: roots: list[Path] = [] for root in (config.paths.git_root, config.paths.obs_root): resolved = root.resolve(strict=False) if resolved not in roots: roots.append(resolved) return roots def _is_relative_to_any(path: Path, roots: list[Path]) -> bool: for root in roots: try: path.relative_to(root) return True except ValueError: pass return False def _parse_dbus_integer(text: str) -> int | None: match = re.search(r"\b(?:int32|uint32|int64|uint64)\s+(-?\d+)", text) return int(match.group(1)) if match else None def _parse_spec(path: Path) -> dict[str, Any]: fields: dict[str, Any] = { "Name": "", "Version": "", "Release": "", "Summary": "", "License": "", "URL": "", "BuildRequires": [], "Requires": [], } for raw_line in path.read_text(encoding="utf-8", errors="replace").splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue for field in ["Name", "Version", "Release", "Summary", "License", "URL"]: prefix = field + ":" if line.startswith(prefix) and not fields[field]: fields[field] = line[len(prefix) :].strip() if line.startswith("BuildRequires:"): fields["BuildRequires"].append(line.split(":", 1)[1].strip()) if line.startswith("Requires:"): fields["Requires"].append(line.split(":", 1)[1].strip()) return fields def _read_only_annotations(title: str) -> dict[str, Any]: return { "title": title, "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, } def _mutating_annotations(title: str, *, destructive: bool = False) -> dict[str, Any]: return { "title": title, "readOnlyHint": False, "destructiveHint": destructive, "idempotentHint": False, "openWorldHint": True, } def _object_schema(properties: dict[str, Any], required: list[str] | None = None) -> dict[str, Any]: return { "type": "object", "properties": properties, "required": required or [], "additionalProperties": False, } def _device_prop() -> dict[str, Any]: return {"type": "string", "description": "Configured device alias or ssh target."} def _timeout_prop(default: int) -> dict[str, Any]: return { "type": "integer", "minimum": 1, "default": default, "description": "Command timeout in seconds.", } def _spec_devices() -> dict[str, Any]: return { "name": "sailfish_devices", "title": "List Sailfish Devices", "description": "List configured Sailfish OS device aliases and SSH targets.", "inputSchema": _object_schema({}), "annotations": _read_only_annotations("List Sailfish Devices"), } def _spec_device_journal() -> dict[str, Any]: return { "name": "sailfish_device_journal", "title": "Read Device Journal", "description": "Read recent journalctl output from a Sailfish OS device.", "inputSchema": _object_schema( { "device": _device_prop(), "unit": {"type": "string"}, "since": {"type": "string"}, "grep": {"type": "string"}, "lines": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 200}, "timeout": _timeout_prop(30), } ), "annotations": _read_only_annotations("Read Device Journal"), } def _spec_device_topmost_pid() -> dict[str, Any]: return { "name": "sailfish_device_topmost_pid", "title": "Topmost Window PID", "description": "Query Lipstick for the current topmost window process id.", "inputSchema": _object_schema({"device": _device_prop(), "timeout": _timeout_prop(20)}), "annotations": _read_only_annotations("Topmost Window PID"), } def _spec_device_proc_maps() -> dict[str, Any]: return { "name": "sailfish_device_proc_maps", "title": "Read Process Maps", "description": "Read or filter /proc//maps on a Sailfish OS device.", "inputSchema": _object_schema( { "device": _device_prop(), "pid": {"type": "integer", "minimum": 1}, "contains": {"type": "string"}, "max_lines": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 200}, "timeout": _timeout_prop(20), }, ["pid"], ), "annotations": _read_only_annotations("Read Process Maps"), } def _spec_device_lipstick_screenshot() -> dict[str, Any]: return { "name": "sailfish_device_lipstick_screenshot", "title": "Lipstick Screenshot", "description": "Ask Lipstick to save a screenshot on the device, optionally pulling it locally.", "inputSchema": _object_schema( { "device": _device_prop(), "remote_path": { "type": "string", "default": "/home/defaultuser/Pictures/Screenshots/lipstick-.png", "description": "Lipstick accepts screenshot paths under the user home directory.", }, "local_path": {"type": "string"}, "privileged": {"type": "boolean", "default": True}, "timeout": _timeout_prop(30), } ), "annotations": _mutating_annotations("Lipstick Screenshot"), } def _spec_device_touch() -> dict[str, Any]: return { "name": "sailfish_device_touch", "title": "Device Touch Input", "description": "Discover the touchscreen input device or inject tap/swipe events over SSH.", "inputSchema": _object_schema( { "device": _device_prop(), "action": { "type": "string", "enum": ["discover", "tap", "swipe"], "description": "discover lists input devices; tap and swipe inject Linux input events.", }, "include_evdev_trace": { "type": "boolean", "default": False, "description": "Also run evdev_trace -i during discovery. Disabled by default because it may block on some devices.", }, "input_device": { "type": "string", "description": "Optional explicit device path, for example /dev/input/event5.", }, "x": {"type": "integer", "minimum": 0, "maximum": 10000}, "y": {"type": "integer", "minimum": 0, "maximum": 10000}, "hold_ms": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 80}, "start_x": {"type": "integer", "minimum": 0, "maximum": 10000}, "start_y": {"type": "integer", "minimum": 0, "maximum": 10000}, "end_x": {"type": "integer", "minimum": 0, "maximum": 10000}, "end_y": {"type": "integer", "minimum": 0, "maximum": 10000}, "duration_ms": {"type": "integer", "minimum": 1, "maximum": 10000, "default": 300}, "steps": {"type": "integer", "minimum": 1, "maximum": 200, "default": 12}, "timeout": _timeout_prop(10), }, ["action"], ), "annotations": _mutating_annotations("Device Touch Input"), } def _spec_device_touch_workflow() -> dict[str, Any]: return { "name": "sailfish_device_touch_workflow", "title": "Screenshot And Touch", "description": "Capture a Lipstick screenshot, optionally list touch inputs, then inject a tap or swipe.", "inputSchema": _object_schema( { "device": _device_prop(), "action": {"type": "string", "enum": ["tap", "swipe"]}, "input_device": { "type": "string", "description": "Optional explicit device path, for example /dev/input/event5.", }, "x": {"type": "integer", "minimum": 0, "maximum": 10000}, "y": {"type": "integer", "minimum": 0, "maximum": 10000}, "hold_ms": {"type": "integer", "minimum": 1, "maximum": 5000, "default": 80}, "start_x": {"type": "integer", "minimum": 0, "maximum": 10000}, "start_y": {"type": "integer", "minimum": 0, "maximum": 10000}, "end_x": {"type": "integer", "minimum": 0, "maximum": 10000}, "end_y": {"type": "integer", "minimum": 0, "maximum": 10000}, "duration_ms": {"type": "integer", "minimum": 1, "maximum": 10000, "default": 300}, "steps": {"type": "integer", "minimum": 1, "maximum": 200, "default": 12}, "screenshot_before": {"type": "boolean", "default": True}, "screenshot_after": {"type": "boolean", "default": False}, "discover_input": {"type": "boolean", "default": True}, "include_evdev_trace": {"type": "boolean", "default": False}, "privileged": {"type": "boolean", "default": True}, "local_path": { "type": "string", "description": "Optional local path for the before screenshot.", }, "before_local_path": {"type": "string"}, "after_local_path": {"type": "string"}, "before_remote_path": {"type": "string"}, "after_remote_path": {"type": "string"}, "timeout": _timeout_prop(30), }, ["action"], ), "annotations": _mutating_annotations("Screenshot And Touch"), } def _spec_device_user_bus_call() -> dict[str, Any]: return { "name": "sailfish_device_user_bus_call", "title": "User Bus Call", "description": "Run a typed dbus-send method call on the defaultuser session bus.", "inputSchema": _object_schema( { "device": _device_prop(), "destination": {"type": "string"}, "path": {"type": "string"}, "interface": {"type": "string"}, "member": {"type": "string"}, "arguments": { "type": "array", "items": {"type": "string"}, "description": "Raw dbus-send argument strings, for example string:/tmp/file.", }, "timeout": _timeout_prop(30), }, ["destination", "path", "interface", "member"], ), "annotations": _mutating_annotations("User Bus Call"), } def _spec_device_user_session_command() -> dict[str, Any]: return { "name": "sailfish_device_user_session_command", "title": "User Session Command", "description": "Run a command with the configured Sailfish user-session D-Bus environment.", "inputSchema": _object_schema( { "device": _device_prop(), "command": { "type": "array", "items": {"type": "string"}, "minItems": 1, "description": "Command argv to run; shell parsing is not applied.", }, "run_as_user": { "type": "boolean", "default": False, "description": "Run through runuser/su as the configured device username.", }, "timeout": _timeout_prop(30), }, ["command"], ), "annotations": _mutating_annotations("User Session Command"), } def _spec_device_install_rpm() -> dict[str, Any]: return { "name": "sailfish_device_install_rpm", "title": "Install Device RPM", "description": "Copy a local RPM to a Sailfish OS device and install it.", "inputSchema": _object_schema( { "device": _device_prop(), "rpm_path": {"type": "string"}, "remote_path": {"type": "string"}, "installer": {"type": "string", "enum": ["pkcon", "rpm"], "default": "pkcon"}, "timeout": _timeout_prop(180), }, ["rpm_path"], ), "annotations": _mutating_annotations("Install Device RPM"), } def _spec_device_restart_service() -> dict[str, Any]: return { "name": "sailfish_device_restart_service", "title": "Manage Device Service", "description": "Run systemctl start/stop/restart/status for a system or user service.", "inputSchema": _object_schema( { "device": _device_prop(), "unit": {"type": "string"}, "action": { "type": "string", "enum": ["restart", "start", "stop", "status"], "default": "restart", }, "mode": {"type": "string", "enum": ["system", "user"], "default": "system"}, "timeout": _timeout_prop(60), }, ["unit"], ), "annotations": _mutating_annotations("Manage Device Service"), } def _spec_device_browser_launch() -> dict[str, Any]: return { "name": "sailfish_device_browser_launch", "title": "Launch Sailfish Browser", "description": "Stop stale browser state, launch Sailfish Browser with display/session env, and report topmost PID details.", "inputSchema": _object_schema( { "device": _device_prop(), "url": {"type": "string"}, "stop_stale": {"type": "boolean", "default": True}, "wait_seconds": {"type": "integer", "minimum": 0, "maximum": 60, "default": 3}, "timeout": _timeout_prop(60), }, ["url"], ), "annotations": _mutating_annotations("Launch Sailfish Browser"), } def _spec_build_rpm() -> dict[str, Any]: return { "name": "sailfish_build_rpm", "title": "Build Sailfish RPM", "description": "Start an asynchronous RPM build; paths.local_sdk defaults to the live installed SDK, with public SDK fallback only for explicit named releases.", "inputSchema": _object_schema( { "project_path": {"type": "string"}, "device": { "type": "string", "description": "Optional configured device to supply default release and architecture.", }, "release": {"type": "string"}, "arch": { "oneOf": [ {"type": "string"}, {"type": "array", "items": {"type": "string"}}, ] }, "artifacts_dir": {"type": "string"}, "all_arches": {"type": "boolean", "default": False}, "clean": {"type": "boolean", "default": False}, "debug": {"type": "boolean", "default": False}, "no_pull": {"type": "boolean", "default": False}, "local_rpms_dir": {"type": "array", "items": {"type": "string"}}, "wait": { "type": "boolean", "default": False, "description": "Wait for the build and return the old synchronous command result.", }, "timeout": _timeout_prop(3600), }, ["project_path"], ), "annotations": _mutating_annotations("Build Sailfish RPM"), } def _spec_build_status() -> dict[str, Any]: return { "name": "sailfish_build_status", "title": "Build Job Status", "description": "Read status and recent log output for asynchronous Sailfish RPM build jobs.", "inputSchema": _object_schema( { "job_id": { "type": "string", "description": "Build job id returned by sailfish_build_rpm. Omit to list recent jobs.", }, "lines": { "type": "integer", "minimum": 0, "maximum": 1000, "default": 80, "description": "Number of trailing log lines to include.", }, } ), "annotations": _read_only_annotations("Build Job Status"), } def _spec_sdk_refresh_metadata() -> dict[str, Any]: return { "name": "sailfish_sdk_refresh_metadata", "title": "Refresh SDK Metadata", "description": "Refresh zypper repository metadata in an installed local SDK main target.", "inputSchema": _object_schema( { "device": { "type": "string", "description": "Optional configured device to supply default release and architecture.", }, "local_sdk": { "type": "string", "description": "Override paths.local_sdk for this call.", }, "target": { "type": "string", "description": "Local SDK target base or .default target, for example aarch64 or aarch64.default.", }, "release": {"type": "string"}, "arch": {"type": "string"}, "force": { "type": "boolean", "default": False, "description": "Append zypper ref -f.", }, "timeout": _timeout_prop(600), } ), "annotations": _mutating_annotations("Refresh SDK Metadata"), } def _spec_obs_results() -> dict[str, Any]: return { "name": "sailfish_obs_results", "title": "OBS Results", "description": "Run osc results using the configured OBS API alias when set.", "inputSchema": _object_schema( { "project": {"type": "string"}, "package": {"type": "string"}, "api_alias": {"type": "string"}, "timeout": _timeout_prop(60), }, ["project"], ), "annotations": _read_only_annotations("OBS Results"), } def _spec_obs_buildlog() -> dict[str, Any]: return { "name": "sailfish_obs_buildlog", "title": "OBS Build Log", "description": "Fetch an OBS build log with osc; defaults to the API nostream form to avoid live streams.", "inputSchema": _object_schema( { "project": {"type": "string"}, "package": {"type": "string"}, "repository": {"type": "string"}, "arch": {"type": "string"}, "api_alias": {"type": "string"}, "nostream": {"type": "boolean", "default": True}, "timeout": _timeout_prop(90), }, ["project", "package", "repository", "arch"], ), "annotations": _read_only_annotations("OBS Build Log"), } def _spec_repo_status() -> dict[str, Any]: return { "name": "sailfish_repo_status", "title": "Repo Status", "description": "Run git status --short --branch under the configured git root.", "inputSchema": _object_schema({"path": {"type": "string"}, "timeout": _timeout_prop(30)}), "annotations": _read_only_annotations("Repo Status"), } def _spec_repo_find() -> dict[str, Any]: return { "name": "sailfish_repo_find", "title": "Repo Find", "description": "Search a repo or subtree under the configured git root.", "inputSchema": _object_schema( { "path": {"type": "string"}, "query": {"type": "string"}, "fixed_strings": {"type": "boolean", "default": True}, "max_count": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100}, "timeout": _timeout_prop(30), }, ["query"], ), "annotations": _read_only_annotations("Repo Find"), } def _spec_spec_summary() -> dict[str, Any]: return { "name": "sailfish_spec_summary", "title": "RPM Spec Summary", "description": "Parse high-level metadata from a Sailfish RPM spec file.", "inputSchema": _object_schema( {"repo_path": {"type": "string"}, "spec_path": {"type": "string"}} ), "annotations": _read_only_annotations("RPM Spec Summary"), } def _spec_qml_find_translations() -> dict[str, Any]: return { "name": "sailfish_qml_find_translations", "title": "Find QML Translations", "description": "Find qsTrId and translator comments in QML files.", "inputSchema": _object_schema({"path": {"type": "string"}, "timeout": _timeout_prop(30)}), "annotations": _read_only_annotations("Find QML Translations"), } def _spec_qml_check_translator_ternaries() -> dict[str, Any]: return { "name": "sailfish_qml_check_translator_ternaries", "title": "Check QML Ternary Translations", "description": "Flag QML ternary qsTrId expressions that need branch-local comments.", "inputSchema": _object_schema({"path": {"type": "string"}}), "annotations": _read_only_annotations("Check QML Ternary Translations"), }