plezy/scripts/maestro/run_maestro.py

#!/usr/bin/env python3
"""Build the Jellyfin fixture and run Plezy's Android Maestro suites."""

from __future__ import annotations

import argparse
from dataclasses import dataclass
import os
from pathlib import Path
import shlex
import signal
import shutil
import subprocess
import sys
import time
from typing import Mapping, Optional, Sequence, TextIO
import urllib.error
import urllib.request


ROOT_DIR = Path(__file__).resolve().parents[2]
APP_ID = "com.edde746.plezy"
FAULTS = ("music-failure", "offline", "recovery")
ANIMATION_SCALES = (
    "window_animation_scale",
    "transition_animation_scale",
    "animator_duration_scale",
)


class RunnerError(RuntimeError):
    """A user-actionable runner failure."""

class RunnerSignal(Exception):
    def __init__(self, signum: int) -> None:
        super().__init__(f"Interrupted by signal {signum}")
        self.exit_status = 128 + signum


def _raise_signal(signum: int, _frame: object) -> None:
    raise RunnerSignal(signum)




@dataclass(frozen=True)
class SuitePreset:
    flow_target: str
    maestro_config: Optional[str]
    jellyfin_log: str
    diagnostics_dir: str
    use_adb_reverse: bool = False
    uninstall_before_install: bool = False


SUITES = {
    "basic": SuitePreset(
        flow_target=".maestro",
        maestro_config=None,
        jellyfin_log="build/maestro/jellyfin.log",
        diagnostics_dir="build/maestro/diagnostics",
    ),
    "catalog": SuitePreset(
        flow_target=".maestro/real_flows",
        maestro_config=None,
        jellyfin_log="build/maestro-real-jellyfin/jellyfin.log",
        diagnostics_dir="build/maestro-real-jellyfin/diagnostics",
        uninstall_before_install=True,
    ),
    "media": SuitePreset(
        flow_target=".maestro/media_flows",
        maestro_config=".maestro/media-config.yaml",
        jellyfin_log="build/maestro-media/jellyfin.log",
        diagnostics_dir="build/maestro/diagnostics",
        use_adb_reverse=True,
        uninstall_before_install=True,
    ),
}


@dataclass(frozen=True)
class RunnerConfig:
    command: str
    jellyfin_host: str
    jellyfin_port: int
    proxy_port: int
    jellyfin_image: str
    skip_jellyfin: bool
    skip_jellyfin_build: bool
    skip_build: bool
    jellyfin_fault: Optional[str]
    use_adb_reverse: bool
    device_id: Optional[str]
    apk_path: Path
    flow_target: Path
    maestro_config: Optional[Path]
    uninstall_before_install: bool
    diagnostics_dir: Path
    jellyfin_log: Path
    proxy_log: Path
    proxy_journal: Path
    host_jellyfin_url: str
    jellyfin_url: Optional[str]
    jellyfin_build_attempts: int


def _positive_int(value: str) -> int:
    try:
        parsed = int(value)
    except ValueError as error:
        raise argparse.ArgumentTypeError("must be a positive integer") from error
    if parsed < 1:
        raise argparse.ArgumentTypeError("must be a positive integer")
    return parsed


def _add_bool_argument(parser: argparse.ArgumentParser, name: str, *, destination: str) -> None:
    group = parser.add_mutually_exclusive_group()
    group.add_argument(f"--{name}", dest=destination, action="store_true")
    group.add_argument(f"--no-{name}", dest=destination, action="store_false")
    parser.set_defaults(**{destination: None})


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "command",
        choices=(*SUITES, "build-image"),
        nargs="?",
        default="basic",
        help="suite to run, or build-image to only build the Jellyfin fixture",
    )
    parser.add_argument("--device", dest="device_id")
    parser.add_argument("--apk", dest="apk_path")
    parser.add_argument("--flow", dest="flow_target")
    parser.add_argument("--config", dest="maestro_config")
    parser.add_argument("--fault", choices=FAULTS, dest="jellyfin_fault")
    parser.add_argument("--jellyfin-host")
    parser.add_argument("--jellyfin-port", type=int)
    parser.add_argument("--proxy-port", type=int)
    parser.add_argument("--jellyfin-image")
    parser.add_argument("--diagnostics-dir")
    parser.add_argument("--jellyfin-log")
    parser.add_argument("--proxy-log")
    parser.add_argument("--proxy-journal")
    parser.add_argument("--host-jellyfin-url")
    parser.add_argument("--jellyfin-url")
    parser.add_argument("--jellyfin-build-attempts", type=_positive_int)
    _add_bool_argument(parser, "skip-jellyfin", destination="skip_jellyfin")
    _add_bool_argument(parser, "skip-jellyfin-build", destination="skip_jellyfin_build")
    _add_bool_argument(parser, "skip-build", destination="skip_build")
    _add_bool_argument(parser, "adb-reverse", destination="use_adb_reverse")
    _add_bool_argument(parser, "uninstall-before-install", destination="uninstall_before_install")
    return parser


def _env_bool(environment: Mapping[str, str], name: str, default: bool) -> bool:
    value = environment.get(name)
    if value is None:
        return default
    normalized = value.strip().lower()
    if normalized in {"1", "true", "yes", "on"}:
        return True
    if normalized in {"0", "false", "no", "off"}:
        return False
    raise RunnerError(f"{name} must be one of 1, 0, true, false, yes, no, on, or off")


def _option(cli_value: object, environment: Mapping[str, str], name: str, default: object) -> object:
    if cli_value is not None:
        return cli_value
    return environment.get(name, default)
def _int_option(
    cli_value: Optional[int],
    environment: Mapping[str, str],
    name: str,
    default: int,
) -> int:
    value = _option(cli_value, environment, name, default)
    try:
        parsed = int(value)
    except (TypeError, ValueError) as error:
        raise RunnerError(f"{name} must be a positive integer") from error
    if parsed < 1:
        raise RunnerError(f"{name} must be a positive integer")
    return parsed




def _bool_option(
    cli_value: Optional[bool],
    environment: Mapping[str, str],
    name: str,
    default: bool,
) -> bool:
    if cli_value is not None:
        return cli_value
    return _env_bool(environment, name, default)


def _root_path(value: object) -> Path:
    path = Path(str(value))
    return path if path.is_absolute() else ROOT_DIR / path


def parse_config(argv: Optional[Sequence[str]] = None, environment: Optional[Mapping[str, str]] = None) -> RunnerConfig:
    args = _parser().parse_args(argv)
    env = os.environ if environment is None else environment
    preset = SUITES.get(args.command, SUITES["basic"])

    jellyfin_host = str(_option(args.jellyfin_host, env, "MAESTRO_JELLYFIN_HOST", "127.0.0.1"))
    jellyfin_port = _int_option(args.jellyfin_port, env, "MAESTRO_JELLYFIN_PORT", 8096)
    proxy_port = _int_option(args.proxy_port, env, "MAESTRO_JELLYFIN_PROXY_PORT", jellyfin_port + 1)
    config_value = _option(args.maestro_config, env, "MAESTRO_CONFIG", preset.maestro_config or "")
    fault_value = _option(args.jellyfin_fault, env, "MAESTRO_JELLYFIN_FAULT", "")
    fault = str(fault_value) or None
    if fault is not None and fault not in FAULTS:
        raise RunnerError(f"Unsupported MAESTRO_JELLYFIN_FAULT: {fault}")

    build_attempts = _int_option(
        args.jellyfin_build_attempts,
        env,
        "MAESTRO_JELLYFIN_BUILD_ATTEMPTS",
        2,
    )

    return RunnerConfig(
        command=args.command,
        jellyfin_host=jellyfin_host,
        jellyfin_port=jellyfin_port,
        proxy_port=proxy_port,
        jellyfin_image=str(
            _option(args.jellyfin_image, env, "MAESTRO_JELLYFIN_IMAGE", "plezy-jellyfin-demo:local")
        ),
        skip_jellyfin=_bool_option(args.skip_jellyfin, env, "MAESTRO_SKIP_JELLYFIN", False),
        skip_jellyfin_build=_bool_option(
            args.skip_jellyfin_build,
            env,
            "MAESTRO_SKIP_JELLYFIN_BUILD",
            False,
        ),
        skip_build=_bool_option(args.skip_build, env, "MAESTRO_SKIP_BUILD", False),
        jellyfin_fault=fault,
        use_adb_reverse=_bool_option(
            args.use_adb_reverse,
            env,
            "MAESTRO_USE_ADB_REVERSE",
            preset.use_adb_reverse,
        ),
        device_id=str(_option(args.device_id, env, "MAESTRO_DEVICE_ID", "")) or None,
        apk_path=_root_path(
            _option(
                args.apk_path,
                env,
                "MAESTRO_APK_PATH",
                "build/app/outputs/flutter-apk/app-debug.apk",
            )
        ),
        flow_target=_root_path(
            _option(args.flow_target, env, "MAESTRO_FLOW_TARGET", preset.flow_target)
        ),
        maestro_config=_root_path(config_value) if config_value else None,
        uninstall_before_install=_bool_option(
            args.uninstall_before_install,
            env,
            "MAESTRO_UNINSTALL_BEFORE_INSTALL",
            preset.uninstall_before_install,
        ),
        diagnostics_dir=_root_path(
            _option(args.diagnostics_dir, env, "MAESTRO_DIAGNOSTICS_DIR", preset.diagnostics_dir)
        ),
        jellyfin_log=_root_path(
            _option(args.jellyfin_log, env, "MAESTRO_JELLYFIN_LOG", preset.jellyfin_log)
        ),
        proxy_log=_root_path(
            _option(
                args.proxy_log,
                env,
                "MAESTRO_JELLYFIN_PROXY_LOG",
                "build/maestro/jellyfin-proxy.log",
            )
        ),
        proxy_journal=_root_path(
            _option(
                args.proxy_journal,
                env,
                "MAESTRO_JELLYFIN_PROXY_JOURNAL",
                "build/maestro/jellyfin-proxy-journal.jsonl",
            )
        ),
        host_jellyfin_url=str(
            _option(
                args.host_jellyfin_url,
                env,
                "MAESTRO_JELLYFIN_HOST_URL",
                f"http://{jellyfin_host}:{jellyfin_port}",
            )
        ),
        jellyfin_url=str(_option(args.jellyfin_url, env, "MAESTRO_JELLYFIN_URL", "")) or None,
        jellyfin_build_attempts=build_attempts,
    )


def _format_command(command: Sequence[object]) -> str:
    values = [str(value) for value in command]
    if os.name == "nt":
        return subprocess.list2cmdline(values)
    return shlex.join(values)


def _run_checked(command: Sequence[object], **kwargs: object) -> subprocess.CompletedProcess[str]:
    values = [str(value) for value in command]
    print(f"+ {_format_command(values)}", flush=True)
    return subprocess.run(values, cwd=ROOT_DIR, check=True, text=True, **kwargs)


def _require_commands(names: Sequence[str]) -> None:
    missing = [name for name in names if shutil.which(name) is None]
    if missing:
        raise RunnerError(f"Required command not found: {', '.join(missing)}")


def flutter_build_command() -> tuple[str, ...]:
    return (
        "flutter",
        "build",
        "apk",
        "--debug",
        "--dart-define=PLEZY_MAESTRO_E2E=true",
    )


def build_jellyfin_image(config: RunnerConfig) -> None:
    _require_commands(("docker",))
    command = (
        "docker",
        "build",
        "--file",
        ROOT_DIR / ".maestro/jellyfin-demo/Dockerfile",
        "--tag",
        config.jellyfin_image,
        ROOT_DIR,
    )
    for attempt in range(1, config.jellyfin_build_attempts + 1):
        try:
            _run_checked(command)
            return
        except subprocess.CalledProcessError:
            if attempt == config.jellyfin_build_attempts:
                raise RunnerError(
                    f"Jellyfin image build failed after {config.jellyfin_build_attempts} attempts"
                )
            print(f"Jellyfin image build attempt {attempt} failed; retrying", file=sys.stderr)
            time.sleep(5)


class MaestroRunner:
    def __init__(self, config: RunnerConfig) -> None:
        self.config = config
        self.device_id = config.device_id
        self.container_name: Optional[str] = None
        self.proxy_process: Optional[subprocess.Popen[str]] = None
        self.proxy_output: Optional[TextIO] = None
        self.reverse_configured = False
        self.device_service_port = config.jellyfin_port
        self.host_jellyfin_url = config.host_jellyfin_url
        self.device_settings: dict[tuple[str, str], str] = {}

    @property
    def adb_prefix(self) -> list[str]:
        prefix = ["adb"]
        if self.device_id:
            prefix.extend(("-s", self.device_id))
        return prefix

    def run(self) -> None:
        required = ["adb", "maestro"]
        if not self.config.skip_build:
            required.append("flutter")
        if not self.config.skip_jellyfin:
            required.append("docker")
        _require_commands(required)
        self._select_device()
        self._prepare_output_directories()

        if not self.config.skip_jellyfin:
            if not self.config.skip_jellyfin_build:
                build_jellyfin_image(self.config)
            self._start_jellyfin()
        self._wait_for_health(self.host_jellyfin_url, attempts=180, interval=1, service="Jellyfin")

        if self.config.jellyfin_fault:
            self._start_proxy()

        if not self.config.skip_build:
            _run_checked(("flutter", "pub", "get"))
            _run_checked(flutter_build_command())

        self._prepare_device()
        _run_checked((*self.adb_prefix, "install", "-r", self.config.apk_path))
        _run_checked(self.maestro_command())

    def maestro_command(self) -> list[str]:
        default_url = f"http://10.0.2.2:{self.device_service_port}"
        if self.config.use_adb_reverse:
            default_url = f"http://127.0.0.1:{self.device_service_port}"
        jellyfin_url = self.config.jellyfin_url or default_url
        command = ["maestro", "test", "-e", f"JELLYFIN_URL={jellyfin_url}"]
        if self.config.jellyfin_fault == "offline":
            command.extend(("-e", f"JELLYFIN_CONTROL_URL={self.host_jellyfin_url}"))
        if self.device_id:
            command.extend(("--device", self.device_id))
        if self.config.maestro_config:
            command.extend(("--config", str(self.config.maestro_config)))
        command.append(str(self.config.flow_target))
        return command

    def _prepare_output_directories(self) -> None:
        for path in (
            self.config.diagnostics_dir,
            self.config.jellyfin_log.parent,
            self.config.proxy_log.parent,
            self.config.proxy_journal.parent,
        ):
            path.mkdir(parents=True, exist_ok=True)

    def _select_device(self) -> None:
        if not self.config.use_adb_reverse or self.device_id:
            return
        result = subprocess.run(
            ("adb", "devices"),
            cwd=ROOT_DIR,
            check=True,
            capture_output=True,
            text=True,
        )
        devices = []
        for line in result.stdout.splitlines()[1:]:
            fields = line.split()
            if len(fields) >= 2 and fields[1] == "device":
                devices.append(fields[0])
        if len(devices) != 1:
            raise RunnerError("Set --device to the Android device serial when using --adb-reverse")
        self.device_id = devices[0]

    def _start_jellyfin(self) -> None:
        name = f"plezy-maestro-jellyfin-{self.config.jellyfin_port}-{os.getpid()}"
        result = _run_checked(
            (
                "docker",
                "run",
                "--detach",
                "--rm",
                "--name",
                name,
                "--publish",
                f"{self.config.jellyfin_host}:{self.config.jellyfin_port}:8096",
                self.config.jellyfin_image,
            ),
            capture_output=True,
        )
        self.container_name = result.stdout.strip() or name

    def _wait_for_health(self, base_url: str, *, attempts: int, interval: float, service: str) -> None:
        health_url = f"{base_url.rstrip('/')}/health"
        last_error: Optional[Exception] = None
        for _ in range(attempts):
            try:
                with urllib.request.urlopen(health_url, timeout=1) as response:
                    status = response.read().decode(errors="replace").strip()
                if status.casefold() == "healthy":
                    return
                last_error = RunnerError(f"{service} health status is {status or 'empty'}")
            except (OSError, urllib.error.URLError) as error:
                last_error = error
            time.sleep(interval)
        raise RunnerError(f"{service} did not become ready at {base_url}: {last_error}")

    def _start_proxy(self) -> None:
        self.proxy_output = self.config.proxy_log.open("w", encoding="utf-8")
        self.proxy_process = subprocess.Popen(
            (
                sys.executable,
                str(ROOT_DIR / "scripts/maestro/maestro_jellyfin_proxy.py"),
                "--host",
                self.config.jellyfin_host,
                "--port",
                str(self.config.proxy_port),
                "--upstream",
                self.host_jellyfin_url,
                "--fault",
                self.config.jellyfin_fault or "",
                "--journal",
                str(self.config.proxy_journal),
            ),
            cwd=ROOT_DIR,
            stdout=self.proxy_output,
            stderr=subprocess.STDOUT,
            text=True,
        )
        proxy_url = f"http://{self.config.jellyfin_host}:{self.config.proxy_port}"
        for _ in range(50):
            if self.proxy_process.poll() is not None:
                raise RunnerError(f"Jellyfin fault proxy exited early; see {self.config.proxy_log}")
            try:
                self._wait_for_health(proxy_url, attempts=1, interval=0, service="Jellyfin fault proxy")
                self.host_jellyfin_url = proxy_url
                self.device_service_port = self.config.proxy_port
                return
            except RunnerError:
                time.sleep(0.1)
        raise RunnerError(f"Jellyfin fault proxy did not become ready; see {self.config.proxy_log}")

    def _adb_run(
        self,
        *arguments: object,
        check: bool = True,
        quiet: bool = False,
        timeout: int = 30,
    ) -> subprocess.CompletedProcess[str]:
        kwargs: dict[str, object] = {}
        if quiet:
            kwargs.update(stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return subprocess.run(
            [*self.adb_prefix, *(str(argument) for argument in arguments)],
            cwd=ROOT_DIR,
            check=check,
            text=True,
            timeout=timeout,
            **kwargs,
        )

    def _adb_capture(self, *arguments: object) -> Optional[str]:
        command = [*self.adb_prefix, *(str(argument) for argument in arguments)]
        try:
            result = subprocess.run(
                command,
                cwd=ROOT_DIR,
                check=False,
                capture_output=True,
                text=True,
                timeout=10,
            )
        except subprocess.TimeoutExpired as error:
            raise RunnerError(f"ADB command timed out: {_format_command(command)}") from error
        return result.stdout.strip() if result.returncode == 0 else None

    def _prepare_device(self) -> None:
        _run_checked((*self.adb_prefix, "wait-for-device"), timeout=60)
        for namespace, key in (
            ("global", "stay_on_while_plugged_in"),
            ("secure", "immersive_mode_confirmations"),
            ("global", "hide_error_dialogs"),
            *((("global", key) for key in ANIMATION_SCALES)),
        ):
            value = self._adb_capture("shell", "settings", "get", namespace, key)
            if value is not None:
                self.device_settings[(namespace, key)] = value

        self._adb_run(
            "shell",
            "settings",
            "put",
            "secure",
            "immersive_mode_confirmations",
            "confirmed",
            check=False,
            quiet=True,
        )
        self._adb_run(
            "shell",
            "settings",
            "put",
            "global",
            "hide_error_dialogs",
            "1",
            check=False,
            quiet=True,
        )
        # Disable animations to avoid Maestro's per-action transition waits;
        # cleanup() restores the captured settings.
        for key in ANIMATION_SCALES:
            self._adb_run("shell", "settings", "put", "global", key, "0", check=False, quiet=True)
        self._adb_run("shell", "input", "keyevent", "KEYCODE_BACK", check=False, quiet=True)
        _run_checked((*self.adb_prefix, "shell", "svc", "power", "stayon", "true"))
        _run_checked((*self.adb_prefix, "shell", "input", "keyevent", "KEYCODE_WAKEUP"))
        _run_checked((*self.adb_prefix, "shell", "wm", "dismiss-keyguard"))

        if self.config.use_adb_reverse:
            _run_checked(
                (
                    *self.adb_prefix,
                    "reverse",
                    f"tcp:{self.device_service_port}",
                    f"tcp:{self.device_service_port}",
                )
            )
            self.reverse_configured = True
        if self.config.uninstall_before_install:
            self._adb_run("uninstall", APP_ID, check=False, quiet=True)

    def collect_failure_diagnostics(self, exit_status: int) -> None:
        try:
            self.config.diagnostics_dir.mkdir(parents=True, exist_ok=True)
            state_path = self.config.diagnostics_dir / "run-state.txt"
            with state_path.open("w", encoding="utf-8") as output:
                output.write(f"exit_status={exit_status}\n")
                output.write(f"jellyfin_host_url={self.host_jellyfin_url}\n")
                output.write(f"jellyfin_container={self.container_name or ''}\n")
                output.write(f"jellyfin_fault={self.config.jellyfin_fault or ''}\n")
                output.write(f"proxy_pid={self.proxy_process.pid if self.proxy_process else ''}\n")
                try:
                    with urllib.request.urlopen(
                        f"{self.host_jellyfin_url.rstrip('/')}/health",
                        timeout=2,
                    ) as response:
                        output.write(response.read().decode(errors="replace"))
                        output.write("\n")
                except Exception as error:  # Keep diagnostics from masking the failure.
                    output.write(f"health_error={error}\n")

            if self.container_name:
                self._write_command_output(
                    ("docker", "logs", self.container_name),
                    self.config.jellyfin_log,
                )
            self._write_command_output(("adb", "devices", "-l"), self.config.diagnostics_dir / "adb-devices.txt")
            for filename, arguments in (
                ("device-properties.txt", ("shell", "getprop")),
                ("device-processes.txt", ("shell", "ps", "-A")),
                ("device-activities.txt", ("shell", "dumpsys", "activity", "activities")),
                ("device-windows.txt", ("shell", "dumpsys", "window", "windows")),
                ("device-logcat.txt", ("logcat", "-d", "-v", "threadtime")),
            ):
                self._write_command_output(
                    (*self.adb_prefix, *arguments),
                    self.config.diagnostics_dir / filename,
                )
            if os.name == "nt" and shutil.which("tasklist"):
                self._write_command_output(("tasklist",), self.config.diagnostics_dir / "host-processes.txt")
            elif shutil.which("ps"):
                self._write_command_output(("ps", "-ef"), self.config.diagnostics_dir / "host-processes.txt")
            if shutil.which("lsof"):
                self._write_command_output(
                    ("lsof", "-nP", f"-iTCP:{self.device_service_port}"),
                    self.config.diagnostics_dir / "jellyfin-listeners.txt",
                )
        except Exception as error:  # Diagnostics are best effort.
            print(f"Failed to collect Maestro diagnostics: {error}", file=sys.stderr)

    def _write_command_output(self, command: Sequence[object], path: Path) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("w", encoding="utf-8") as output:
            subprocess.run(
                [str(value) for value in command],
                cwd=ROOT_DIR,
                check=False,
                stdout=output,
                stderr=subprocess.STDOUT,
                text=True,
                timeout=15,
            )

    def _adb_best_effort(self, *arguments: object) -> None:
        try:
            self._adb_run(*arguments, check=False, quiet=True, timeout=5)
        except (OSError, subprocess.SubprocessError):
            pass

    def cleanup(self) -> None:
        for (namespace, key), value in self.device_settings.items():
            operation = "delete" if value == "null" else "put"
            arguments = ["shell", "settings", operation, namespace, key]
            if operation == "put":
                arguments.append(value)
            self._adb_best_effort(*arguments)

        if self.reverse_configured:
            self._adb_best_effort(
                "reverse",
                "--remove",
                f"tcp:{self.device_service_port}",
            )
        try:
            if self.proxy_process:
                self.proxy_process.terminate()
                try:
                    self.proxy_process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    self.proxy_process.kill()
                    self.proxy_process.wait()
        finally:
            if self.proxy_output:
                self.proxy_output.close()

        if self.container_name:
            try:
                self._write_command_output(
                    ("docker", "logs", self.container_name),
                    self.config.jellyfin_log,
                )
            except (OSError, subprocess.SubprocessError):
                pass
            try:
                subprocess.run(
                    ("docker", "stop", "--time", "15", self.container_name),
                    cwd=ROOT_DIR,
                    check=False,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    text=True,
                    timeout=30,
                )
            except (OSError, subprocess.SubprocessError):
                pass


def run(config: RunnerConfig) -> int:
    if config.command == "build-image":
        build_jellyfin_image(config)
        return 0

    runner = MaestroRunner(config)
    exit_status = 0
    try:
        runner.run()
    except KeyboardInterrupt:
        exit_status = 130
        print("Maestro run interrupted", file=sys.stderr)
    except RunnerSignal as error:
        exit_status = error.exit_status
        print(str(error), file=sys.stderr)
    except subprocess.CalledProcessError as error:
        exit_status = error.returncode or 1
        print(f"Command failed ({exit_status}): {_format_command(error.cmd)}", file=sys.stderr)
    except (OSError, RunnerError) as error:
        exit_status = 1
        print(error, file=sys.stderr)
    finally:
        if exit_status:
            runner.collect_failure_diagnostics(exit_status)
        runner.cleanup()
    return exit_status


def main(argv: Optional[Sequence[str]] = None) -> int:
    signal.signal(signal.SIGINT, _raise_signal)
    signal.signal(signal.SIGTERM, _raise_signal)
    try:
        config = parse_config(argv)
        return run(config)
    except RunnerSignal as error:
        return error.exit_status
    except KeyboardInterrupt:
        return 130
    except (OSError, RunnerError, subprocess.CalledProcessError) as error:
        print(error, file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())