diff options
Diffstat (limited to 'src/sailfish_devel_mcp')
| -rw-r--r-- | src/sailfish_devel_mcp/config.py | 9 | ||||
| -rw-r--r-- | src/sailfish_devel_mcp/tools.py | 738 | ||||
| -rwxr-xr-x | src/sailfish_devel_mcp/vendor/build_sailfishos.py | 27 |
3 files changed, 758 insertions, 16 deletions
diff --git a/src/sailfish_devel_mcp/config.py b/src/sailfish_devel_mcp/config.py index 6515918..73740d1 100644 --- a/src/sailfish_devel_mcp/config.py +++ b/src/sailfish_devel_mcp/config.py @@ -39,6 +39,7 @@ class DeviceConfig: @dataclass(frozen=True) class PathConfig: git_root: Path = Path.home() / "git" + obs_root: Path = Path.home() / "OBS" ssh_config: Path = Path.home() / ".ssh" / "config" build_sailfishos: Path = BUNDLED_BUILD_HELPER local_sdk: Path | None = None @@ -47,6 +48,7 @@ class PathConfig: def public_dict(self) -> dict[str, str | None]: return { "git_root": str(self.git_root), + "obs_root": str(self.obs_root), "ssh_config": str(self.ssh_config), "build_sailfishos": str(self.build_sailfishos), "local_sdk": str(self.local_sdk) if self.local_sdk else None, @@ -155,6 +157,13 @@ def _load_paths(raw_paths: Any) -> PathConfig: or defaults.git_root ) ).expanduser(), + obs_root=Path( + str( + raw_paths.get("obs_root") + or os.environ.get("SAILFISH_MCP_OBS_ROOT") + or defaults.obs_root + ) + ).expanduser(), ssh_config=Path( str( raw_paths.get("ssh_config") diff --git a/src/sailfish_devel_mcp/tools.py b/src/sailfish_devel_mcp/tools.py index 1061e6c..854cd00 100644 --- a/src/sailfish_devel_mcp/tools.py +++ b/src/sailfish_devel_mcp/tools.py @@ -1,12 +1,18 @@ from __future__ import annotations from dataclasses import dataclass -from datetime import datetime +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 ( @@ -19,6 +25,7 @@ from .runner import ( truncate, user_bus_env, ) +from .vendor import build_sailfishos ToolHandler = Callable[[dict[str, Any]], dict[str, Any]] @@ -45,15 +52,32 @@ def build_registry(config: Config) -> dict[str, Tool]: 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)), @@ -152,7 +176,7 @@ def handle_device_proc_maps(config: Config, args: dict[str, Any]) -> dict[str, A def handle_device_lipstick_screenshot(config: Config, args: dict[str, Any]) -> dict[str, Any]: device = _device(config, args) - timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S") + 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" @@ -238,6 +262,66 @@ def handle_device_touch(config: Config, args: dict[str, Any]) -> dict[str, Any]: ) +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") @@ -260,6 +344,24 @@ def handle_device_user_bus_call(config: Config, args: dict[str, Any]) -> dict[st 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) @@ -304,6 +406,21 @@ def handle_device_restart_service(config: Config, args: dict[str, Any]) -> dict[ ) +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 @@ -317,6 +434,8 @@ def handle_build_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: 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): @@ -344,7 +463,248 @@ def handle_build_rpm(config: Config, args: dict[str, Any]) -> dict[str, Any]: 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) - return command_result("build Sailfish RPM", run(command, timeout=timeout)) + 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]: @@ -368,10 +728,20 @@ def handle_obs_buildlog(config: Config, args: dict[str, Any]) -> dict[str, Any]: 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] - command += ["remotebuildlog", project, package, repository, arch] + 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)) @@ -489,6 +859,11 @@ def _run_ssh( ) +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) @@ -519,6 +894,160 @@ fi 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 @@ -758,6 +1287,27 @@ def command_result( } +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}], @@ -850,6 +1400,17 @@ def _string_list_arg(args: dict[str, Any], name: str) -> list[str]: 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): @@ -862,7 +1423,7 @@ def _safe_input_path(config: Config, value: str, *, allow_tmp: bool) -> Path: if not path.is_absolute(): path = config.paths.git_root / path path = path.resolve(strict=False) - roots = [config.paths.git_root.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): @@ -875,12 +1436,21 @@ def _safe_output_path(config: Config, value: str) -> Path: if not path.is_absolute(): path = config.paths.git_root / path path = path.resolve(strict=False) - roots = [config.paths.git_root.resolve(strict=False), Path("/tmp").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: @@ -1083,6 +1653,49 @@ def _spec_device_touch() -> dict[str, Any]: } +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", @@ -1108,6 +1721,33 @@ def _spec_device_user_bus_call() -> dict[str, Any]: } +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", @@ -1150,11 +1790,30 @@ def _spec_device_restart_service() -> dict[str, Any]: } +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": "Run the local build-sailfishos helper; paths.local_sdk is used only when it has a matching target.", + "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"}, @@ -1175,6 +1834,11 @@ def _spec_build_rpm() -> dict[str, Any]: "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"], @@ -1183,6 +1847,63 @@ def _spec_build_rpm() -> dict[str, Any]: } +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", @@ -1205,7 +1926,7 @@ def _spec_obs_buildlog() -> dict[str, Any]: return { "name": "sailfish_obs_buildlog", "title": "OBS Build Log", - "description": "Fetch an OBS remote build log with osc.", + "description": "Fetch an OBS build log with osc; defaults to the API nostream form to avoid live streams.", "inputSchema": _object_schema( { "project": {"type": "string"}, @@ -1213,6 +1934,7 @@ def _spec_obs_buildlog() -> dict[str, Any]: "repository": {"type": "string"}, "arch": {"type": "string"}, "api_alias": {"type": "string"}, + "nostream": {"type": "boolean", "default": True}, "timeout": _timeout_prop(90), }, ["project", "package", "repository", "arch"], diff --git a/src/sailfish_devel_mcp/vendor/build_sailfishos.py b/src/sailfish_devel_mcp/vendor/build_sailfishos.py index 98b52c0..c9ad480 100755 --- a/src/sailfish_devel_mcp/vendor/build_sailfishos.py +++ b/src/sailfish_devel_mcp/vendor/build_sailfishos.py @@ -848,12 +848,9 @@ def normalize_local_release(release: str | None) -> str: def requested_release(project_dirs: Iterable[Path], explicit_release: str | None) -> str | None: - if explicit_release: - return explicit_release - - env_release = os.environ.get("SAILFISHOS_RELEASE") - if env_release: - return env_release + requested = explicit_requested_release(explicit_release) + if requested: + return requested seen: set[Path] = set() for project_dir in project_dirs: @@ -867,6 +864,21 @@ def requested_release(project_dirs: Iterable[Path], explicit_release: str | None return None +def explicit_requested_release(explicit_release: str | None) -> str | None: + if explicit_release: + return explicit_release + + env_release = os.environ.get("SAILFISHOS_RELEASE") + if env_release: + return env_release + + return None + + +def local_sdk_requested_release(explicit_release: str | None) -> str: + return normalize_local_release(explicit_requested_release(explicit_release)) or LIVE_RELEASE + + def select_local_sdk_builds( local_sdk: Path, release: str, @@ -1334,11 +1346,10 @@ def main() -> int: project_dir = resolve_project_dir(requested_project_dir) local_sdk_path = Path(args.local_sdk).expanduser().resolve(strict=False) if args.local_sdk else None - raw_release = requested_release((requested_project_dir, project_dir), args.release) - local_release = normalize_local_release(raw_release) or LIVE_RELEASE local_builds: list[LocalSdkBuild] | None = None if local_sdk_path: + local_release = local_sdk_requested_release(args.release) local_builds = select_local_sdk_builds( local_sdk_path, local_release, |