Initial Sailfish devel MCP

1 files changed, 115 insertions, 0 deletions

diff --git a/src/sailfish_devel_mcp/runner.py b/src/sailfish_devel_mcp/runner.py
new file mode 100644
index 0000000..b0b88bf
--- /dev/null
+++ b/src/sailfish_devel_mcp/runner.py
@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+import shlex
+import subprocess
+from typing import Iterable, Sequence
+
+from .config import DeviceConfig
+
+
+@dataclass(frozen=True)
+class CommandResult:
+    argv: tuple[str, ...]
+    returncode: int
+    stdout: str
+    stderr: str
+
+    @property
+    def ok(self) -> bool:
+        return self.returncode == 0
+
+    def public_dict(self, limit: int = 20000) -> dict[str, object]:
+        stdout, stdout_truncated = truncate(self.stdout, limit)
+        stderr, stderr_truncated = truncate(self.stderr, limit)
+        return {
+            "argv": list(self.argv),
+            "returncode": self.returncode,
+            "stdout": stdout,
+            "stderr": stderr,
+            "stdout_truncated": stdout_truncated,
+            "stderr_truncated": stderr_truncated,
+        }
+
+
+def truncate(text: str, limit: int) -> tuple[str, bool]:
+    if len(text) <= limit:
+        return text, False
+    return text[:limit] + f"\n[truncated after {limit} characters]", True
+
+
+def run(
+    argv: Sequence[str],
+    *,
+    cwd: str | Path | None = None,
+    timeout: int = 60,
+) -> CommandResult:
+    try:
+        completed = subprocess.run(
+            list(argv),
+            cwd=str(cwd) if cwd is not None else None,
+            text=True,
+            capture_output=True,
+            timeout=timeout,
+            check=False,
+        )
+        return CommandResult(
+            argv=tuple(str(arg) for arg in argv),
+            returncode=completed.returncode,
+            stdout=completed.stdout,
+            stderr=completed.stderr,
+        )
+    except subprocess.TimeoutExpired as exc:
+        return CommandResult(
+            argv=tuple(str(arg) for arg in argv),
+            returncode=124,
+            stdout=exc.stdout or "",
+            stderr=(exc.stderr or "") + f"\ncommand timed out after {timeout}s",
+        )
+
+
+def ssh_argv(device: DeviceConfig, ssh_config: Path | None, remote: str) -> list[str]:
+    argv = ["ssh"]
+    if ssh_config:
+        argv += ["-F", str(ssh_config)]
+    argv += [device.ssh_target, remote]
+    return argv
+
+
+def scp_to_argv(
+    device: DeviceConfig,
+    ssh_config: Path | None,
+    local_path: Path,
+    remote_path: str,
+) -> list[str]:
+    argv = ["scp"]
+    if ssh_config:
+        argv += ["-F", str(ssh_config)]
+    argv += [str(local_path), f"{device.ssh_target}:{remote_path}"]
+    return argv
+
+
+def scp_from_argv(
+    device: DeviceConfig,
+    ssh_config: Path | None,
+    remote_path: str,
+    local_path: Path,
+) -> list[str]:
+    argv = ["scp"]
+    if ssh_config:
+        argv += ["-F", str(ssh_config)]
+    argv += [f"{device.ssh_target}:{remote_path}", str(local_path)]
+    return argv
+
+
+def remote_command(argv: Iterable[str]) -> str:
+    return shlex.join([str(arg) for arg in argv])
+
+
+def user_bus_env(device: DeviceConfig) -> list[str]:
+    return [
+        "env",
+        f"XDG_RUNTIME_DIR={device.user_bus_runtime_dir}",
+        f"DBUS_SESSION_BUS_ADDRESS={device.user_bus_address}",
+    ]