from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from pathlib import Path, PurePosixPath
import re
import shlex
import shutil
from typing import Any, Callable

from .config import Config, DeviceConfig
from .runner import (
    CommandResult,
    remote_command,
    run,
    scp_from_argv,
    scp_to_argv,
    ssh_argv,
    truncate,
    user_bus_env,
)


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_user_bus_call(),
            lambda args: handle_device_user_bus_call(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_build_rpm(), lambda args: handle_build_rpm(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.utcnow().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<N>")

    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_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_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_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 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)
    return command_result("build Sailfish RPM", run(command, timeout=timeout))


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)
    command = ["osc"]
    if api_alias:
        command += ["-A", api_alias]
    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 _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 _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 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 _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 = [config.paths.git_root.resolve(strict=False)]
    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 = [config.paths.git_root.resolve(strict=False), 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 _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/<pid>/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-<timestamp>.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_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_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_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.",
        "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"}},
                "timeout": _timeout_prop(3600),
            },
            ["project_path"],
        ),
        "annotations": _mutating_annotations("Build Sailfish RPM"),
    }


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 remote build log with osc.",
        "inputSchema": _object_schema(
            {
                "project": {"type": "string"},
                "package": {"type": "string"},
                "repository": {"type": "string"},
                "arch": {"type": "string"},
                "api_alias": {"type": "string"},
                "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"),
    }