#!/usr/bin/env python3
"""ADB screen observer and hybrid OCR/template runner for BlueStacks Air.

Actions are gated by visible state: macOS Vision OCR for stable labels and
OpenCV templates for graphical controls or fast fallbacks.
"""

from __future__ import annotations

import argparse
import contextlib
import difflib
import fcntl
import json
import os
import re
import signal
import shutil
import subprocess
import sys
import tempfile
import time
import unicodedata
import zipfile
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

try:
    import cv2  # type: ignore
    import numpy as np  # type: ignore
except ImportError as exc:  # pragma: no cover - exercised by dependency check
    cv2 = None
    np = None
    CV_IMPORT_ERROR = exc
else:
    CV_IMPORT_ERROR = None


DEFAULT_CONF = Path("/Users/Shared/Library/Application Support/BlueStacks/bluestacks.conf")
DEFAULT_ENGINE_ROOT = Path("/Users/Shared/Library/Application Support/BlueStacks/Engine")
DEFAULT_MIM_METADATA = DEFAULT_ENGINE_ROOT / "UserData" / "MimMetaData.json"
DEFAULT_PACKAGE = "com.garena.game.kgtw"
DEFAULT_PLAYER = Path("/Applications/BlueStacks.app/Contents/MacOS/BlueStacks")
DEFAULT_MIM = Path(
    "/Applications/BlueStacks Air multi-instance manager.app/Contents/MacOS/"
    "BlueStacks Air multi-instance manager"
)
SKILL_DIR = Path(__file__).resolve().parents[1]
DEFAULT_WORKFLOW = SKILL_DIR / "references" / "workflow.json"
RESULT_SCREENSHOTS = (
    "01_activity_rewards.png",
    "02_lsr_box_rewards.png",
    "03_41_draw_rewards.png",
)
VISION_OCR_SOURCE = SKILL_DIR / "scripts" / "vision_ocr.swift"
VISION_OCR_BINARY = Path(tempfile.gettempdir()) / "aov-reroll-vision-ocr"
BLUESTACKS_OVERLAY_SOURCE = SKILL_DIR / "scripts" / "bluestacks_overlay.swift"
BLUESTACKS_OVERLAY_BINARY = (
    Path(tempfile.gettempdir()) / "aov-reroll-bluestacks-overlay"
)
DEFAULT_LIFECYCLE_LOCK = (
    Path(tempfile.gettempdir()) / "aov-bluestacks-air-lifecycle.lock"
)


@dataclass(frozen=True)
class Instance:
    engine: str
    display_name: str
    adb_port: int
    running: bool

    @property
    def serial(self) -> str:
        return f"127.0.0.1:{self.adb_port}"


def require_cv() -> None:
    if cv2 is None or np is None:
        raise SystemExit(
            "OpenCV dependencies are missing. Install numpy and "
            "opencv-python-headless in an isolated environment. "
            f"Original error: {CV_IMPORT_ERROR}"
        )


def adb_binary() -> str:
    candidate = shutil.which("adb")
    if candidate:
        return candidate
    bundled = "/Applications/BlueStacks.app/Contents/MacOS/hd-adb"
    if Path(bundled).exists():
        return bundled
    raise SystemExit("adb was not found on PATH or inside BlueStacks.app")


def run_process(args: list[str], *, binary: bool = False, timeout: float = 30) -> Any:
    return subprocess.run(
        args,
        check=True,
        capture_output=True,
        text=not binary,
        timeout=timeout,
    ).stdout


def running_engines() -> set[str]:
    output = run_process(["ps", "-ax", "-o", "command="])
    return set(re.findall(r"BlueStacks --instance ([A-Za-z0-9_-]+)", output))


def read_instances(conf_path: Path = DEFAULT_CONF) -> list[Instance]:
    if not conf_path.exists():
        raise SystemExit(f"BlueStacks configuration not found: {conf_path}")
    values: dict[str, dict[str, str]] = {}
    pattern = re.compile(r'^bst\.instance\.([^.]+)\.([^=]+)="(.*)"$')
    for line in conf_path.read_text(encoding="utf-8", errors="replace").splitlines():
        match = pattern.match(line)
        if match:
            engine, key, value = match.groups()
            values.setdefault(engine, {})[key] = value
    active = running_engines()
    instances: list[Instance] = []
    for engine, fields in values.items():
        port = fields.get("adb_port")
        if not port or not port.isdigit():
            continue
        instances.append(
            Instance(
                engine=engine,
                display_name=fields.get("display_name", engine),
                adb_port=int(port),
                running=engine in active,
            )
        )
    return sorted(instances, key=lambda item: item.engine)


def instance_by_engine(engine: str) -> Instance:
    matches = [item for item in read_instances() if item.engine == engine]
    if len(matches) != 1:
        raise RuntimeError(f"BlueStacks instance {engine!r} is not present in the configuration")
    return matches[0]


def resolve_instance(name: str | None) -> Instance:
    instances = read_instances()
    if name:
        matches = [item for item in instances if name in {item.engine, item.display_name}]
        if len(matches) != 1:
            choices = ", ".join(f"{x.engine} ({x.display_name})" for x in instances)
            raise SystemExit(f"Expected one exact instance named {name!r}. Available: {choices}")
        return matches[0]
    running = [item for item in instances if item.running]
    if len(running) != 1:
        choices = ", ".join(f"{x.engine} ({x.display_name})" for x in running) or "none"
        raise SystemExit(f"Specify --instance; running instances: {choices}")
    return running[0]


def adb(instance: Instance, *args: str, binary: bool = False, timeout: float = 30) -> Any:
    return run_process([adb_binary(), "-s", instance.serial, *args], binary=binary, timeout=timeout)


def connect(instance: Instance) -> None:
    run_process([adb_binary(), "connect", instance.serial])


def available_gib(path: Path) -> float:
    probe = path
    while not probe.exists() and probe != probe.parent:
        probe = probe.parent
    return shutil.disk_usage(probe).free / (1024**3)


def command_runs_executable(command: str, executable: Path) -> bool:
    """Match one ps command to an executable with or without arguments."""
    normalized = command.strip()
    executable_text = str(executable)
    if normalized == executable_text:
        return True
    return normalized.startswith(executable_text) and normalized[
        len(executable_text) : len(executable_text) + 1
    ].isspace()


def mim_running() -> bool:
    return bool(mim_pids())


def mim_pids() -> list[int]:
    output = run_process(["ps", "-ax", "-o", "pid=,command="])
    pids: list[int] = []
    for line in output.splitlines():
        fields = line.strip().split(maxsplit=1)
        if len(fields) != 2 or not fields[0].isdigit():
            continue
        if command_runs_executable(fields[1], DEFAULT_MIM):
            pids.append(int(fields[0]))
    return pids


def mim_window_ready() -> bool:
    completed = subprocess.run(
        [
            "osascript",
            "-e",
            'tell application "System Events" to tell process '
            '"BlueStacks Air multi-instance manager" to get name of every window',
        ],
        capture_output=True,
        text=True,
        timeout=10,
        check=False,
    )
    return (
        completed.returncode == 0
        and "BlueStacks Air Manager" in completed.stdout
    )


def ensure_mim_running(timeout: float = 20, *, reveal_window: bool = False) -> None:
    if not DEFAULT_MIM.is_file():
        raise SystemExit(f"BlueStacks multi-instance manager not found: {DEFAULT_MIM}")
    if not mim_running():
        if reveal_window:
            subprocess.run(
                ["open", "-a", "BlueStacks Air multi-instance manager"],
                capture_output=True,
                text=True,
                timeout=10,
                check=False,
            )
        else:
            subprocess.Popen(
                [str(DEFAULT_MIM), "--hidden"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=True,
            )
    elif reveal_window:
        subprocess.run(
            ["open", "-a", "BlueStacks Air multi-instance manager"],
            capture_output=True,
            text=True,
            timeout=10,
            check=False,
        )
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if mim_running() and (not reveal_window or mim_window_ready()):
            time.sleep(0.5)
            return
        time.sleep(0.25)
    required_state = "process and window" if reveal_window else "process"
    raise RuntimeError(
        f"BlueStacks multi-instance manager {required_state} did not become ready"
    )


def restart_mim_for_stale_ui(timeout: float = 20) -> None:
    """Restart only the Manager process, never a BlueStacks player engine."""
    stop_mim(timeout=timeout)
    ensure_mim_running(timeout=timeout, reveal_window=True)


def stop_mim(timeout: float = 20) -> None:
    """Stop only the Manager process and wait for its private IPC lock to release."""
    for pid in mim_pids():
        os.kill(pid, signal.SIGTERM)
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if not mim_pids():
            return
        time.sleep(0.25)
    raise RuntimeError("BlueStacks multi-instance manager did not stop for UI refresh")


def mim_metadata_contains_engine(
    engine: str,
    path: Path = DEFAULT_MIM_METADATA,
) -> bool:
    if not path.is_file():
        return False
    payload = json.loads(path.read_text(encoding="utf-8"))
    organization = payload.get("Organization")
    if not isinstance(organization, list):
        raise RuntimeError(f"BlueStacks Manager metadata has invalid Organization: {path}")
    return any(
        isinstance(item, dict) and item.get("InstanceName") == engine
        for item in organization
    )


def restore_mim_metadata_entry(
    instance: Instance,
    path: Path = DEFAULT_MIM_METADATA,
) -> bool:
    """Restore a missing Manager index row without touching the engine itself.

    BlueStacks 5.21 can create a valid second engine in ``bluestacks.conf`` yet
    omit it from ``MimMetaData.json``.  Its Manager then hides the exact engine,
    making safe UI deletion impossible.  The deletion caller has already proved
    the stopped instance's exact config identity; append only that one missing
    row, then let BlueStacks' own Manager perform the actual deletion.
    """
    if not path.is_file():
        raise RuntimeError(f"BlueStacks Manager metadata is missing: {path}")
    payload = json.loads(path.read_text(encoding="utf-8"))
    organization = payload.get("Organization")
    if not isinstance(organization, list) or not all(
        isinstance(item, dict) for item in organization
    ):
        raise RuntimeError(f"BlueStacks Manager metadata is malformed: {path}")
    if any(item.get("InstanceName") == instance.engine for item in organization):
        return False
    used_ids = [item.get("ID") for item in organization if isinstance(item.get("ID"), int)]
    organization.append(
        {
            "ID": max(used_ids, default=-1) + 1,
            "Name": instance.display_name,
            "IsFolder": False,
            "ParentFolder": -1,
            "IsOpen": False,
            "IsVisible": True,
            "InstanceName": instance.engine,
        }
    )
    temp_path = path.with_name(f".{path.name}.aov-{os.getpid()}.tmp")
    temp_path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=4) + "\n",
        encoding="utf-8",
    )
    os.chmod(temp_path, path.stat().st_mode)
    os.replace(temp_path, path)
    print(
        f"restored missing BlueStacks Manager metadata row: "
        f"{instance.engine} ({instance.display_name})",
        flush=True,
    )
    return True


@contextlib.contextmanager
def lifecycle_lock(path: Path | None, timeout: float = 600) -> Any:
    """Serialize BlueStacks Manager/configuration operations across processes."""
    if path is None:
        yield
        return
    path = path.expanduser().resolve()
    path.parent.mkdir(parents=True, exist_ok=True)
    handle = path.open("a+")
    deadline = time.monotonic() + timeout
    last_report = 0.0
    try:
        while True:
            try:
                fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                now = time.monotonic()
                if now >= deadline:
                    raise RuntimeError(
                        f"Timed out waiting for BlueStacks lifecycle lock: {path}"
                    )
                if now - last_report >= 15:
                    print(f"waiting for BlueStacks lifecycle lock: {path}", flush=True)
                    last_report = now
                time.sleep(1)
        handle.seek(0)
        handle.truncate()
        handle.write(f"pid={os.getpid()} acquired={datetime.now().isoformat()}\n")
        handle.flush()
        print(f"acquired BlueStacks lifecycle lock: {path}", flush=True)
        yield
    finally:
        try:
            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        finally:
            handle.close()


def validate_running_peers(allowed: set[str]) -> None:
    active = running_engines()
    unexpected = sorted(active - allowed)
    if unexpected:
        raise SystemExit(
            "fresh-run found running player engines not owned by this batch: "
            + ", ".join(unexpected)
        )


def create_fresh_instance(
    timeout: float = 90,
    *,
    cpu: int = 4,
    ram_mb: int = 4096,
) -> Instance:
    """Create a genuinely fresh engine through BlueStacks' local manager RPC."""
    before = {item.engine for item in read_instances()}
    ensure_mim_running()
    command = [
        str(DEFAULT_MIM),
        "--hidden",
        "--cmd",
        "createFreshInstance",
        "--imageName",
        "Tiramisu64",
        "--cpu",
        str(cpu),
        "--ram",
        str(ram_mb),
        "--graphicEngine",
        "aga",
        "--graphicRenderer",
        "vlcn",
        "--deviceProfile",
        "stou",
        "--abiValue",
        "8",
        "--source",
        "aov-reroll",
        "--noexit",
    ]
    # When an existing manager accepts the RPC, the short-lived sender exits with
    # BlueStacks' private code 114. The configuration delta is the source of truth.
    completed = subprocess.run(command, capture_output=True, text=True, check=False)
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        current = read_instances()
        created = [item for item in current if item.engine not in before]
        if len(created) == 1:
            print(
                f"created fresh engine: {created[0].engine} "
                f"({created[0].display_name})",
                flush=True,
            )
            return created[0]
        if len(created) > 1:
            names = ", ".join(item.engine for item in created)
            raise RuntimeError(f"More than one new BlueStacks engine appeared: {names}")
        time.sleep(0.5)
    details = (completed.stderr or completed.stdout).strip()
    raise RuntimeError(
        "BlueStacks did not create a fresh engine within the timeout"
        + (f": {details}" if details else "")
    )


def configure_instance_for_workflow(instance: Instance) -> None:
    """Set the stopped fresh engine to the resolution used by recorded templates."""
    updates = {
        f"bst.instance.{instance.engine}.fb_width": "1440",
        f"bst.instance.{instance.engine}.fb_height": "2560",
        f"bst.instance.{instance.engine}.dpi": "320",
        # All game input is sent through ADB.  Disabling BlueStacks' host-side
        # key mapping prevents its first-game keyboard/mouse tutorial from
        # pausing the Android surface behind an invisible-to-ADB modal.
        f"bst.instance.{instance.engine}.game_controls_enabled": "0",
    }
    lines = DEFAULT_CONF.read_text(encoding="utf-8", errors="strict").splitlines()
    found: set[str] = set()
    rewritten: list[str] = []
    for line in lines:
        key = line.split("=", 1)[0]
        if key in updates:
            rewritten.append(f'{key}="{updates[key]}"')
            found.add(key)
        else:
            rewritten.append(line)
    missing = sorted(set(updates) - found)
    if missing:
        raise RuntimeError(f"Fresh engine is missing configuration keys: {', '.join(missing)}")
    temp_path = DEFAULT_CONF.with_name(f".{DEFAULT_CONF.name}.aov-{os.getpid()}.tmp")
    temp_path.write_text("\n".join(rewritten) + "\n", encoding="utf-8")
    os.chmod(temp_path, DEFAULT_CONF.stat().st_mode)
    os.replace(temp_path, DEFAULT_CONF)
    print(
        f"configured {instance.engine} base framebuffer 1440x2560 at 320 DPI "
        "(game rotates it to 2560x1440 landscape)",
        flush=True,
    )


def wait_for_instance_boot(engine: str, timeout: float = 240) -> Instance:
    deadline = time.monotonic() + timeout
    last_error = ""
    while time.monotonic() < deadline:
        instance = instance_by_engine(engine)
        try:
            subprocess.run(
                [adb_binary(), "connect", instance.serial],
                capture_output=True,
                text=True,
                timeout=10,
                check=False,
            )
            state = subprocess.run(
                [adb_binary(), "-s", instance.serial, "get-state"],
                capture_output=True,
                text=True,
                timeout=10,
                check=False,
            )
            if state.returncode == 0 and state.stdout.strip() == "device":
                booted = subprocess.run(
                    [
                        adb_binary(),
                        "-s",
                        instance.serial,
                        "shell",
                        "getprop",
                        "sys.boot_completed",
                    ],
                    capture_output=True,
                    text=True,
                    timeout=10,
                    check=False,
                )
                if booted.stdout.strip() == "1":
                    print(f"ADB ready: {engine} on {instance.serial}", flush=True)
                    return instance
            last_error = (state.stderr or state.stdout).strip()
        except (OSError, subprocess.SubprocessError) as exc:
            last_error = str(exc)
        time.sleep(2)
    raise RuntimeError(
        f"BlueStacks engine {engine} did not finish booting"
        + (f": {last_error}" if last_error else "")
    )


def start_instance(instance: Instance, timeout: float = 240) -> Instance:
    if not DEFAULT_PLAYER.is_file():
        raise SystemExit(f"BlueStacks player not found: {DEFAULT_PLAYER}")
    subprocess.Popen(
        [str(DEFAULT_PLAYER), "--instance", instance.engine],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    print(f"starting engine: {instance.engine}", flush=True)
    return wait_for_instance_boot(instance.engine, timeout)


def engine_pids(engine: str) -> list[int]:
    output = run_process(["ps", "-ax", "-o", "pid=,command="])
    pattern = re.compile(
        rf"^\s*(\d+)\s+{re.escape(str(DEFAULT_PLAYER))}\s+--instance\s+"
        rf"{re.escape(engine)}(?:\s|$)"
    )
    return [int(match.group(1)) for line in output.splitlines() if (match := pattern.match(line))]


def close_instance(engine: str, timeout: float = 60) -> str:
    """Close the created player but preserve its engine and guest-account data."""
    pids = engine_pids(engine)
    if not pids:
        return "already_stopped"
    # BlueStacks handles SIGTERM through its normal aboutToQuit path: it stops the
    # VM, flushes its configuration, and releases the instance lock. Target only
    # the PID whose command line contains this exact engine name.
    method = "exact_sigterm"
    for pid in pids:
        os.kill(pid, signal.SIGTERM)
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if not engine_pids(engine):
            print(f"closed engine: {engine} ({method}); engine was not deleted", flush=True)
            return method
        time.sleep(1)
    raise RuntimeError(f"BlueStacks engine {engine} did not close within {timeout:.0f}s")


def delete_instance_with_manager(
    engine: str,
    expected_display_name: str,
    *,
    timeout: float = 180,
) -> str:
    """Delete one stopped, exact instance using BlueStacks' own Manager UI.

    The caller must separately prove ownership (the batch supervisor does so by
    taking an instance snapshot before it starts the worker).  This function
    has multiple exact-target gates and never clicks Delete if filtering leaves
    zero or more than one row.
    """
    if not re.fullmatch(r"Tiramisu64(?:_[0-9]+)?", engine):
        raise RuntimeError(f"Refusing unexpected engine name: {engine!r}")
    instance = instance_by_engine(engine)
    if instance.display_name != expected_display_name:
        raise RuntimeError(
            f"Display name changed for {engine}: expected {expected_display_name!r}, "
            f"found {instance.display_name!r}"
        )
    if engine_pids(engine):
        raise RuntimeError(f"Refusing to delete running engine: {engine}")
    if not mim_metadata_contains_engine(engine):
        # BlueStacks can leave a valid stopped instance in its config and on
        # disk while omitting the row from MimMetaData.json.  Never compensate
        # by clicking the only visible row (it may be a protected user engine).
        # Stop only Manager, restore this exact verified row, then still use
        # BlueStacks' own exact-filtered UI deletion below.
        stop_mim()
        restore_mim_metadata_entry(instance)
    ensure_mim_running(reveal_window=True)
    apple_script = r'''
on run argv
    set exactName to item 1 of argv
    tell application "System Events"
        if not (exists process "BlueStacks Air multi-instance manager") then error "Manager process is missing"
        tell process "BlueStacks Air multi-instance manager"
            set frontmost to true
            if not (exists window "BlueStacks Air Manager") then error "Manager window is missing"
            set managerWindow to window "BlueStacks Air Manager"
            if (count of text fields of managerWindow) is not 1 then error "Expected one search field"
            set value of text field 1 of managerWindow to exactName
            delay 1
            set searchPosition to position of text field 1 of managerWindow
            set rowMinimumY to (item 2 of searchPosition) + 20

            set rowCheckboxes to 0
            set filteredRowX to missing value
            set filteredRowY to missing value
            repeat with elementItem in UI elements of managerWindow
                try
                    if role of elementItem is "AXCheckBox" then
                        set elementPosition to position of elementItem
                        if item 2 of elementPosition > rowMinimumY then
                            set rowCheckboxes to rowCheckboxes + 1
                            set filteredRowX to item 1 of elementPosition
                            set filteredRowY to item 2 of elementPosition
                        end if
                    end if
                end try
            end repeat
            if rowCheckboxes is not 1 then error "Exact-name filter did not leave exactly one row"

            -- A stopped filtered row exposes a pencil near its name, followed
            -- by the right-side management cluster: settings, clone, trash,
            -- and sometimes a final collapse chevron.  The pencil can also be
            -- reported as 12x12, so counting every compact button on the row
            -- is not stable.  Restrict candidates to the right-side cluster.
            -- With three candidates trash is rightmost; with four candidates
            -- the collapse chevron is rightmost and trash is second-rightmost.
            set actionButtonCount to 0
            set rightmostButton to missing value
            set rightmostButtonX to -1
            set secondRightmostButton to missing value
            set secondRightmostButtonX to -1
            repeat with elementItem in UI elements of managerWindow
                try
                    if role of elementItem is "AXButton" then
                        set elementPosition to position of elementItem
                        set elementSize to size of elementItem
                        set verticalDistance to (item 2 of elementPosition) - filteredRowY
                        if verticalDistance < 0 then set verticalDistance to -verticalDistance
                        set buttonWidth to item 1 of elementSize
                        set buttonHeight to item 2 of elementSize
                        if buttonWidth >= 8 and buttonWidth <= 32 and buttonHeight >= 8 and buttonHeight <= 32 and verticalDistance < 18 and item 1 of elementPosition > filteredRowX + 400 then
                            set actionButtonCount to actionButtonCount + 1
                            set candidateX to item 1 of elementPosition
                            if candidateX > rightmostButtonX then
                                set secondRightmostButton to rightmostButton
                                set secondRightmostButtonX to rightmostButtonX
                                set rightmostButton to contents of elementItem
                                set rightmostButtonX to candidateX
                            else if candidateX > secondRightmostButtonX then
                                set secondRightmostButton to contents of elementItem
                                set secondRightmostButtonX to candidateX
                            end if
                        end if
                    end if
                end try
            end repeat
            if actionButtonCount is 3 then
                set deleteButton to rightmostButton
            else if actionButtonCount is 4 then
                set deleteButton to secondRightmostButton
            else
                error "Expected three or four right-side action buttons in the filtered row"
            end if
            if deleteButton is missing value then error "Delete button candidate is missing"
            click deleteButton
            delay 1

            if exists window "刪除引擎" then
                if not (exists button "刪除" of window "刪除引擎") then error "Chinese delete confirmation is missing"
                click button "刪除" of window "刪除引擎"
            else if exists window "Delete instance" then
                if not (exists button "Delete" of window "Delete instance") then error "English delete confirmation is missing"
                click button "Delete" of window "Delete instance"
            else
                error "Delete confirmation window did not appear"
            end if
            delay 1
            try
                set value of text field 1 of managerWindow to ""
            end try
        end tell
    end tell
end run
'''
    for ui_attempt in range(2):
        completed = subprocess.run(
            ["osascript", "-", expected_display_name],
            input=apple_script,
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )
        if completed.returncode == 0:
            break
        details = (completed.stderr or completed.stdout).strip()
        if (
            ui_attempt == 0
            and "Exact-name filter did not leave exactly one row" in details
        ):
            print(
                "BlueStacks Manager exact-name filter is stale; restarting only "
                "the Manager UI and retrying once",
                flush=True,
            )
            restart_mim_for_stale_ui()
            continue
        raise RuntimeError(f"BlueStacks Manager delete UI failed: {details}")
    else:  # pragma: no cover - loop either breaks or raises
        raise RuntimeError("BlueStacks Manager delete UI retry was exhausted")
    deadline = time.monotonic() + timeout
    engine_dir = DEFAULT_ENGINE_ROOT / engine
    while time.monotonic() < deadline:
        configured = any(item.engine == engine for item in read_instances())
        if not configured and not engine_dir.exists():
            print(f"deleted engine: {engine} (BlueStacks Manager)", flush=True)
            return "bluestacks_manager_exact_filtered_row"
        time.sleep(1)
    configured = any(item.engine == engine for item in read_instances())
    raise RuntimeError(
        f"BlueStacks Manager did not fully delete {engine} within {timeout:.0f}s "
        f"(configured={configured}, directory_exists={engine_dir.exists()})"
    )


def screenshot_bytes(instance: Instance) -> bytes:
    connect(instance)
    data = adb(instance, "exec-out", "screencap", "-p", binary=True)
    if not data.startswith(b"\x89PNG"):
        raise RuntimeError("ADB screencap did not return PNG data")
    return data


def foreground_window(instance: Instance) -> str:
    output = adb(instance, "shell", "dumpsys", "window", timeout=20)
    for line in output.splitlines():
        if "mCurrentFocus=" in line:
            return line.strip()
    return ""


def png_dimensions(data: bytes) -> tuple[int, int]:
    """Read a PNG's width and height without requiring OpenCV."""
    if len(data) < 24 or not data.startswith(b"\x89PNG\r\n\x1a\n"):
        raise RuntimeError("ADB screencap did not return a complete PNG header")
    return (
        int.from_bytes(data[16:20], byteorder="big"),
        int.from_bytes(data[20:24], byteorder="big"),
    )


def decode_png(data: bytes) -> Any:
    require_cv()
    image = cv2.imdecode(np.frombuffer(data, dtype=np.uint8), cv2.IMREAD_COLOR)
    if image is None:
        raise RuntimeError("OpenCV could not decode the ADB screenshot")
    return image


def save_bytes(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)


def timestamp() -> str:
    return datetime.now().strftime("%Y%m%d_%H%M%S")


def unique_timestamped_screenshot_path(
    output_dir: Path,
    instance: Instance,
    label: str,
) -> Path:
    """Return an engine-linked screenshot path without overwriting prior events."""
    safe_label = re.sub(r"[^A-Za-z0-9._-]+", "_", label).strip("._") or "event"
    stem = f"{timestamp()}_{instance.engine}_{safe_label}"
    path = output_dir / f"{stem}.png"
    sequence = 2
    while path.exists():
        path = output_dir / f"{stem}_{sequence}.png"
        sequence += 1
    return path


def normalized_difference(previous: Any, current: Any) -> float:
    require_cv()
    size = (160, 90)
    a = cv2.resize(previous, size, interpolation=cv2.INTER_AREA)
    b = cv2.resize(current, size, interpolation=cv2.INTER_AREA)
    return float(cv2.absdiff(a, b).mean() / 255.0)


def load_workflow(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as handle:
        workflow = json.load(handle)
    if not isinstance(workflow.get("rules", []), list):
        raise SystemExit(f"Workflow rules must be a list: {path}")
    if not workflow.get("stages") and not workflow.get("rules"):
        raise SystemExit(f"Workflow has neither stages nor rules: {path}")
    return workflow


def match_template(
    frame: Any,
    template_path: Path,
    match_mode: str = "color",
) -> tuple[float, tuple[int, int]]:
    require_cv()
    template = cv2.imread(str(template_path), cv2.IMREAD_COLOR)
    if template is None:
        raise RuntimeError(f"Template cannot be read: {template_path}")
    if template.shape[0] > frame.shape[0] or template.shape[1] > frame.shape[1]:
        return 0.0, (0, 0)
    if match_mode == "gray":
        match_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        match_template_image = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY)
    elif match_mode == "edge":
        match_frame = cv2.Canny(cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY), 80, 160)
        match_template_image = cv2.Canny(
            cv2.cvtColor(template, cv2.COLOR_BGR2GRAY), 80, 160
        )
    elif match_mode == "color":
        match_frame = frame
        match_template_image = template
    else:
        raise RuntimeError(f"Unsupported match_mode {match_mode!r}")
    result = cv2.matchTemplate(match_frame, match_template_image, cv2.TM_CCOEFF_NORMED)
    _, score, _, location = cv2.minMaxLoc(result)
    center = (location[0] + template.shape[1] // 2, location[1] + template.shape[0] // 2)
    return float(score), center


def ensure_vision_ocr() -> Path:
    """Compile the local macOS Vision helper once, outside the game loop."""
    if not VISION_OCR_SOURCE.is_file():
        raise RuntimeError(f"Vision OCR source is missing: {VISION_OCR_SOURCE}")
    if (
        VISION_OCR_BINARY.is_file()
        and VISION_OCR_BINARY.stat().st_mtime >= VISION_OCR_SOURCE.stat().st_mtime
    ):
        return VISION_OCR_BINARY
    print("building local macOS Vision OCR helper (one-time)...", flush=True)
    VISION_OCR_BINARY.parent.mkdir(parents=True, exist_ok=True)
    subprocess.run(
        [
            "xcrun",
            "swiftc",
            "-O",
            "-framework",
            "Vision",
            "-framework",
            "AppKit",
            str(VISION_OCR_SOURCE),
            "-o",
            str(VISION_OCR_BINARY),
        ],
        check=True,
        capture_output=True,
        text=True,
        timeout=120,
    )
    return VISION_OCR_BINARY


def ensure_bluestacks_overlay_helper() -> Path:
    """Compile the exact-window BlueStacks host-overlay OCR helper once."""
    if not BLUESTACKS_OVERLAY_SOURCE.is_file():
        raise RuntimeError(
            f"BlueStacks overlay helper source is missing: {BLUESTACKS_OVERLAY_SOURCE}"
        )
    if (
        BLUESTACKS_OVERLAY_BINARY.is_file()
        and BLUESTACKS_OVERLAY_BINARY.stat().st_mtime
        >= BLUESTACKS_OVERLAY_SOURCE.stat().st_mtime
    ):
        return BLUESTACKS_OVERLAY_BINARY
    print("building BlueStacks host-overlay helper (one-time)...", flush=True)
    BLUESTACKS_OVERLAY_BINARY.parent.mkdir(parents=True, exist_ok=True)
    subprocess.run(
        [
            "xcrun",
            "swiftc",
            "-O",
            "-framework",
            "Vision",
            "-framework",
            "AppKit",
            "-framework",
            "ApplicationServices",
            str(BLUESTACKS_OVERLAY_SOURCE),
            "-o",
            str(BLUESTACKS_OVERLAY_BINARY),
        ],
        check=True,
        capture_output=True,
        text=True,
        timeout=120,
    )
    return BLUESTACKS_OVERLAY_BINARY


def mark_game_controls_hint_seen(
    instance: Instance,
    package: str = DEFAULT_PACKAGE,
    *,
    timeout: float = 10.0,
) -> bool:
    """Disable BlueStacks' per-app keyboard/mouse tutorial before first launch."""
    path = DEFAULT_ENGINE_ROOT / instance.engine / "AppCache" / "AppCache.json"
    deadline = time.monotonic() + max(0.0, timeout)
    payload: Any = None
    last_error = "entry not ready"
    while True:
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
            if isinstance(payload, list) and any(
                isinstance(item, dict) and item.get("package") == package
                for item in payload
            ):
                break
            last_error = "entry not ready"
        except (OSError, json.JSONDecodeError) as exc:
            last_error = str(exc)
        if time.monotonic() >= deadline:
            print(f"BlueStacks controls-hint cache unavailable: {last_error}", flush=True)
            return False
        time.sleep(0.5)
    if not isinstance(payload, list):
        return False
    changed = False
    matched = False
    for item in payload:
        if not isinstance(item, dict) or item.get("package") != package:
            continue
        matched = True
        if item.get("isGameControlsToogleHintShown") is not True:
            item["isGameControlsToogleHintShown"] = True
            changed = True
    if not matched:  # Defensive: the readiness loop above already proved it.
        return False
    if changed:
        temporary = path.with_name(f".{path.name}.aov-{os.getpid()}.tmp")
        temporary.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        os.chmod(temporary, path.stat().st_mode)
        os.replace(temporary, path)
    print(
        f"disabled BlueStacks keyboard/mouse tutorial for {instance.engine}",
        flush=True,
    )
    return True


def dismiss_bluestacks_controls_hint(instance: Instance) -> bool:
    """Click Continue only when OCR proves the exact host overlay is present."""
    try:
        output = run_process(
            [
                str(ensure_bluestacks_overlay_helper()),
                instance.display_name,
                "click",
            ],
            timeout=25,
        )
        result = json.loads(output)
    except (OSError, subprocess.SubprocessError, json.JSONDecodeError) as exc:
        print(f"BlueStacks host-overlay check unavailable: {exc}", flush=True)
        return False
    if not isinstance(result, dict):
        return False
    if result.get("clicked"):
        print(
            f"dismissed BlueStacks keyboard/mouse tutorial for {instance.engine}",
            flush=True,
        )
        return True
    if result.get("overlayDetected"):
        print(
            "BlueStacks keyboard/mouse tutorial was detected but its unique "
            f"Continue button was not safely clickable: {result.get('error') or 'unknown'}",
            flush=True,
        )
    return False


def normalize_ocr_text(value: str) -> str:
    value = unicodedata.normalize("NFKC", value).casefold()
    return "".join(character for character in value if character.isalnum())


def ocr_observations(
    frame: Any,
    region: list[float] | None,
    cache: dict[tuple[float, ...], list[dict[str, Any]]],
) -> list[dict[str, Any]]:
    """Recognize a normalized region and return boxes in full-frame pixels."""
    require_cv()
    height, width = frame.shape[:2]
    key = tuple(region) if region is not None else (0.0, 0.0, 1.0, 1.0)
    if key in cache:
        return cache[key]
    x_ratio, y_ratio, width_ratio, height_ratio = key
    x0 = max(0, min(width - 1, round(width * x_ratio)))
    y0 = max(0, min(height - 1, round(height * y_ratio)))
    x1 = max(x0 + 1, min(width, round(width * (x_ratio + width_ratio))))
    y1 = max(y0 + 1, min(height, round(height * (y_ratio + height_ratio))))
    crop = frame[y0:y1, x0:x1]
    encoded, png = cv2.imencode(".png", crop)
    if not encoded:
        raise RuntimeError("OpenCV could not encode an OCR crop")
    temporary_path = ""
    try:
        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as handle:
            handle.write(png.tobytes())
            temporary_path = handle.name
        output = run_process(
            [str(ensure_vision_ocr()), temporary_path],
            timeout=20,
        )
        raw = json.loads(output)
    finally:
        if temporary_path:
            Path(temporary_path).unlink(missing_ok=True)
    observations: list[dict[str, Any]] = []
    for item in raw:
        if not isinstance(item, dict) or not isinstance(item.get("text"), str):
            continue
        observations.append(
            {
                **item,
                "x": float(item.get("x", 0)) + x0,
                "y": float(item.get("y", 0)) + y0,
                "width": float(item.get("width", 0)),
                "height": float(item.get("height", 0)),
                "confidence": float(item.get("confidence", 0)),
            }
        )
    cache[key] = observations
    return observations


def match_ocr(
    frame: Any,
    expected: str,
    region: list[float] | None,
    cache: dict[tuple[float, ...], list[dict[str, Any]]],
) -> tuple[float, tuple[int, int], str]:
    target = normalize_ocr_text(expected)
    if not target:
        raise RuntimeError("OCR target must contain letters or digits")
    best_score = 0.0
    best_center = (0, 0)
    best_text = ""
    for item in ocr_observations(frame, region, cache):
        observed = normalize_ocr_text(str(item["text"]))
        if not observed:
            continue
        if target in observed:
            similarity = 1.0
        elif observed in target and len(observed) >= max(2, len(target) // 2):
            similarity = len(observed) / len(target)
        else:
            similarity = difflib.SequenceMatcher(None, target, observed).ratio()
        confidence = max(0.0, min(1.0, float(item["confidence"])))
        score = similarity * (0.85 + 0.15 * confidence)
        if score > best_score:
            best_score = score
            best_center = (
                round(float(item["x"]) + float(item["width"]) / 2),
                round(float(item["y"]) + float(item["height"]) / 2),
            )
            best_text = str(item["text"])
    return best_score, best_center, best_text


def tap(instance: Instance, x: int, y: int, dry_run: bool) -> None:
    print(f"tap x={x} y={y}" + (" [dry-run]" if dry_run else ""), flush=True)
    if not dry_run:
        adb(instance, "shell", "input", "tap", str(x), str(y))


def swipe(
    instance: Instance,
    start: tuple[int, int],
    end: tuple[int, int],
    duration_ms: int,
    dry_run: bool,
) -> None:
    print(
        f"swipe x1={start[0]} y1={start[1]} x2={end[0]} y2={end[1]} "
        f"duration={duration_ms}ms" + (" [dry-run]" if dry_run else ""),
        flush=True,
    )
    if not dry_run:
        adb(
            instance,
            "shell",
            "input",
            "swipe",
            str(start[0]),
            str(start[1]),
            str(end[0]),
            str(end[1]),
            str(duration_ms),
        )


def rule_action(
    instance: Instance,
    rule: dict[str, Any],
    center: tuple[int, int],
    frame: Any,
    output_dir: Path,
    dry_run: bool,
) -> bool:
    action = rule.get("action", "wait")
    name = str(rule["name"])
    save_as = rule.get("save_as")
    if save_as:
        path = output_dir / str(save_as)
        path.parent.mkdir(parents=True, exist_ok=True)
        if not cv2.imwrite(str(path), frame):
            raise RuntimeError(f"OpenCV could not save checkpoint: {path}")
        print(f"saved checkpoint: {path}", flush=True)
    save_timestamped_as = rule.get("save_timestamped_as")
    if save_timestamped_as:
        path = unique_timestamped_screenshot_path(
            output_dir,
            instance,
            str(save_timestamped_as),
        )
        path.parent.mkdir(parents=True, exist_ok=True)
        if not cv2.imwrite(str(path), frame):
            raise RuntimeError(f"OpenCV could not save event screenshot: {path}")
        print(f"saved event screenshot: {path}", flush=True)
    delay = float(rule.get("pre_delay_seconds", 0))
    if delay > 0 and not dry_run:
        print(f"waiting {delay:.1f}s before action", flush=True)
        time.sleep(delay)
    if action == "wait":
        return False
    if action == "tap_match":
        tap(instance, center[0], center[1], dry_run)
        return False
    if action == "tap_normalized":
        x_ratio, y_ratio = rule["tap"]
        height, width = frame.shape[:2]
        tap(instance, round(width * x_ratio), round(height * y_ratio), dry_run)
        return False
    if action == "swipe_normalized":
        start_x, start_y = rule["start"]
        end_x, end_y = rule["end"]
        height, width = frame.shape[:2]
        swipe(
            instance,
            (round(width * start_x), round(height * start_y)),
            (round(width * end_x), round(height * end_y)),
            int(rule.get("duration_ms", 400)),
            dry_run,
        )
        return False
    if action == "save_and_stop":
        if not save_as:
            path = output_dir / f"{timestamp()}_{instance.engine}_{name}.png"
            cv2.imwrite(str(path), frame)
            print(f"saved result: {path}", flush=True)
        return True
    if action == "fail":
        if not save_as and not save_timestamped_as:
            path = output_dir / f"{timestamp()}_{instance.engine}_{name}.png"
            if not cv2.imwrite(str(path), frame):
                raise RuntimeError(f"OpenCV could not save failure screenshot: {path}")
            print(f"saved failure screenshot: {path}", flush=True)
        message = str(
            rule.get(
                "error_message",
                f"Workflow entered forbidden state {name!r}",
            )
        )
        raise SystemExit(message)
    raise RuntimeError(f"Unsupported action {action!r} in rule {name!r}")


def handle_interrupts(
    instance: Instance,
    workflow: dict[str, Any],
    frame: Any,
    output_dir: Path,
    dry_run: bool,
    stage_name: str,
    ocr_cache: dict[tuple[float, ...], list[dict[str, Any]]] | None = None,
) -> bool:
    """Handle optional overlays without advancing the current workflow stage."""
    matches: list[tuple[int, float, dict[str, Any], tuple[int, int], str]] = []
    if ocr_cache is None:
        ocr_cache = {}
    for interrupt in workflow.get("interrupts", []):
        allowed_stages = interrupt.get("stages")
        if allowed_stages is not None and stage_name not in allowed_stages:
            continue
        excluded_stages = interrupt.get("excluded_stages", [])
        if stage_name in excluded_stages:
            continue
        passing: list[tuple[float, tuple[int, int], str]] = []
        if "ocr_text" in interrupt:
            ocr_score, ocr_center, observed_text = match_ocr(
                frame,
                str(interrupt["ocr_text"]),
                interrupt.get("ocr_region"),
                ocr_cache,
            )
            if ocr_score >= float(interrupt.get("ocr_threshold", 0.86)):
                passing.append((ocr_score, ocr_center, f"OCR {observed_text!r}"))
        if "template" in interrupt:
            template_path = SKILL_DIR / str(interrupt["template"])
            if not template_path.exists() and interrupt.get("optional_missing"):
                template_path = None
            if template_path is not None:
                template_score, template_center = match_template(
                    frame,
                    template_path,
                    str(interrupt.get("match_mode", "color")),
                )
                if template_score >= float(interrupt.get("threshold", 0.88)):
                    passing.append((template_score, template_center, "template"))
        if passing:
            score, center, match_source = max(passing, key=lambda item: item[0])
            matches.append(
                (
                    int(interrupt.get("priority", 0)),
                    score,
                    interrupt,
                    center,
                    match_source,
                )
            )
    if not matches:
        return False
    _, score, interrupt, center, match_source = max(
        matches,
        key=lambda item: (item[0], item[1]),
    )
    print(
        f"interrupt {interrupt['name']}: matched via {match_source} score={score:.3f}",
        flush=True,
    )
    if dry_run:
        print("dry-run stops after the interrupt match", flush=True)
        return True
    rule_action(instance, interrupt, center, frame, output_dir, False)
    time.sleep(float(interrupt.get("after_action_seconds", 0.35)))
    return True


def validate_stages(workflow: dict[str, Any]) -> tuple[dict[str, dict[str, Any]], str]:
    raw_stages = workflow.get("stages")
    if not isinstance(raw_stages, dict) or not raw_stages:
        raise SystemExit("Workflow stages must be a non-empty object")
    stages: dict[str, dict[str, Any]] = {}
    for name, value in raw_stages.items():
        if not isinstance(value, dict):
            raise SystemExit(f"Stage {name!r} must be an object")
        stage = dict(value)
        stage.setdefault("name", name)
        transitions = stage.get("transitions")
        if (
            "template" not in stage
            and "ocr_text" not in stage
            and stage.get("match") != "always"
            and not isinstance(transitions, list)
        ):
            raise SystemExit(
                f"Stage {name!r} has neither a template, OCR text, "
                "an always match, nor transitions"
            )
        if isinstance(transitions, list):
            if not transitions:
                raise SystemExit(f"Stage {name!r} has an empty transitions list")
            for index, transition in enumerate(transitions):
                if (
                    not isinstance(transition, dict)
                    or not (
                        {"template", "ocr_text"} & set(transition)
                        or transition.get("match") == "always"
                    )
                ):
                    raise SystemExit(
                        f"Stage {name!r} transition {index} must contain a template, "
                        "OCR text, or an always match"
                    )
                priority = transition.get("priority")
                if priority is not None and (
                    isinstance(priority, bool) or not isinstance(priority, int)
                ):
                    raise SystemExit(
                        f"Stage {name!r} transition {index} priority must be an integer"
                    )
                require_all = transition.get("require_all")
                if require_all is not None and not isinstance(require_all, bool):
                    raise SystemExit(
                        f"Stage {name!r} transition {index} require_all must be boolean"
                    )
        max_self_transitions = stage.get("max_self_transitions")
        if max_self_transitions is not None and (
            isinstance(max_self_transitions, bool)
            or not isinstance(max_self_transitions, int)
            or max_self_transitions < 1
        ):
            raise SystemExit(
                f"Stage {name!r} max_self_transitions must be a positive integer"
            )
        stages[str(name)] = stage
    entry = str(workflow.get("entry_stage", ""))
    if entry not in stages:
        raise SystemExit(f"Unknown entry_stage {entry!r}")
    for name, stage in stages.items():
        targets = [stage.get("next_stage")]
        targets.extend(
            transition.get("next_stage")
            for transition in stage.get("transitions", [])
            if isinstance(transition, dict)
        )
        for next_stage in targets:
            if next_stage is not None and str(next_stage) not in stages:
                raise SystemExit(f"Stage {name!r} points to unknown next_stage {next_stage!r}")
    interrupts = workflow.get("interrupts", [])
    if not isinstance(interrupts, list):
        raise SystemExit("Workflow interrupts must be a list")
    for index, interrupt in enumerate(interrupts):
        if not isinstance(interrupt, dict) or not ({"template", "ocr_text"} & set(interrupt)):
            raise SystemExit(
                f"Workflow interrupt {index} must be an object containing a template or OCR text"
            )
        priority = interrupt.get("priority")
        if priority is not None and (
            isinstance(priority, bool) or not isinstance(priority, int)
        ):
            raise SystemExit(f"Workflow interrupt {index} priority must be an integer")
        for field in ("stages", "excluded_stages"):
            stage_filter = interrupt.get(field)
            if stage_filter is None:
                continue
            if not isinstance(stage_filter, list) or not all(
                isinstance(item, str) for item in stage_filter
            ):
                raise SystemExit(f"Workflow interrupt {index} {field} must be a string list")
            unknown = sorted(set(stage_filter) - set(stages))
            if unknown:
                raise SystemExit(
                    f"Workflow interrupt {index} {field} contains unknown stages: "
                    + ", ".join(unknown)
                )
    return stages, entry


def run_staged_workflow(
    instance: Instance,
    workflow: dict[str, Any],
    output_dir: Path,
    dry_run: bool,
    start_stage: str | None = None,
) -> int:
    stages, stage_name = validate_stages(workflow)
    if start_stage is not None:
        if start_stage not in stages:
            choices = ", ".join(stages)
            raise SystemExit(f"Unknown --start-stage {start_stage!r}. Available: {choices}")
        stage_name = start_stage
    if any(
        "ocr_text" in candidate
        for stage in stages.values()
        for candidate in (stage.get("transitions") or [stage])
    ) or any("ocr_text" in interrupt for interrupt in workflow.get("interrupts", [])):
        ensure_vision_ocr()
    global_deadline = time.monotonic() + float(workflow.get("global_timeout_seconds", 1800))
    stage_started = time.monotonic()
    progress_path = output_dir / "progress.json"
    heartbeat_seconds = float(workflow.get("heartbeat_seconds", 15.0))
    last_heartbeat = 0.0
    last_action: dict[str, float] = {}
    miss_attempts: dict[str, int] = {}
    self_transition_counts: dict[str, int] = {}
    wrong_orientation_since: float | None = None
    last_orientation_report = 0.0
    orientation_relaunches = 0
    while time.monotonic() < global_deadline:
        stage = stages[stage_name]
        mode = str(stage.get("mode", "normal"))
        stage_timeout = float(
            stage.get(
                "timeout_seconds",
                workflow.get(
                    "battle_timeout_seconds" if mode == "battle" else "unknown_timeout_seconds",
                    1200 if mode == "battle" else 60,
                ),
            )
        )
        now = time.monotonic()
        if now - last_heartbeat >= heartbeat_seconds:
            elapsed = now - stage_started
            print(f"heartbeat stage={stage_name} elapsed={elapsed:.0f}s", flush=True)
            write_progress(
                progress_path,
                engine=instance.engine,
                stage=stage_name,
                status="running",
                detail=f"stage elapsed {elapsed:.0f}s",
            )
            last_heartbeat = now
        data = screenshot_bytes(instance)
        frame = decode_png(data)
        expected_size = workflow.get("frame_size")
        if expected_size and [frame.shape[1], frame.shape[0]] != list(expected_size):
            actual_size = [frame.shape[1], frame.shape[0]]
            reversed_size = list(reversed(expected_size))
            if actual_size == reversed_size:
                now = time.monotonic()
                if wrong_orientation_since is None:
                    wrong_orientation_since = now
                orientation_timeout = float(
                    workflow.get("orientation_timeout_seconds", 180)
                )
                elapsed = now - wrong_orientation_since
                relaunch_after = float(
                    workflow.get("orientation_relaunch_after_seconds", 12)
                )
                max_relaunches = int(workflow.get("orientation_max_relaunches", 2))
                expected_package = str(workflow.get("package", DEFAULT_PACKAGE))
                if (
                    not dry_run
                    and elapsed >= relaunch_after
                    and orientation_relaunches < max_relaunches
                ):
                    foreground = foreground_window(instance)
                    orientation_relaunches += 1
                    print(
                        "portrait framebuffer persisted even though Android may report "
                        "a stale game foreground; forcing a landscape-verified relaunch "
                        f"({orientation_relaunches}/{max_relaunches}): "
                        f"{foreground or 'unknown foreground'}",
                        flush=True,
                    )
                    launch_game(
                        instance,
                        expected_package,
                        attempts=max(
                            1,
                            max_relaunches - orientation_relaunches + 1,
                        ),
                        verify_timeout=60,
                        expected_size=tuple(int(value) for value in expected_size),
                    )
                    wrong_orientation_since = None
                    last_orientation_report = 0.0
                    continue
                if elapsed < orientation_timeout:
                    if now - last_orientation_report >= 10:
                        print(
                            f"waiting for landscape rotation: {actual_size[0]}x"
                            f"{actual_size[1]} ({elapsed:.0f}s/{orientation_timeout:.0f}s)",
                            flush=True,
                        )
                        last_orientation_report = now
                    time.sleep(float(workflow.get("poll_seconds", 1.0)))
                    continue
            path = output_dir / f"{timestamp()}_{instance.engine}_wrong_size.png"
            save_bytes(path, data)
            raise SystemExit(
                f"Expected frame {expected_size}, got {actual_size}; saved {path}"
            )
        wrong_orientation_since = None
        orientation_relaunches = 0
        if time.monotonic() - stage_started >= stage_timeout:
            path = output_dir / f"{timestamp()}_{instance.engine}_{stage_name}_timeout.png"
            save_bytes(path, data)
            raise SystemExit(
                f"Stage {stage_name!r} timed out after {stage_timeout:.0f}s; saved {path}"
            )
        ocr_cache: dict[tuple[float, ...], list[dict[str, Any]]] = {}
        if handle_interrupts(
            instance,
            workflow,
            frame,
            output_dir,
            dry_run,
            stage_name,
            ocr_cache,
        ):
            if dry_run:
                return 0
            continue
        candidates = stage.get("transitions") or [stage]
        matches: list[tuple[int, float, dict[str, Any], tuple[int, int]]] = []
        best_score = 0.0
        for index, candidate in enumerate(candidates):
            passing: list[tuple[float, tuple[int, int], str]] = []
            condition_count = 0
            if candidate.get("match") == "always":
                condition_count += 1
                passing.append((1.0, (frame.shape[1] // 2, frame.shape[0] // 2), "state sequence"))
            if "ocr_text" in candidate:
                condition_count += 1
                ocr_score, ocr_center, observed_text = match_ocr(
                    frame,
                    str(candidate["ocr_text"]),
                    candidate.get("ocr_region"),
                    ocr_cache,
                )
                best_score = max(best_score, ocr_score)
                if ocr_score >= float(candidate.get("ocr_threshold", 0.86)):
                    passing.append(
                        (ocr_score, ocr_center, f"OCR {observed_text!r}")
                    )
            if "template" in candidate:
                condition_count += 1
                template_path = SKILL_DIR / str(candidate["template"])
                if not template_path.exists() and candidate.get("optional_missing"):
                    template_path = None
                if template_path is not None:
                    template_score, template_center = match_template(
                        frame,
                        template_path,
                        str(candidate.get("match_mode", "color")),
                    )
                    best_score = max(best_score, template_score)
                    if template_score >= float(
                        candidate.get("threshold", stage.get("threshold", 0.88))
                    ):
                        passing.append((template_score, template_center, "template"))
            require_all = bool(candidate.get("require_all", False))
            if require_all and len(passing) != condition_count:
                continue
            if passing:
                center = max(passing, key=lambda item: item[0])[1]
                if require_all:
                    score = min(item[0] for item in passing)
                    match_source = " AND ".join(item[2] for item in passing)
                else:
                    score, center, match_source = max(passing, key=lambda item: item[0])
                action_rule = dict(stage)
                action_rule.pop("transitions", None)
                action_rule.update(candidate)
                action_rule.setdefault("name", f"{stage_name}:{index}")
                action_rule["_match_source"] = match_source
                matches.append((int(candidate.get("priority", 0)), score, action_rule, center))
        if not matches:
            miss_rule = stage.get("on_miss")
            if isinstance(miss_rule, dict):
                miss_key = f"{stage_name}:on_miss"
                now = time.monotonic()
                after_seconds = float(miss_rule.get("after_seconds", 2.0))
                cooldown = float(miss_rule.get("cooldown_seconds", 3.0))
                if (
                    now - stage_started >= after_seconds
                    and now - last_action.get(miss_key, 0.0) >= cooldown
                ):
                    attempt = miss_attempts.get(stage_name, 0)
                    sequence = miss_rule.get("sequence")
                    if isinstance(sequence, list):
                        actions: list[dict[str, Any]] = []
                        for item in sequence:
                            if not isinstance(item, dict):
                                raise RuntimeError(
                                    f"Stage {stage_name!r} has a non-object on_miss sequence item"
                                )
                            repeat = int(item.get("repeat", 1))
                            if repeat < 1:
                                raise RuntimeError(
                                    f"Stage {stage_name!r} has an invalid on_miss repeat"
                                )
                            action = dict(miss_rule)
                            action.pop("sequence", None)
                            action.update(item)
                            action.pop("repeat", None)
                            actions.extend(dict(action) for _ in range(repeat))
                        if attempt >= len(actions):
                            path = output_dir / (
                                f"{timestamp()}_{instance.engine}_{stage_name}_search_exhausted.png"
                            )
                            save_bytes(path, data)
                            raise SystemExit(
                                f"Stage {stage_name!r} exhausted its bounded search "
                                f"after {attempt} actions; saved {path}"
                            )
                        action_rule = actions[attempt]
                    else:
                        max_attempts = int(miss_rule.get("max_attempts", 0))
                        if max_attempts > 0 and attempt >= max_attempts:
                            path = output_dir / (
                                f"{timestamp()}_{instance.engine}_{stage_name}_search_exhausted.png"
                            )
                            save_bytes(path, data)
                            raise SystemExit(
                                f"Stage {stage_name!r} exhausted its bounded search "
                                f"after {attempt} actions; saved {path}"
                            )
                        action_rule = dict(miss_rule)
                    action_rule.setdefault("name", f"{stage_name}_on_miss")
                    print(
                        f"stage {stage_name}: running {action_rule['name']} "
                        f"(search {attempt + 1})",
                        flush=True,
                    )
                    rule_action(instance, action_rule, (0, 0), frame, output_dir, dry_run)
                    if dry_run:
                        return 0
                    miss_attempts[stage_name] = attempt + 1
                    last_action[miss_key] = now
                    time.sleep(float(action_rule.get("after_action_seconds", 1.0)))
                    continue
            poll_key = "battle_poll_seconds" if mode == "battle" else "poll_seconds"
            time.sleep(float(workflow.get(poll_key, 10.0 if mode == "battle" else 1.0)))
            continue
        _, score, action_rule, center = max(matches, key=lambda item: (item[0], item[1]))
        print(
            f"stage {stage_name}: matched {action_rule['name']} "
            f"via {action_rule.get('_match_source', 'visual')} score={score:.3f}",
            flush=True,
        )
        write_progress(
            progress_path,
            engine=instance.engine,
            stage=stage_name,
            status="matched",
            detail=str(action_rule["name"]),
        )
        if dry_run:
            print("dry-run stops after the current stage match", flush=True)
            return 0
        action_key = f"{stage_name}:{action_rule['name']}"
        now = time.monotonic()
        cooldown = float(action_rule.get("cooldown_seconds", 1.0))
        if now - last_action.get(action_key, 0.0) < cooldown:
            time.sleep(float(workflow.get("poll_seconds", 0.35)))
            continue
        default_next = stage_name if stage.get("transitions") else stage.get("next_stage")
        raw_next_stage = action_rule.get("next_stage", default_next)
        next_stage = None if raw_next_stage is None else str(raw_next_stage)
        is_input_action = action_rule.get("action", "wait") in {
            "tap_match",
            "tap_normalized",
            "swipe_normalized",
        }
        is_self_transition = is_input_action and next_stage == stage_name
        max_self_transitions = stage.get("max_self_transitions")
        if is_self_transition and max_self_transitions is not None:
            completed = self_transition_counts.get(stage_name, 0)
            if completed >= int(max_self_transitions):
                path = output_dir / (
                    f"{timestamp()}_{instance.engine}_{stage_name}_"
                    "self_transition_overflow.png"
                )
                save_bytes(path, data)
                raise SystemExit(
                    f"Stage {stage_name!r} exceeded max_self_transitions="
                    f"{max_self_transitions} while matching {action_rule['name']!r}; "
                    f"saved {path}"
                )
        if rule_action(instance, action_rule, center, frame, output_dir, False):
            return 0
        if is_self_transition:
            completed = self_transition_counts.get(stage_name, 0) + 1
            self_transition_counts[stage_name] = completed
            if max_self_transitions is not None:
                print(
                    f"stage {stage_name}: self-transition {completed}/"
                    f"{max_self_transitions}",
                    flush=True,
                )
        miss_attempts[stage_name] = 0
        last_action[action_key] = now
        if next_stage is None:
            raise SystemExit(f"Stage {stage_name!r} completed without next_stage or stop action")
        if next_stage != stage_name:
            self_transition_counts.pop(stage_name, None)
            stage_name = next_stage
            stage_started = time.monotonic()
            last_heartbeat = 0.0
        time.sleep(
            float(action_rule.get("after_action_seconds", workflow.get("poll_seconds", 1.0)))
        )
    path = output_dir / f"{timestamp()}_{instance.engine}_global_timeout.png"
    save_bytes(path, screenshot_bytes(instance))
    raise SystemExit(f"Global timeout; saved {path}")


def command_status(_: argparse.Namespace) -> int:
    for item in read_instances():
        state = "running" if item.running else "stopped"
        print(f"{item.engine}\t{item.display_name}\tport={item.adb_port}\t{state}")
    return 0


def command_screenshot(args: argparse.Namespace) -> int:
    instance = resolve_instance(args.instance)
    data = screenshot_bytes(instance)
    path = Path(args.output).expanduser().resolve()
    save_bytes(path, data)
    frame = decode_png(data)
    print(f"saved {frame.shape[1]}x{frame.shape[0]} screenshot: {path}")
    return 0


def command_observe(args: argparse.Namespace) -> int:
    instance = resolve_instance(args.instance)
    output_dir = Path(args.output_dir).expanduser().resolve()
    output_dir.mkdir(parents=True, exist_ok=True)
    deadline = time.monotonic() + args.duration
    previous = None
    saved = 0
    print(f"observing {instance.engine} on {instance.serial}; Ctrl-C to stop")
    try:
        while time.monotonic() < deadline:
            data = screenshot_bytes(instance)
            frame = decode_png(data)
            difference = 1.0 if previous is None else normalized_difference(previous, frame)
            if previous is None or difference >= args.change_threshold:
                path = output_dir / f"{timestamp()}_{saved:04d}.png"
                save_bytes(path, data)
                print(f"saved {path.name} change={difference:.4f}", flush=True)
                saved += 1
            previous = frame
            time.sleep(args.interval)
    except KeyboardInterrupt:
        pass
    print(f"observer stopped; {saved} frames saved to {output_dir}")
    return 0


def command_crop_template(args: argparse.Namespace) -> int:
    require_cv()
    source = Path(args.source).expanduser().resolve()
    destination = Path(args.output).expanduser().resolve()
    frame = cv2.imread(str(source), cv2.IMREAD_COLOR)
    if frame is None:
        raise SystemExit(f"Source image cannot be read: {source}")
    height, width = frame.shape[:2]
    x, y, crop_width, crop_height = args.rect
    if x < 0 or y < 0 or crop_width <= 0 or crop_height <= 0:
        raise SystemExit("Crop rectangle must be positive and inside the source image")
    if x + crop_width > width or y + crop_height > height:
        raise SystemExit(f"Crop rectangle exceeds source size {width}x{height}")
    destination.parent.mkdir(parents=True, exist_ok=True)
    if not cv2.imwrite(str(destination), frame[y : y + crop_height, x : x + crop_width]):
        raise SystemExit(f"Could not save template: {destination}")
    print(f"saved {crop_width}x{crop_height} template: {destination}")
    return 0


def launcher_activity(instance: Instance, package: str) -> str | None:
    output = adb(
        instance,
        "shell",
        "cmd",
        "package",
        "resolve-activity",
        "--brief",
        "-c",
        "android.intent.category.LAUNCHER",
        package,
        timeout=30,
    )
    for line in reversed(output.splitlines()):
        candidate = line.strip()
        if candidate.startswith(f"{package}/"):
            return candidate
    return None


def launch_game(
    instance: Instance,
    package: str = DEFAULT_PACKAGE,
    *,
    attempts: int = 2,
    verify_timeout: float = 90,
    expected_size: tuple[int, int] | None = None,
) -> None:
    connect(instance)
    component = launcher_activity(instance, package)
    last_foreground = ""
    last_frame_size: tuple[int, int] | None = None
    last_launch_error = ""
    last_host_overlay_check = 0.0
    for attempt in range(1, attempts + 1):
        try:
            adb(instance, "shell", "am", "force-stop", package, timeout=30)
            # Clear the BlueStacks Store surface before bringing the game to
            # the foreground.  On a busy dual-engine launch, mCurrentFocus can
            # otherwise report SGameRealActivity while screencap still shows
            # the Store or its white loading page.
            adb(instance, "shell", "input", "keyevent", "3", timeout=30)
            # The base BlueStacks framebuffer is portrait.  Some fresh engines
            # let the app override the user's locked rotation.  Force the
            # window manager to honor the user rotation, then lock landscape.
            adb(
                instance,
                "shell",
                "wm",
                "fixed-to-user-rotation",
                "enabled",
                timeout=30,
            )
            adb(
                instance,
                "shell",
                "wm",
                "user-rotation",
                "lock",
                "1",
                timeout=30,
            )
        except subprocess.SubprocessError as exc:
            last_launch_error = f"launch preparation failed: {type(exc).__name__}: {exc}"
            print(
                f"game launch attempt {attempt}/{attempts} preparation failed; "
                f"retrying without touching another engine: {exc}",
                flush=True,
            )
            connect(instance)
            continue
        time.sleep(2)
        try:
            if component:
                output = adb(
                    instance,
                    "shell",
                    "am",
                    "start",
                    "-W",
                    "-n",
                    component,
                    timeout=90,
                )
            else:
                output = adb(
                    instance,
                    "shell",
                    "monkey",
                    "-p",
                    package,
                    "-c",
                    "android.intent.category.LAUNCHER",
                    "1",
                    timeout=60,
                )
            print(output.strip(), flush=True)
        except subprocess.SubprocessError as exc:
            # `am start -W` itself can time out while the activity continues
            # loading.  Do not immediately force-stop it; verify the actual
            # framebuffer for the full window first.
            last_launch_error = f"activity start command delayed: {type(exc).__name__}: {exc}"
            print(
                f"game launch attempt {attempt}/{attempts} start command was delayed; "
                "keeping the activity alive while verifying its framebuffer: "
                f"{exc}",
                flush=True,
            )
        deadline = time.monotonic() + verify_timeout
        stable_landscape_frames = 0
        while time.monotonic() < deadline:
            now = time.monotonic()
            if now - last_host_overlay_check >= 10.0:
                last_host_overlay_check = now
                if dismiss_bluestacks_controls_hint(instance):
                    stable_landscape_frames = 0
                    time.sleep(1)
            try:
                last_foreground = foreground_window(instance)
                if package in last_foreground and expected_size is not None:
                    last_frame_size = png_dimensions(screenshot_bytes(instance))
                elif package in last_foreground:
                    last_frame_size = None
            except (RuntimeError, subprocess.SubprocessError) as exc:
                stable_landscape_frames = 0
                last_launch_error = f"framebuffer verification delayed: {type(exc).__name__}: {exc}"
                time.sleep(1)
                continue
            if package in last_foreground:
                if expected_size is None:
                    stable_landscape_frames += 1
                elif last_frame_size == expected_size:
                    stable_landscape_frames += 1
                else:
                    stable_landscape_frames = 0
                if stable_landscape_frames >= 3:
                    frame_detail = (
                        f"; frame={last_frame_size[0]}x{last_frame_size[1]}"
                        if last_frame_size is not None
                        else ""
                    )
                    print(
                        "verified stable game foreground and landscape framebuffer "
                        f"on launch attempt {attempt}: {last_foreground}{frame_detail}",
                        flush=True,
                    )
                    return
            else:
                stable_landscape_frames = 0
            time.sleep(1)
        print(
            f"game launch attempt {attempt}/{attempts} did not reach a verified "
            f"landscape game surface; foreground={last_foreground or 'unknown'}; "
            f"frame={last_frame_size or 'unknown'}",
            flush=True,
        )
    raise RuntimeError(
        f"Game launch did not reach a landscape {package} surface after "
        f"{attempts} attempts; foreground={last_foreground or 'unknown'}; "
        f"frame={last_frame_size or 'unknown'}; "
        f"last_launch_error={last_launch_error or 'none'}"
    )


def command_launch(args: argparse.Namespace) -> int:
    instance = resolve_instance(args.instance)
    launch_game(instance, args.package)
    return 0


def install_xapk(instance: Instance, source: Path) -> None:
    if not source.is_file():
        raise SystemExit(f"XAPK not found: {source}")
    required = ["com.garena.game.kgtw.apk", "config.arm64_v8a.apk", "assetpack.apk"]
    with zipfile.ZipFile(source) as archive:
        missing = [name for name in required if name not in archive.namelist()]
        if missing:
            raise SystemExit(f"XAPK is missing required APKs: {', '.join(missing)}")
        with tempfile.TemporaryDirectory(prefix="aov-xapk-") as temp_dir:
            extracted: list[str] = []
            for name in required:
                archive.extract(name, temp_dir)
                extracted.append(str(Path(temp_dir) / name))
            connect(instance)
            output = adb(
                instance,
                "install-multiple",
                "-r",
                *extracted,
                timeout=300,
            )
    print(output.strip(), flush=True)


def command_install_xapk(args: argparse.Namespace) -> int:
    instance = resolve_instance(args.instance)
    source = Path(args.xapk).expanduser().resolve()
    install_xapk(instance, source)
    return 0


def write_manifest(path: Path, manifest: dict[str, Any]) -> None:
    temp_path = path.with_suffix(".tmp")
    temp_path.write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(temp_path, path)


def write_progress(
    path: Path,
    *,
    engine: str,
    stage: str,
    status: str,
    detail: str = "",
) -> None:
    """Write a small heartbeat that an outer watchdog/dashboard can inspect."""
    value = {
        "engine": engine,
        "stage": stage,
        "status": status,
        "detail": detail,
        "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
    }
    temporary = path.with_suffix(".tmp")
    temporary.write_text(
        json.dumps(value, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(temporary, path)


def prefix_result_screenshots(output_dir: Path, engine: str) -> list[str]:
    results: list[str] = []
    for base_name in RESULT_SCREENSHOTS:
        source = output_dir / base_name
        destination = output_dir / f"{engine}_{base_name}"
        if source.exists() and not destination.exists():
            source.rename(destination)
        if destination.exists():
            results.append(destination.name)
    return results


def command_fresh_run(args: argparse.Namespace) -> int:
    """Create, configure, run, record, and close one fresh BlueStacks engine."""
    require_cv()
    if not 1 <= args.engine_cpu <= 8:
        raise SystemExit("--engine-cpu must be between 1 and 8")
    if not 2048 <= args.engine_ram_mb <= 8192:
        raise SystemExit("--engine-ram-mb must be between 2048 and 8192")
    if args.lifecycle_lock_timeout < 30:
        raise SystemExit("--lifecycle-lock-timeout must be at least 30 seconds")
    if not 0 <= args.launch_restart_attempts <= 2:
        raise SystemExit("--launch-restart-attempts must be between 0 and 2")
    if not 0 <= args.launch_restart_settle_seconds <= 60:
        raise SystemExit(
            "--launch-restart-settle-seconds must be between 0 and 60 seconds"
        )
    xapk = Path(args.xapk).expanduser().resolve()
    if not xapk.is_file():
        raise SystemExit(f"XAPK not found: {xapk}")
    if args.skip_install and not args.adopt_instance:
        raise SystemExit("--skip-install may only be used together with --adopt-instance")
    workflow_path = Path(args.workflow).expanduser().resolve()
    workflow = load_workflow(workflow_path)
    output_root = Path(args.output_root).expanduser().resolve()
    output_root.mkdir(parents=True, exist_ok=True)
    free_gib = available_gib(output_root)
    if free_gib < args.min_free_gb:
        raise SystemExit(
            f"Only {free_gib:.1f} GiB is free; fresh-run requires at least "
            f"{args.min_free_gb:.1f} GiB"
        )
    allowed_peers = set(args.allow_running_engine or [])
    lifecycle_path = (
        Path(args.lifecycle_lock).expanduser().resolve()
        if args.lifecycle_lock
        else None
    )
    startup_failure: BaseException | None = None
    with lifecycle_lock(lifecycle_path, args.lifecycle_lock_timeout):
        validate_running_peers(allowed_peers)
        run_id = timestamp()
        if args.adopt_instance:
            instance = instance_by_engine(args.adopt_instance)
            if instance.running:
                raise SystemExit(f"Adopted engine must be stopped: {instance.engine}")
            print(f"adopting already-created blank engine: {instance.engine}", flush=True)
        else:
            instance = create_fresh_instance(
                args.create_timeout,
                cpu=args.engine_cpu,
                ram_mb=args.engine_ram_mb,
            )
        output_dir = output_root / f"{run_id}_{instance.engine}"
        output_dir.mkdir(parents=True, exist_ok=False)
        manifest_path = output_dir / "run.json"
        manifest: dict[str, Any] = {
            "run_id": run_id,
            "engine": instance.engine,
            "display_name": instance.display_name,
            "adb_port": instance.adb_port,
            "xapk": str(xapk),
            "workflow": str(workflow_path),
            "start_stage": args.start_stage,
            "output_dir": str(output_dir),
            "started_at": datetime.now().astimezone().isoformat(timespec="seconds"),
            "status": "created",
            "step": "configure",
            "screenshots": [],
            "engine_deleted": False,
            "engine_cpu": args.engine_cpu,
            "engine_ram_mb": args.engine_ram_mb,
            "allowed_peer_engines": sorted(allowed_peers),
            "lifecycle_lock": str(lifecycle_path) if lifecycle_path else None,
        }
        write_manifest(manifest_path, manifest)
        try:
            configure_instance_for_workflow(instance)
            manifest["step"] = "boot"
            write_manifest(manifest_path, manifest)
            instance = start_instance(instance, args.boot_timeout)
            manifest.update(
                {
                    "display_name": instance.display_name,
                    "adb_port": instance.adb_port,
                    "status": "running",
                    "step": "install_xapk",
                }
            )
            write_manifest(manifest_path, manifest)
        except BaseException as exc:
            startup_failure = exc

    failure: BaseException | None = None
    close_error: str | None = None
    try:
        if startup_failure is not None:
            raise startup_failure
        if args.skip_install:
            print(f"skipping XAPK reinstall for adopted engine: {instance.engine}", flush=True)
        else:
            install_xapk(instance, xapk)
        mark_game_controls_hint_seen(
            instance,
            str(workflow.get("package", DEFAULT_PACKAGE)),
        )
        manifest["step"] = "game_workflow"
        write_manifest(manifest_path, manifest)
        package = str(workflow.get("package", DEFAULT_PACKAGE))
        expected_size = tuple(int(value) for value in workflow["frame_size"])
        launch_recoveries: list[dict[str, Any]] = []
        manifest["launch_recoveries"] = launch_recoveries
        for launch_index in range(args.launch_restart_attempts + 1):
            try:
                launch_game(
                    instance,
                    package,
                    attempts=2 if launch_index == 0 else 3,
                    expected_size=expected_size,
                )
                if launch_recoveries:
                    launch_recoveries[-1]["status"] = "recovered"
                    launch_recoveries[-1]["recovered_at"] = (
                        datetime.now().astimezone().isoformat(timespec="seconds")
                    )
                    write_manifest(manifest_path, manifest)
                break
            except Exception as launch_error:
                if launch_index >= args.launch_restart_attempts:
                    prior = "; ".join(
                        str(item.get("error", "unknown"))
                        for item in launch_recoveries
                    )
                    detail = f"{type(launch_error).__name__}: {launch_error}"
                    if prior:
                        detail = f"{prior}; after restart: {detail}"
                    raise RuntimeError(
                        "Game launch remained unusable after "
                        f"{args.launch_restart_attempts} exact-engine restart(s): "
                        f"{detail}"
                    ) from launch_error

                recovery: dict[str, Any] = {
                    "attempt": launch_index + 1,
                    "status": "restarting",
                    "error": f"{type(launch_error).__name__}: {launch_error}",
                    "started_at": datetime.now()
                    .astimezone()
                    .isoformat(timespec="seconds"),
                }
                launch_recoveries.append(recovery)
                recovery_path = output_dir / (
                    f"{instance.engine}_launch_recovery_{launch_index + 1}.png"
                )
                try:
                    save_bytes(recovery_path, screenshot_bytes(instance))
                    recovery["screenshot"] = recovery_path.name
                except Exception as screenshot_error:
                    recovery["screenshot_error"] = (
                        f"{type(screenshot_error).__name__}: {screenshot_error}"
                    )
                manifest["step"] = "game_launch_recovery"
                write_manifest(manifest_path, manifest)
                write_progress(
                    output_dir / "progress.json",
                    engine=instance.engine,
                    stage="game_launch_recovery",
                    status="recovering",
                    detail=(
                        "initial game surface was unusable; restarting only this "
                        f"owned engine ({launch_index + 1}/"
                        f"{args.launch_restart_attempts})"
                    ),
                )
                print(
                    "game surface/ADB remained unusable; restarting exact owned "
                    f"engine {instance.engine} before retrying the launch "
                    f"({launch_index + 1}/{args.launch_restart_attempts}): "
                    f"{launch_error}",
                    flush=True,
                )
                with lifecycle_lock(lifecycle_path, args.lifecycle_lock_timeout):
                    close_instance(instance.engine, args.close_timeout)
                    try:
                        subprocess.run(
                            [adb_binary(), "disconnect", instance.serial],
                            capture_output=True,
                            text=True,
                            timeout=10,
                            check=False,
                        )
                    except (OSError, subprocess.SubprocessError):
                        pass
                    if args.launch_restart_settle_seconds:
                        time.sleep(args.launch_restart_settle_seconds)
                    instance = start_instance(
                        instance_by_engine(instance.engine),
                        args.boot_timeout,
                    )
                manifest["adb_port"] = instance.adb_port
                recovery["status"] = "restarted"
                recovery["restarted_at"] = (
                    datetime.now().astimezone().isoformat(timespec="seconds")
                )
                write_manifest(manifest_path, manifest)
                mark_game_controls_hint_seen(instance, package)
        manifest["step"] = "game_workflow"
        write_manifest(manifest_path, manifest)
        time.sleep(args.launch_wait)
        if workflow.get("stages"):
            run_staged_workflow(
                instance,
                workflow,
                output_dir,
                False,
                args.start_stage,
            )
        else:
            raise SystemExit("fresh-run requires a staged workflow")
        write_progress(
            output_dir / "progress.json",
            engine=instance.engine,
            stage="finished",
            status="completed",
            detail="all workflow checkpoints completed",
        )
        manifest["status"] = "completed"
        manifest["step"] = "close"
    except BaseException as exc:
        failure = exc
        try:
            write_progress(
                output_dir / "progress.json",
                engine=instance.engine,
                stage=str(manifest.get("step", "unknown")),
                status="failed",
                detail=f"{type(exc).__name__}: {exc}",
            )
        except Exception:
            pass
        manifest["status"] = "failed"
        manifest["error"] = f"{type(exc).__name__}: {exc}"
        manifest["step"] = "close_after_failure"
        try:
            failure_path = output_dir / f"{instance.engine}_failure.png"
            save_bytes(failure_path, screenshot_bytes(instance))
            manifest["failure_screenshot"] = failure_path.name
        except Exception:
            pass
    finally:
        manifest["screenshots"] = prefix_result_screenshots(output_dir, instance.engine)
        write_manifest(manifest_path, manifest)
        try:
            with lifecycle_lock(lifecycle_path, args.lifecycle_lock_timeout):
                manifest["close_method"] = close_instance(
                    instance.engine,
                    args.close_timeout,
                )
            manifest["engine_closed"] = True
        except Exception as exc:
            close_error = str(exc)
            manifest["engine_closed"] = False
            manifest["close_error"] = close_error
            if failure is None:
                failure = exc
                manifest["status"] = "failed"
        manifest["finished_at"] = datetime.now().astimezone().isoformat(timespec="seconds")
        manifest["step"] = "finished"
        manifest["screenshots"] = prefix_result_screenshots(output_dir, instance.engine)
        write_manifest(manifest_path, manifest)

    print(f"run folder: {output_dir}", flush=True)
    print(f"run manifest: {manifest_path}", flush=True)
    if failure is not None:
        raise failure
    if close_error is not None:
        raise RuntimeError(close_error)
    return 0


def command_run(args: argparse.Namespace) -> int:
    instance = resolve_instance(args.instance)
    workflow_path = Path(args.workflow).expanduser().resolve()
    workflow = load_workflow(workflow_path)
    output_dir = Path(args.output_dir).expanduser().resolve()
    output_dir.mkdir(parents=True, exist_ok=True)
    if args.launch:
        launch_game(
            instance,
            str(workflow.get("package", DEFAULT_PACKAGE)),
            expected_size=tuple(int(value) for value in workflow["frame_size"]),
        )
        time.sleep(args.launch_wait)
    if workflow.get("stages"):
        return run_staged_workflow(
            instance,
            workflow,
            output_dir,
            args.dry_run,
            args.start_stage,
        )
    rules = workflow["rules"]
    if not rules:
        raise SystemExit(
            "Workflow has no visual rules. Run observe, create templates, and register rules first."
        )
    start = time.monotonic()
    global_deadline = start + float(workflow.get("global_timeout_seconds", 1800))
    unknown_since: float | None = None
    last_action: dict[str, float] = {}
    while time.monotonic() < global_deadline:
        data = screenshot_bytes(instance)
        frame = decode_png(data)
        matches: list[tuple[float, dict[str, Any], tuple[int, int]]] = []
        for rule in rules:
            template_path = SKILL_DIR / str(rule["template"])
            score, center = match_template(
                frame, template_path, str(rule.get("match_mode", "color"))
            )
            threshold = float(rule.get("threshold", 0.88))
            if score >= threshold:
                matches.append((score, rule, center))
        if not matches:
            foreground = foreground_window(instance)
            if "SGameRealActivity" in foreground:
                unknown_since = None
                print("battle activity detected; waiting without input", flush=True)
                time.sleep(float(workflow.get("battle_poll_seconds", 10.0)))
                continue
            if unknown_since is None:
                unknown_since = time.monotonic()
            timeout = float(workflow.get("unknown_timeout_seconds", 60))
            if time.monotonic() - unknown_since >= timeout:
                path = output_dir / f"{timestamp()}_{instance.engine}_unknown.png"
                save_bytes(path, data)
                raise SystemExit(f"Unknown state timed out; saved {path}")
            time.sleep(float(workflow.get("poll_seconds", 2.0)))
            continue
        unknown_since = None
        score, rule, center = max(matches, key=lambda item: item[0])
        name = str(rule["name"])
        cooldown = float(rule.get("cooldown_seconds", 2.0))
        now = time.monotonic()
        print(f"matched {name} score={score:.3f}", flush=True)
        if now - last_action.get(name, 0.0) >= cooldown:
            if rule_action(instance, rule, center, frame, output_dir, args.dry_run):
                return 0
            last_action[name] = now
        poll = float(
            workflow.get("battle_poll_seconds", 10.0)
            if rule.get("mode") == "battle"
            else workflow.get("poll_seconds", 2.0)
        )
        time.sleep(poll)
    path = output_dir / f"{timestamp()}_{instance.engine}_global_timeout.png"
    save_bytes(path, screenshot_bytes(instance))
    raise SystemExit(f"Global timeout; saved {path}")


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    sub = result.add_subparsers(dest="command", required=True)

    status = sub.add_parser("status", help="list BlueStacks instances and ADB ports")
    status.set_defaults(func=command_status)

    shot = sub.add_parser("screenshot", help="save one native ADB screenshot")
    shot.add_argument("--instance")
    shot.add_argument("--output", required=True)
    shot.set_defaults(func=command_screenshot)

    observe = sub.add_parser("observe", help="save materially changed screens")
    observe.add_argument("--instance")
    observe.add_argument("--output-dir", required=True)
    observe.add_argument("--duration", type=float, default=1800)
    observe.add_argument("--interval", type=float, default=2.0)
    observe.add_argument("--change-threshold", type=float, default=0.025)
    observe.set_defaults(func=command_observe)

    crop = sub.add_parser("crop-template", help="crop a template from a captured frame")
    crop.add_argument("--source", required=True)
    crop.add_argument("--rect", type=int, nargs=4, metavar=("X", "Y", "W", "H"), required=True)
    crop.add_argument("--output", required=True)
    crop.set_defaults(func=command_crop_template)

    launch = sub.add_parser("launch", help="launch Arena of Valor in one running instance")
    launch.add_argument("--instance")
    launch.add_argument("--package", default=DEFAULT_PACKAGE)
    launch.set_defaults(func=command_launch)

    install = sub.add_parser(
        "install-xapk", help="install the three Arena of Valor APK splits from an XAPK"
    )
    install.add_argument("--instance", required=True)
    install.add_argument("--xapk", required=True)
    install.set_defaults(func=command_install_xapk)

    fresh = sub.add_parser(
        "fresh-run",
        help="create a fresh engine, install the XAPK, run the workflow, and close it",
    )
    fresh.add_argument("--xapk", required=True)
    fresh.add_argument("--workflow", default=str(DEFAULT_WORKFLOW))
    fresh.add_argument("--output-root", default="outputs/aov-reroll-runs")
    fresh.add_argument("--min-free-gb", type=float, default=20.0)
    fresh.add_argument("--create-timeout", type=float, default=90.0)
    fresh.add_argument("--boot-timeout", type=float, default=240.0)
    fresh.add_argument("--close-timeout", type=float, default=60.0)
    fresh.add_argument("--launch-wait", type=float, default=2.0)
    fresh.add_argument(
        "--launch-restart-attempts",
        type=int,
        default=1,
        help=(
            "restart only the owned BlueStacks engine and retry when the first "
            "game launch leaves a stale portrait/Store surface"
        ),
    )
    fresh.add_argument(
        "--launch-restart-settle-seconds",
        type=float,
        default=10.0,
        help="wait after closing a launch-stalled engine before restarting it",
    )
    fresh.add_argument("--engine-cpu", type=int, default=4)
    fresh.add_argument("--engine-ram-mb", type=int, default=4096)
    fresh.add_argument(
        "--allow-running-engine",
        action="append",
        default=[],
        help="parallel supervisor only: exact peer engine allowed to be running",
    )
    fresh.add_argument(
        "--lifecycle-lock",
        default=str(DEFAULT_LIFECYCLE_LOCK),
        help="serialize BlueStacks create/configure/start operations across workers",
    )
    fresh.add_argument("--lifecycle-lock-timeout", type=float, default=600.0)
    fresh.add_argument("--start-stage", help="resume recognition from this workflow stage")
    fresh.add_argument(
        "--skip-install",
        action="store_true",
        help="recovery only: do not reinstall the XAPK on an adopted engine",
    )
    fresh.add_argument(
        "--adopt-instance",
        help="recovery/testing only: use one existing stopped blank engine instead of creating one",
    )
    fresh.set_defaults(func=command_fresh_run)

    run = sub.add_parser("run", help="run a template-gated workflow")
    run.add_argument("--instance")
    run.add_argument("--workflow", default=str(DEFAULT_WORKFLOW))
    run.add_argument("--output-dir", default="outputs/aov-reroll-results")
    run.add_argument("--dry-run", action="store_true")
    run.add_argument("--start-stage", help="resume a staged workflow from this exact stage")
    run.add_argument("--launch", action="store_true", help="launch the game before recognition")
    run.add_argument("--launch-wait", type=float, default=2.0)
    run.set_defaults(func=command_run)
    return result


def main() -> int:
    args = parser().parse_args()
    return int(args.func(args))


if __name__ == "__main__":
    sys.exit(main())
