Cli Anything Openclaw

# cli_anything/macrocli package

"""Enable: python3 -m cli_anything.macrocli"""
from cli_anything.macrocli.macrocli_cli import cli

if __name__ == "__main__":
    cli()

"""Backend base classes and result types.

All execution backends inherit from Backend and return StepResult.
"""

from __future__ import annotations

import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional


@dataclass
class StepResult:
    """Result of a single macro step execution."""
    success: bool
    output: dict = field(default_factory=dict)
    error: str = ""
    duration_ms: float = 0.0
    backend_used: str = ""

    def to_dict(self) -> dict:
        return {
            "success": self.success,
            "output": self.output,
            "error": self.error,
            "duration_ms": self.duration_ms,
            "backend_used": self.backend_used,
        }


class BackendContext:
    """Runtime context passed to each backend during step execution."""

    def __init__(
        self,
        params: dict,
        previous_results: Optional[list[StepResult]] = None,
        dry_run: bool = False,
        timeout_ms: int = 30_000,
    ):
        self.params = params
        self.previous_results: list[StepResult] = previous_results or []
        self.dry_run = dry_run
        self.timeout_ms = timeout_ms
        self._start = time.time()

    def elapsed_ms(self) -> float:
        return (time.time() - self._start) * 1000


class Backend(ABC):
    """Abstract base class for all execution backends.

    Concrete backends implement execute() and return a StepResult.
    """

    name: str = "base"
    priority: int = 0

    @abstractmethod
    def execute(
        self,
        step: "MacroStep",  # type: ignore[name-defined]
        params: dict,
        context: BackendContext,
    ) -> StepResult:
        """Execute a macro step.

        Args:
            step: The MacroStep definition being executed.
            params: Fully resolved (substituted) parameters.
            context: Runtime context with previous results and flags.

        Returns:
            StepResult describing success/failure and captured output.
        """

    def is_available(self) -> bool:
        """Return True if this backend can be used in the current environment."""
        return True

    def describe(self) -> dict:
        return {
            "name": self.name,
            "priority": self.priority,
            "available": self.is_available(),
        }

"""FileTransformBackend — read, transform, and write project files.

Supports XML (ElementTree), JSON, and plain text transformations.

Example macro step:

    - backend: file_transform
      action: json_set
      params:
        input_file: ${project_file}
        output_file: ${project_file}
        path: settings.grid_size
        value: 20

    - backend: file_transform
      action: xml_set_attr
      params:
        input_file: diagram.drawio
        output_file: diagram.drawio
        xpath: .//mxCell[@id='1']
        attr: style
        value: rounded=1;

    - backend: file_transform
      action: text_replace
      params:
        input_file: config.ini
        output_file: config.ini
        find: "theme=default"
        replace: "theme=dark"
"""

from __future__ import annotations

import json
import os
import time
from pathlib import Path

from cli_anything.macrocli.backends.base import Backend, BackendContext, StepResult
from cli_anything.macrocli.core.macro_model import MacroStep, substitute


class FileTransformBackend(Backend):
    """Transform project files without invoking the target application."""

    name = "file_transform"
    priority = 70

    def execute(self, step: MacroStep, params: dict, context: BackendContext) -> StepResult:
        t0 = time.time()
        action = step.action
        step_params = substitute(step.params, params)

        if context.dry_run:
            return StepResult(
                success=True,
                output={"dry_run": True, "action": action},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        dispatch = {
            "json_get": self._json_get,
            "json_set": self._json_set,
            "json_delete": self._json_delete,
            "xml_set_attr": self._xml_set_attr,
            "xml_get_attr": self._xml_get_attr,
            "text_replace": self._text_replace,
            "copy_file": self._copy_file,
        }

        handler = dispatch.get(action)
        if handler is None:
            return StepResult(
                success=False,
                error=f"FileTransformBackend: unknown action '{action}'.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        try:
            output = handler(step_params)
            return StepResult(
                success=True,
                output=output or {},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )
        except Exception as exc:
            return StepResult(
                success=False,
                error=f"FileTransformBackend.{action}: {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

    # ── JSON actions ─────────────────────────────────────────────────────

    def _json_get(self, p: dict) -> dict:
        """Read a value from a JSON file by dot-path."""
        data = self._load_json(p["input_file"])
        val = self._dotpath_get(data, p["path"])
        return {"value": val, "path": p["path"]}

    def _json_set(self, p: dict) -> dict:
        """Set a value in a JSON file by dot-path and write it back."""
        path = p.get("path", "")
        value = p["value"]
        data = self._load_json(p["input_file"]) if Path(p["input_file"]).is_file() else {}
        self._dotpath_set(data, path, value)
        self._save_json(p.get("output_file", p["input_file"]), data)
        return {"path": path, "value": value}

    def _json_delete(self, p: dict) -> dict:
        """Delete a key from a JSON file by dot-path."""
        data = self._load_json(p["input_file"])
        self._dotpath_delete(data, p["path"])
        self._save_json(p.get("output_file", p["input_file"]), data)
        return {"deleted": p["path"]}

    # ── XML actions ──────────────────────────────────────────────────────

    def _xml_set_attr(self, p: dict) -> dict:
        """Set an XML element attribute matched by XPath."""
        from xml.etree import ElementTree as ET
        from defusedxml.ElementTree import parse as _defused_parse
        tree = _defused_parse(p["input_file"])
        root = tree.getroot()
        elements = root.findall(p["xpath"])
        if not elements:
            raise ValueError(f"XPath matched nothing: {p['xpath']}")
        for el in elements:
            el.set(p["attr"], str(p["value"]))
        tree.write(p.get("output_file", p["input_file"]), encoding="unicode", xml_declaration=True)
        return {"matched": len(elements), "attr": p["attr"]}

    def _xml_get_attr(self, p: dict) -> dict:
        """Get an XML element attribute matched by XPath."""
        from xml.etree import ElementTree as ET
        from defusedxml.ElementTree import parse as _defused_parse
        tree = _defused_parse(p["input_file"])
        root = tree.getroot()
        elements = root.findall(p["xpath"])
        values = [el.get(p["attr"]) for el in elements]
        return {"values": values, "attr": p["attr"]}

    # ── Text actions ─────────────────────────────────────────────────────

    def _text_replace(self, p: dict) -> dict:
        """Simple find-and-replace in a text file."""
        content = Path(p["input_file"]).read_text(encoding="utf-8")
        count = content.count(p["find"])
        content = content.replace(p["find"], p["replace"])
        out = p.get("output_file", p["input_file"])
        Path(out).write_text(content, encoding="utf-8")
        return {"replacements": count}

    def _copy_file(self, p: dict) -> dict:
        """Copy a file from src to dst."""
        import shutil
        shutil.copy2(p["src"], p["dst"])
        size = os.path.getsize(p["dst"])
        return {"src": p["src"], "dst": p["dst"], "size": size}

    # ── Helpers ──────────────────────────────────────────────────────────

    def _load_json(self, path: str) -> dict:
        with open(path, encoding="utf-8") as f:
            return json.load(f)

    def _save_json(self, path: str, data: dict) -> None:
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

    def _dotpath_get(self, data: dict, path: str):
        keys = path.split(".")
        cur = data
        for k in keys:
            if isinstance(cur, dict) and k in cur:
                cur = cur[k]
            else:
                return None
        return cur

    def _dotpath_set(self, data: dict, path: str, value) -> None:
        keys = path.split(".")
        cur = data
        for k in keys[:-1]:
            if k not in cur or not isinstance(cur[k], dict):
                cur[k] = {}
            cur = cur[k]
        cur[keys[-1]] = value

    def _dotpath_delete(self, data: dict, path: str) -> None:
        keys = path.split(".")
        cur = data
        for k in keys[:-1]:
            if isinstance(cur, dict) and k in cur:
                cur = cur[k]
            else:
                return
        if isinstance(cur, dict) and keys[-1] in cur:
            del cur[keys[-1]]

"""GUIAgentBackend — execute a macro step by letting a vision model
look at the screen and decide what to do.

This backend is used for steps that cannot be expressed as fixed
coordinates or hotkeys because the interface state is unpredictable.
The macro author provides:
  - description:           what needs to be accomplished in this step
  - end_state_description: text description of the desired end state
  - end_state_snapshot:    screenshot of the desired end state (taken
                           by the macro author at recording time)

At runtime the backend:
  1. Takes a screenshot of the current screen
  2. Sends current screenshot + end_state_snapshot + description to the model
  3. Model returns the next action (click x,y / type text / hotkey)
  4. Executes the action
  5. Takes another screenshot
  6. Asks model: "have we reached the end state?"
  7. Loops until end state reached or max_steps exceeded

The backend uses the OpenAI SDK, which is compatible with any
OpenAI-compatible API provider (OpenAI, Azure, local vLLM, Ollama,
LiteLLM, etc.).  Configure model and endpoint via environment
variables or per-step YAML params:

  Environment variables:
    MACROCLI_MODEL    — model name (required, no default)
    MACROCLI_API_KEY  — API key
    MACROCLI_BASE_URL — base URL for non-OpenAI providers

Example YAML step:

    - id: select_png_format
      backend: gui_agent
      action: instruct
      params:
        description: >
          The export dialog is open. Find the Format dropdown and
          select PNG. Then ensure Resolution shows 300.
        end_state_description: >
          Format dropdown shows PNG, Resolution input shows 300.
        end_state_snapshot: snapshots/step_003_end_state.png
        max_steps: 8
        model: ${MACROCLI_MODEL}
        api_key: ${MACROCLI_API_KEY}
        base_url: ${MACROCLI_BASE_URL}
"""

from __future__ import annotations

import base64
import json
import os
import time
from pathlib import Path
from typing import Optional

from cli_anything.macrocli.backends.base import Backend, BackendContext, StepResult
from cli_anything.macrocli.core.macro_model import MacroStep, substitute

# ── Strict action space prompt ────────────────────────────────────────────────

_SYSTEM_PROMPT = """\
You are a GUI automation agent. You will be shown:
1. A screenshot of the CURRENT screen state
2. A screenshot of the TARGET end state (optional)
3. A description of what needs to be accomplished

Your job is to figure out what single action to take next.

OUTPUT FORMAT: Respond with ONLY a JSON object, one of:

  {"action": "click", "x": <int>, "y": <int>, "button": "left"}
  {"action": "double_click", "x": <int>, "y": <int>}
  {"action": "right_click", "x": <int>, "y": <int>}
  {"action": "drag", "from_x": <int>, "from_y": <int>, "to_x": <int>, "to_y": <int>, "duration_ms": 300}
  {"action": "type", "text": "<string>"}
  {"action": "hotkey", "keys": "<key1+key2+...>"}
  {"action": "scroll", "x": <int>, "y": <int>, "dy": <int>}
  {"action": "done"}

Use {"action": "done"} ONLY when the current state matches the target state.

RULES:
- Output RAW JSON ONLY. No markdown, no explanation.
- Use pixel coordinates from the CURRENT screenshot.
- For drag: from_x/from_y is where you start pressing, to_x/to_y is where you release.
- Prefer clicking on visible labeled controls over guessing coordinates.
- If the target state is already achieved, output {"action": "done"}.
- Never output any action not listed above.
"""

_CHECK_PROMPT = """\
Compare these two screenshots:
1. CURRENT state
2. TARGET end state

Has the current state reached the target end state?
Answer with ONLY: {"reached": true} or {"reached": false, "reason": "<brief reason>"}
"""


# ── Image helpers ─────────────────────────────────────────────────────────────

def _screenshot_b64() -> str:
    """Take a screenshot and return as base64 PNG string."""
    try:
        import mss
        from PIL import Image
        import io
        with mss.mss() as sct:
            monitor = sct.monitors[1]
            raw = sct.grab(monitor)
            img = Image.frombytes("RGB", raw.size, raw.bgra, "raw", "BGRX")
            buf = io.BytesIO()
            img.save(buf, format="PNG")
            return base64.b64encode(buf.getvalue()).decode("utf-8")
    except ImportError:
        raise ImportError("mss and Pillow required: pip install mss Pillow")


def _file_to_b64(path: str) -> str:
    with open(path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


# ── Action executor ───────────────────────────────────────────────────────────

def _execute_action(action_dict: dict, context: BackendContext) -> None:
    """Execute a single action returned by the model."""
    from cli_anything.macrocli.backends.visual_anchor import (
        _mouse_click, _mouse_drag, _require_pynput
    )

    action = action_dict.get("action", "")

    if action == "click":
        x, y = int(action_dict["x"]), int(action_dict["y"])
        _mouse_click(x, y, button=action_dict.get("button", "left"))

    elif action == "double_click":
        x, y = int(action_dict["x"]), int(action_dict["y"])
        _mouse_click(x, y, double=True)

    elif action == "right_click":
        x, y = int(action_dict["x"]), int(action_dict["y"])
        _mouse_click(x, y, button="right")

    elif action == "type":
        text = action_dict.get("text", "")
        _, keyboard_mod = _require_pynput()
        ctrl = keyboard_mod.Controller()
        for char in text:
            ctrl.press(char)
            ctrl.release(char)
            time.sleep(0.03)

    elif action == "hotkey":
        keys_str = action_dict.get("keys", "")
        _, keyboard_mod = _require_pynput()
        Key = keyboard_mod.Key
        ctrl = keyboard_mod.Controller()
        _KEY_MAP = {
            "ctrl": Key.ctrl, "shift": Key.shift, "alt": Key.alt,
            "enter": Key.enter, "tab": Key.tab, "esc": Key.esc,
            "escape": Key.esc, "space": Key.space, "backspace": Key.backspace,
        }
        keys = [_KEY_MAP.get(k.lower(), k) for k in keys_str.split("+")]
        for k in keys:
            ctrl.press(k)
        for k in reversed(keys):
            ctrl.release(k)

    elif action == "scroll":
        x, y = int(action_dict["x"]), int(action_dict["y"])
        dy = int(action_dict.get("dy", -3))
        mouse_mod, _ = _require_pynput()
        ctrl = mouse_mod.Controller()
        ctrl.position = (x, y)
        ctrl.scroll(0, dy)

    elif action == "drag":
        fx, fy = int(action_dict["from_x"]), int(action_dict["from_y"])
        tx, ty = int(action_dict["to_x"]), int(action_dict["to_y"])
        duration_ms = int(action_dict.get("duration_ms", 300))
        from cli_anything.macrocli.backends.visual_anchor import _mouse_drag
        _mouse_drag(fx, fy, tx, ty, duration_ms=duration_ms)

    elif action == "done":
        pass  # caller checks for done

    else:
        raise ValueError(f"GUIAgentBackend: unknown action '{action}'")


# ── Backend ───────────────────────────────────────────────────────────────────

class GUIAgentBackend(Backend):
    """Execute GUI steps using a vision model (OpenAI-compatible API) to decide actions."""

    name = "gui_agent"
    priority = 60  # between semantic_ui(50) and file_transform(70)

    def execute(
        self, step: MacroStep, params: dict, context: BackendContext
    ) -> StepResult:
        t0 = time.time()
        p = substitute(step.params, params)

        if context.dry_run:
            return StepResult(
                success=True,
                output={"dry_run": True, "action": step.action},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        if step.action == "instruct":
            return self._instruct(p, context, t0)
        elif step.action == "instruct_with_refine":
            return self._instruct_with_refine(p, context, t0)
        else:
            return StepResult(
                success=False,
                error=f"GUIAgentBackend: unknown action '{step.action}'. "
                      "Supported: 'instruct', 'instruct_with_refine'.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

    def is_available(self) -> bool:
        try:
            import openai  # noqa: F401
            import mss  # noqa: F401
            return True
        except ImportError:
            return False

    def _instruct(
        self, p: dict, context: BackendContext, t0: float
    ) -> StepResult:
        """Execute exactly ONE action decided by the vision model.

        The macro author is responsible for:
        - focusing the target window before calling gui_agent
        - writing multiple gui_agent steps if multiple actions are needed
        - verifying the outcome via postconditions or subsequent steps

        This step:
          1. Takes a screenshot
          2. Sends it + description + end_state_snapshot to the model
          3. Model returns one action (click/type/hotkey/scroll/done)
          4. Executes that action
          5. Returns success with the action taken
        """
        description: str = p.get("description", "")
        end_state_desc: str = p.get("end_state_description", "")
        snapshot_path: str = p.get("end_state_snapshot", "")
        window_title: str = p.get("window_title", "")  # focus this window first
        model_name: str = p.get("model", os.environ.get("MACROCLI_MODEL", ""))
        api_key: str = p.get("api_key", os.environ.get("MACROCLI_API_KEY", ""))
        base_url: str = p.get("base_url", os.environ.get("MACROCLI_BASE_URL", ""))

        if not model_name:
            return StepResult(
                success=False,
                error="GUIAgentBackend: model required. Set MACROCLI_MODEL env var or pass model in step params.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        if not api_key:
            return StepResult(
                success=False,
                error="GUIAgentBackend: api_key required. Set MACROCLI_API_KEY env var.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        try:
            from openai import OpenAI
        except ImportError:
            return StepResult(
                success=False,
                error="openai required: pip install openai",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        client_kwargs = {"api_key": api_key}
        if base_url:
            client_kwargs["base_url"] = base_url
        client = OpenAI(**client_kwargs)

        def _call_model(messages: list, max_tokens: int = 1024) -> str:
            resp = client.chat.completions.create(
                model=model_name,
                messages=messages,
                max_tokens=max_tokens,
            )
            return resp.choices[0].message.content.strip()

        def _extract_json(raw: str) -> dict:
            """Extract JSON from model output robustly."""
            if raw.startswith("```"):
                raw = "\n".join(
                    l for l in raw.split("\n") if not l.startswith("```")
                ).strip()
            start = raw.find('{')
            end = raw.rfind('}')
            if start != -1 and end != -1:
                raw = raw[start:end+1]
            return json.loads(raw)

        # Step 1: Focus the target window if specified
        if window_title and not context.dry_run:
            import shutil, subprocess
            env = os.environ.copy()
            if "DISPLAY" not in env:
                env["DISPLAY"] = ":0"
            if shutil.which("wmctrl"):
                subprocess.run(["wmctrl", "-a", window_title],
                               capture_output=True, env=env)
            elif shutil.which("xdotool"):
                subprocess.run(
                    ["xdotool", "search", "--name", window_title,
                     "windowfocus", "--sync"],
                    capture_output=True, env=env
                )
            time.sleep(0.3)

        if context.dry_run:
            return StepResult(
                success=True,
                output={"dry_run": True, "description": description},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        # Step 2: Take screenshot
        current_b64 = _screenshot_b64()

        # Step 3: Load end state snapshot if provided
        end_state_b64: Optional[str] = None
        if snapshot_path and Path(snapshot_path).is_file():
            end_state_b64 = _file_to_b64(snapshot_path)

        # Step 4: Build prompt
        content = []
        content.append({"type": "text", "text": "CURRENT SCREEN STATE:"})
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/png;base64,{current_b64}"}
        })
        if end_state_b64:
            content.append({"type": "text", "text": "TARGET END STATE:"})
            content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{end_state_b64}"}
            })
        task_text = f"TASK: {description}"
        if end_state_desc:
            task_text += f"\nTARGET: {end_state_desc}"
        task_text += "\nOutput ONE action as JSON only."
        content.append({"type": "text", "text": task_text})

        messages = [
            {"role": "system", "content": _SYSTEM_PROMPT},
            {"role": "user", "content": content},
        ]

        # Step 5: Ask model for one action
        try:
            raw = _call_model(messages)
        except Exception as exc:
            return StepResult(
                success=False,
                error=f"GUIAgentBackend: model error: {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        try:
            action_dict = _extract_json(raw)
        except json.JSONDecodeError:
            return StepResult(
                success=False,
                error=f"GUIAgentBackend: invalid JSON from model: {raw[:200]}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        action_name = action_dict.get("action", "")
        print(f"[gui_agent] action: {action_dict}", flush=True)

        # Step 6: Execute the action (unless model says done)
        if action_name != "done":
            try:
                _execute_action(action_dict, context)
            except Exception as exc:
                return StepResult(
                    success=False,
                    error=f"GUIAgentBackend: action execution failed: {exc}",
                    backend_used=self.name,
                    duration_ms=(time.time() - t0) * 1000,
                )
            time.sleep(0.5)

        return StepResult(
            success=True,
            output={
                "action": action_dict,
                "done": action_name == "done",
            },
            backend_used=self.name,
            duration_ms=(time.time() - t0) * 1000,
        )

    def _instruct_with_refine(
        self, p: dict, context: BackendContext, t0: float
    ) -> StepResult:
        """Execute one action, then compare result vs end_state_snapshot,
        and if needed undo and re-execute with a refined action.

        Sends to model on refine:
          - Screenshot BEFORE first action (original state)
          - The first action that was taken
          - Screenshot AFTER first action (current result)
          - end_state_snapshot (target state)
          - Request for corrected action

        This allows the model to see exactly what went wrong and correct it.
        """
        snapshot_path: str = p.get("end_state_snapshot", "")

        if not snapshot_path or not Path(snapshot_path).is_file():
            # No end state snapshot → fall back to single instruct
            return self._instruct(p, context, t0)

        # ── Round 1: initial action ───────────────────────────────────────────
        before_b64 = _screenshot_b64()
        result1 = self._instruct(p, context, t0)

        if not result1.success:
            return result1

        first_action = result1.output.get("action", {})
        if first_action.get("action") == "done":
            return result1

        time.sleep(0.5)
        after_b64 = _screenshot_b64()
        end_state_b64 = _file_to_b64(snapshot_path)

        # ── Round 2: compare and refine ───────────────────────────────────────
        description: str = p.get("description", "")
        end_state_desc: str = p.get("end_state_description", "")
        model_name: str = p.get("model", os.environ.get("MACROCLI_MODEL", ""))
        api_key: str = p.get("api_key", os.environ.get("MACROCLI_API_KEY", ""))
        base_url: str = p.get("base_url", os.environ.get("MACROCLI_BASE_URL", ""))

        from openai import OpenAI
        client_kwargs = {"api_key": api_key}
        if base_url:
            client_kwargs["base_url"] = base_url
        client = OpenAI(**client_kwargs)

        def _call(messages):
            resp = client.chat.completions.create(
                model=model_name, messages=messages, max_tokens=1024,
            )
            return resp.choices[0].message.content.strip()

        def _extract_json(raw):
            if raw.startswith("```"):
                raw = "\n".join(l for l in raw.split("\n") if not l.startswith("```")).strip()
            s, e = raw.find('{'), raw.rfind('}')
            if s != -1 and e != -1:
                raw = raw[s:e+1]
            return json.loads(raw)

        refine_prompt = f"""You are refining a GUI automation action.

ORIGINAL TASK: {description}
TARGET STATE: {end_state_desc}

WHAT HAPPENED:
- First action taken: {json.dumps(first_action)}

Now compare these three screenshots:

1. BEFORE (original state before any action):
2. AFTER FIRST ACTION (current result):
3. TARGET END STATE (what it should look like):

The first action was not quite right. Looking at:
- Where the rectangle was drawn vs where it should be
- The difference between AFTER and TARGET

Provide a corrected drag action with better coordinates.
Output ONE JSON action only."""

        content = [
            {"type": "text", "text": refine_prompt},
            {"type": "text", "text": "BEFORE:"},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{before_b64}"}},
            {"type": "text", "text": "AFTER FIRST ACTION:"},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{after_b64}"}},
            {"type": "text", "text": "TARGET END STATE:"},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{end_state_b64}"}},
        ]

        messages = [
            {"role": "system", "content": _SYSTEM_PROMPT},
            {"role": "user", "content": content},
        ]

        try:
            raw = _call(messages)
            refined_action = _extract_json(raw)
            print(f"[gui_agent] refined action: {refined_action}", flush=True)
        except Exception as exc:
            # Refine failed, return original result
            print(f"[gui_agent] refine failed: {exc}, keeping original", flush=True)
            return result1

        if refined_action.get("action") == "done":
            return result1

        # Undo the first action, then execute the refined one
        import shutil, subprocess as sp
        env = os.environ.copy()
        if "DISPLAY" not in env:
            env["DISPLAY"] = ":0"
        if shutil.which("xdotool"):
            sp.run(["xdotool", "key", "ctrl+z"], env=env)
            time.sleep(0.3)

        try:
            _execute_action(refined_action, context)
        except Exception as exc:
            return StepResult(
                success=False,
                error=f"GUIAgentBackend: refine action failed: {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        time.sleep(0.5)

        return StepResult(
            success=True,
            output={
                "action": refined_action,
                "first_action": first_action,
                "refined": True,
                "done": False,
            },
            backend_used=self.name,
            duration_ms=(time.time() - t0) * 1000,
        )

"""GUIMacroBackend — replay precompiled coordinate-based macro sequences.

A compiled macro is a JSON blob describing an exact sequence of mouse clicks,
key presses, and wait conditions. These are fast to execute but fragile to
layout changes.

Compiled macro format (stored separately, referenced by step params):

    {
      "version": 1,
      "screen_resolution": "1920x1080",
      "layout_hash": "abc123",
      "steps": [
        {"type": "click", "x": 100, "y": 200, "button": "left", "delay_ms": 200},
        {"type": "key", "keys": "ctrl+s", "delay_ms": 100},
        {"type": "type", "text": "output.png", "delay_ms": 50},
        {"type": "wait_file", "path": "/tmp/out.png", "timeout_ms": 5000},
        {"type": "sleep", "ms": 500}
      ]
    }

Example macro step:

    - backend: gui_macro
      action: replay
      params:
        macro_file: macros/compiled/export_png.json
        layout_strict: false   # if true, fail when screen res changes
"""

from __future__ import annotations

import json
import time
from pathlib import Path

from cli_anything.macrocli.backends.base import Backend, BackendContext, StepResult
from cli_anything.macrocli.core.macro_model import MacroStep, substitute


class GUIMacroBackend(Backend):
    """Replay precompiled GUI automation sequences."""

    name = "gui_macro"
    priority = 80

    def execute(self, step: MacroStep, params: dict, context: BackendContext) -> StepResult:
        t0 = time.time()
        action = step.action
        step_params = substitute(step.params, params)

        if context.dry_run:
            return StepResult(
                success=True,
                output={"dry_run": True, "action": action},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        if action != "replay":
            return StepResult(
                success=False,
                error=f"GUIMacroBackend: unknown action '{action}'. Expected 'replay'.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        return self._replay(step_params, context, t0)

    def is_available(self) -> bool:
        """Available when at least one automation library is present."""
        for lib in ("pyautogui", "pynput"):
            try:
                __import__(lib)
                return True
            except ImportError:
                pass
        return False

    def _replay(self, p: dict, context: BackendContext, t0: float) -> StepResult:
        """Load and replay a compiled macro file."""
        macro_file = p.get("macro_file", "")
        if not macro_file:
            return StepResult(
                success=False,
                error="GUIMacroBackend.replay: 'macro_file' param is required.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        macro_path = Path(macro_file)
        if not macro_path.is_file():
            return StepResult(
                success=False,
                error=f"GUIMacroBackend: compiled macro not found: {macro_file}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        try:
            with open(macro_path, encoding="utf-8") as f:
                macro_blob = json.load(f)
        except Exception as exc:
            return StepResult(
                success=False,
                error=f"GUIMacroBackend: failed to load macro file: {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        layout_strict: bool = p.get("layout_strict", False)
        if layout_strict:
            check = self._check_layout(macro_blob)
            if check:
                return StepResult(
                    success=False,
                    error=f"GUIMacroBackend: layout mismatch — {check}",
                    backend_used=self.name,
                    duration_ms=(time.time() - t0) * 1000,
                )

        try:
            steps_run = self._execute_steps(macro_blob.get("steps", []), context)
            return StepResult(
                success=True,
                output={"steps_executed": steps_run, "macro_file": macro_file},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )
        except Exception as exc:
            return StepResult(
                success=False,
                error=f"GUIMacroBackend.replay: {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

    def _check_layout(self, macro_blob: dict) -> str:
        """Return error string if current screen doesn't match expected."""
        expected_res = macro_blob.get("screen_resolution", "")
        if not expected_res:
            return ""
        try:
            import pyautogui
            w, h = pyautogui.size()
            current_res = f"{w}x{h}"
            if current_res != expected_res:
                return f"screen is {current_res}, macro expects {expected_res}"
        except ImportError:
            pass  # Can't verify — allow through
        return ""

    def _execute_steps(self, steps: list, context: BackendContext) -> int:
        """Execute each step in the compiled macro."""
        try:
            import pyautogui
            has_pyautogui = True
        except ImportError:
            has_pyautogui = False

        count = 0
        for s in steps:
            stype = s.get("type", "")
            delay = s.get("delay_ms", 100) / 1000.0

            if stype == "click":
                if not has_pyautogui:
                    raise ImportError("pyautogui required for click steps. pip install pyautogui")
                button = s.get("button", "left")
                pyautogui.click(s["x"], s["y"], button=button)

            elif stype == "key":
                if not has_pyautogui:
                    raise ImportError("pyautogui required for key steps. pip install pyautogui")
                keys = s.get("keys", "").split("+")
                if len(keys) == 1:
                    pyautogui.press(keys[0])
                else:
                    pyautogui.hotkey(*keys)

            elif stype == "type":
                if not has_pyautogui:
                    raise ImportError("pyautogui required for type steps. pip install pyautogui")
                pyautogui.typewrite(s.get("text", ""), interval=0.03)

            elif stype == "wait_file":
                deadline = time.time() + s.get("timeout_ms", 5000) / 1000.0
                path = s.get("path", "")
                while time.time() < deadline:
                    if Path(path).exists():
                        break
                    time.sleep(0.1)
                else:
                    raise TimeoutError(f"wait_file timed out: {path}")
                delay = 0  # no additional delay after file wait

            elif stype == "sleep":
                time.sleep(s.get("ms", 500) / 1000.0)
                delay = 0

            if delay > 0:
                time.sleep(delay)
            count += 1

        return count

"""NativeAPIBackend — executes macro steps via subprocess.

Supports these action types (configured in macro step params):

    action: run_command
    params:
      command: [inkscape, --export-filename, /tmp/out.png, input.svg]
      cwd: /optional/working/dir      # optional
      env: {KEY: value}               # optional extra env vars
      capture_stdout: true            # store stdout in output.stdout

    action: find_executable
    params:
      name: inkscape
      candidates: [inkscape, inkscape-1.0, /usr/bin/inkscape]
      install_hint: "apt install inkscape"
"""

from __future__ import annotations

import os
import shutil
import subprocess
import time
from typing import Any

from cli_anything.macrocli.backends.base import Backend, BackendContext, StepResult
from cli_anything.macrocli.core.macro_model import MacroStep, substitute


class NativeAPIBackend(Backend):
    """Execute a macro step by running an external command."""

    name = "native_api"
    priority = 100

    def execute(self, step: MacroStep, params: dict, context: BackendContext) -> StepResult:
        t0 = time.time()
        action = step.action

        if action == "find_executable":
            return self._find_executable(step, params, context, t0)
        elif action == "run_command":
            return self._run_command(step, params, context, t0)
        elif action == "start_process":
            return self._start_process(step, params, context, t0)
        else:
            return StepResult(
                success=False,
                error=f"NativeAPIBackend: unknown action '{action}'.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

    # ── Actions ──────────────────────────────────────────────────────────

    def _find_executable(
        self, step: MacroStep, params: dict, context: BackendContext, t0: float
    ) -> StepResult:
        """Check that an executable exists; return its path."""
        step_params = substitute(step.params, params)
        exe_name = step_params.get("name", "")
        candidates: list[str] = step_params.get("candidates", [exe_name] if exe_name else [])
        install_hint: str = step_params.get("install_hint", f"Install {exe_name}")

        for candidate in candidates:
            found = shutil.which(candidate)
            if found:
                return StepResult(
                    success=True,
                    output={"executable": found, "name": candidate},
                    backend_used=self.name,
                    duration_ms=(time.time() - t0) * 1000,
                )

        return StepResult(
            success=False,
            error=(
                f"Executable not found: {exe_name}. "
                f"Tried: {candidates}. "
                f"Install with: {install_hint}"
            ),
            backend_used=self.name,
            duration_ms=(time.time() - t0) * 1000,
        )

    def _run_command(
        self, step: MacroStep, params: dict, context: BackendContext, t0: float
    ) -> StepResult:
        """Run an external command."""
        step_params = substitute(step.params, params)
        command: list[str] = step_params.get("command", [])
        if not command:
            return StepResult(
                success=False,
                error="NativeAPIBackend.run_command: 'command' param is required.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        if isinstance(command, str):
            import shlex
            command = shlex.split(command)
        command = [str(c) for c in command]

        cwd: str = step_params.get("cwd", "")
        extra_env: dict = step_params.get("env", {})
        capture_stdout: bool = step_params.get("capture_stdout", False)

        env = os.environ.copy()
        if extra_env:
            env.update({k: str(v) for k, v in extra_env.items()})

        timeout_s = context.timeout_ms / 1000.0

        if context.dry_run:
            return StepResult(
                success=True,
                output={"dry_run": True, "command": command},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        try:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=timeout_s,
                cwd=cwd or None,
                env=env,
            )
        except FileNotFoundError as exc:
            return StepResult(
                success=False,
                error=f"Command not found: {command[0]}. {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )
        except subprocess.TimeoutExpired:
            return StepResult(
                success=False,
                error=f"Command timed out after {timeout_s:.0f}s: {command}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        duration = (time.time() - t0) * 1000
        if result.returncode != 0:
            return StepResult(
                success=False,
                error=(
                    f"Command failed (exit {result.returncode}): {command}\n"
                    f"stderr: {result.stderr.strip()}"
                ),
                output={"returncode": result.returncode, "stderr": result.stderr},
                backend_used=self.name,
                duration_ms=duration,
            )

        output: dict[str, Any] = {"returncode": 0}
        if capture_stdout:
            output["stdout"] = result.stdout
        return StepResult(
            success=True,
            output=output,
            backend_used=self.name,
            duration_ms=duration,
        )

    def _start_process(
        self, step: MacroStep, params: dict, context: BackendContext, t0: float
    ) -> StepResult:
        """Launch a GUI application in the background without waiting for it to exit.

        Use this instead of run_command for GUI apps like gedit, inkscape, etc.
        The process is detached immediately after launch.

        Params:
          command:    list[str] — the command to run
          cwd:        str       — working directory (optional)
          env:        dict      — extra environment variables (optional)
          log_file:   str       — redirect stdout+stderr here (default /dev/null)
        """
        import subprocess
        step_params = substitute(step.params, params)
        command: list[str] = step_params.get("command", [])
        if not command:
            return StepResult(
                success=False,
                error="NativeAPIBackend.start_process: 'command' param is required.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        if isinstance(command, str):
            import shlex
            command = shlex.split(command)
        command = [str(c) for c in command]

        cwd: str = step_params.get("cwd", "")
        extra_env: dict = step_params.get("env", {})
        log_file: str = step_params.get("log_file", "/dev/null")

        env = os.environ.copy()
        if extra_env:
            env.update({k: str(v) for k, v in extra_env.items()})

        if context.dry_run:
            return StepResult(
                success=True,
                output={"dry_run": True, "command": command},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        try:
            with open(log_file, "a") as log:
                proc = subprocess.Popen(
                    command,
                    stdout=log,
                    stderr=log,
                    cwd=cwd or None,
                    env=env,
                    start_new_session=True,   # detach from current process group
                )
            return StepResult(
                success=True,
                output={"pid": proc.pid, "command": command},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )
        except FileNotFoundError as exc:
            return StepResult(
                success=False,
                error=f"Command not found: {command[0]}. {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )
        except Exception as exc:
            return StepResult(
                success=False,
                error=f"start_process failed: {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

"""RecoveryBackend — retry and fallback orchestration.

This backend wraps another backend and retries failed steps with exponential
backoff. It can also fall back to an alternative backend on exhausted retries.

Example macro step using recovery explicitly:

    - backend: recovery
      action: retry_with_fallback
      params:
        primary_backend: native_api
        fallback_backend: file_transform
        max_retries: 3
        backoff_ms: [1000, 2000, 5000]
        step:
          action: run_command
          params:
            command: [inkscape, --export-filename, ${output}, input.svg]

The MacroRuntime also uses the RecoveryBackend automatically when a step
specifies retry_max > 0 in the macro definition.
"""

from __future__ import annotations

import time

from cli_anything.macrocli.backends.base import Backend, BackendContext, StepResult
from cli_anything.macrocli.core.macro_model import MacroStep, substitute


class RecoveryBackend(Backend):
    """Retry and fallback orchestration backend."""

    name = "recovery"
    priority = 10   # lowest — last resort

    def __init__(self, backends: dict[str, "Backend"] | None = None):
        """
        Args:
            backends: Dict of backend_name -> Backend instance.
                      Injected by the RoutingEngine at runtime.
        """
        self._backends = backends or {}

    def register_backend(self, backend: "Backend") -> None:
        self._backends[backend.name] = backend

    def execute(self, step: MacroStep, params: dict, context: BackendContext) -> StepResult:
        t0 = time.time()
        action = step.action

        if action not in ("retry", "retry_with_fallback"):
            return StepResult(
                success=False,
                error=f"RecoveryBackend: unknown action '{action}'. "
                      "Use 'retry' or 'retry_with_fallback'.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        step_params = substitute(step.params, params)
        inner_step_raw = step_params.get("step", {})
        if not inner_step_raw:
            return StepResult(
                success=False,
                error="RecoveryBackend: 'step' param is required.",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        # Build an inner MacroStep from the nested definition
        inner_step = MacroStep(
            id=inner_step_raw.get("id", "recovery_inner"),
            backend=step_params.get("primary_backend", inner_step_raw.get("backend", "native_api")),
            action=inner_step_raw.get("action", ""),
            params=inner_step_raw.get("params", {}),
            timeout_ms=context.timeout_ms,
        )

        max_retries: int = int(step_params.get("max_retries", step.retry_max or 2))
        backoff_ms: list[int] = step_params.get("backoff_ms", [1000, 2000, 5000])
        fallback_name: str = step_params.get("fallback_backend", "")

        last_result = None
        for attempt in range(max_retries + 1):
            backend = self._backends.get(inner_step.backend)
            if backend is None:
                return StepResult(
                    success=False,
                    error=f"RecoveryBackend: backend '{inner_step.backend}' not registered.",
                    backend_used=self.name,
                    duration_ms=(time.time() - t0) * 1000,
                )

            last_result = backend.execute(inner_step, params, context)
            if last_result.success:
                last_result.backend_used = f"{self.name}({inner_step.backend}, attempt={attempt + 1})"
                return last_result

            # Failed — decide whether to retry or fall back
            if attempt < max_retries:
                wait = backoff_ms[min(attempt, len(backoff_ms) - 1)] / 1000.0
                time.sleep(wait)
            elif fallback_name and fallback_name in self._backends:
                # Switch to fallback backend for one final attempt
                inner_step = MacroStep(
                    id=inner_step.id,
                    backend=fallback_name,
                    action=inner_step.action,
                    params=inner_step.params,
                    timeout_ms=inner_step.timeout_ms,
                )
                fallback_result = self._backends[fallback_name].execute(inner_step, params, context)
                if fallback_result.success:
                    fallback_result.backend_used = f"{self.name}(fallback={fallback_name})"
                    return fallback_result
                last_result = fallback_result

        if last_result is None:
            last_result = StepResult(
                success=False,
                error="RecoveryBackend: no attempts made.",
                backend_used=self.name,
            )
        last_result.duration_ms = (time.time() - t0) * 1000
        last_result.backend_used = self.name
        return last_result

"""SemanticUIBackend — drive applications via accessibility APIs and keyboard shortcuts.

Backends by platform:
  Linux:   AT-SPI via python3-pyatspi  (apt install python3-pyatspi)
           Fallback: xdotool for keyboard/shortcuts
  macOS:   ApplicationServices / Quartz via pyobjc
           Fallback: osascript (AppleScript)
  Windows: UI Automation via pywinauto  (pip install pywinauto)

Action space:

  shortcut        — send keyboard shortcut to focused window
  type_text       — type text into focused control
  menu_click      — activate a menu item by path
  button_click    — click a button by label/role
  wait_for_window — wait for a window with given title to appear
  focus_window    — bring a window to foreground
  get_controls    — list interactive controls (for discovery)

Example YAML steps:

    - backend: semantic_ui
      action: menu_click
      params:
        menu_path: [File, Export As, PNG Image]

    - backend: semantic_ui
      action: shortcut
      params:
        keys: ctrl+shift+e

    - backend: semantic_ui
      action: wait_for_window
      params:
        title_contains: Export
        timeout_ms: 5000

    - backend: semantic_ui
      action: button_click
      params:
        label: OK

    - backend: semantic_ui
      action: focus_window
      params:
        title_contains: Inkscape

    - backend: semantic_ui
      action: get_controls
      params:
        window_title: Inkscape
"""

from __future__ import annotations

import os
import platform
import shutil
import subprocess
import time
from typing import Optional

from cli_anything.macrocli.backends.base import Backend, BackendContext, StepResult
from cli_anything.macrocli.core.macro_model import MacroStep, substitute

_SYSTEM = platform.system()


def _x_env() -> dict:
    """Return env dict with DISPLAY set, for subprocess calls to X tools."""
    env = os.environ.copy()
    if "DISPLAY" not in env:
        env["DISPLAY"] = ":0"
    return env


# ── AT-SPI helpers (Linux) ────────────────────────────────────────────────────

def _atspi_available() -> bool:
    try:
        import pyatspi  # noqa: F401
        return True
    except ImportError:
        return False


def _atspi_find_app(name_fragment: str):
    """Return the first AT-SPI application matching name_fragment."""
    import pyatspi
    desktop = pyatspi.Registry.getDesktop(0)
    for app in desktop:
        if app and name_fragment.lower() in (app.name or "").lower():
            return app
    return None


def _atspi_find_control(root, role_name: str, label_fragment: str, max_depth: int = 20):
    """BFS search for a control by role and label."""
    import pyatspi
    role_map = {
        "button": pyatspi.ROLE_PUSH_BUTTON,
        "menu": pyatspi.ROLE_MENU,
        "menu_item": pyatspi.ROLE_MENU_ITEM,
        "menu_bar": pyatspi.ROLE_MENU_BAR,
        "text": pyatspi.ROLE_TEXT,
        "combo_box": pyatspi.ROLE_COMBO_BOX,
        "check_box": pyatspi.ROLE_CHECK_BOX,
        "radio": pyatspi.ROLE_RADIO_BUTTON,
        "list_item": pyatspi.ROLE_LIST_ITEM,
        "dialog": pyatspi.ROLE_DIALOG,
        "window": pyatspi.ROLE_FRAME,
    }
    target_role = role_map.get(role_name.lower())

    from collections import deque
    queue = deque([(root, 0)])
    while queue:
        node, depth = queue.popleft()
        if depth > max_depth:
            continue
        try:
            node_role = node.getRole()
            node_name = node.name or ""
            if (target_role is None or node_role == target_role):
                if label_fragment.lower() in node_name.lower():
                    return node
            for i in range(node.childCount):
                child = node.getChildAtIndex(i)
                if child:
                    queue.append((child, depth + 1))
        except Exception:
            continue
    return None


def _atspi_menu_path(app, menu_path: list[str]):
    """Navigate a menu path and activate the final item."""
    import pyatspi

    # Find the menu bar
    menu_bar = _atspi_find_control(app, "menu_bar", "", max_depth=3)
    if menu_bar is None:
        raise RuntimeError("AT-SPI: menu bar not found in application.")

    current = menu_bar
    for label in menu_path:
        item = _atspi_find_control(current, "menu", label)
        if item is None:
            item = _atspi_find_control(current, "menu_item", label)
        if item is None:
            raise RuntimeError(f"AT-SPI: menu item '{label}' not found.")
        # Activate / click
        try:
            action = item.queryAction()
            for i in range(action.nActions):
                if action.getName(i).lower() in ("click", "activate", "open"):
                    action.doAction(i)
                    break
        except Exception:
            pass
        current = item
        time.sleep(0.15)

    return True


# ── xdotool helpers (Linux fallback) ─────────────────────────────────────────

def _xdotool_key(keys: str) -> None:
    if not shutil.which("xdotool"):
        raise RuntimeError("xdotool not found. Install with: apt install xdotool")
    # ctrl+shift+e → ctrl+shift+e (xdotool accepts this format directly)
    subprocess.run(["xdotool", "key", "--clearmodifiers", keys], check=True, env=_x_env())


def _xdotool_type(text: str) -> None:
    if not shutil.which("xdotool"):
        raise RuntimeError("xdotool not found. Install with: apt install xdotool")
    subprocess.run(["xdotool", "type", "--clearmodifiers", "--delay", "30", text], check=True, env=_x_env())


def _xdotool_focus(title: str) -> None:
    if not shutil.which("xdotool"):
        raise RuntimeError("xdotool not found. Install with: apt install xdotool")
    subprocess.run(
        ["xdotool", "search", "--name", title, "windowfocus", "--sync"],
        check=True, env=_x_env(),
    )


# ── osascript helpers (macOS) ─────────────────────────────────────────────────

def _osascript(script: str) -> str:
    r = subprocess.run(
        ["osascript", "-e", script], capture_output=True, text=True
    )
    if r.returncode != 0:
        raise RuntimeError(f"osascript failed: {r.stderr.strip()}")
    return r.stdout.strip()


def _macos_menu_click(app_name: str, menu_path: list[str]) -> None:
    if len(menu_path) < 2:
        raise ValueError("menu_path needs at least 2 elements (menu name + item).")
    menu_name = menu_path[0]
    items = menu_path[1:]
    # Build nested AppleScript path
    item_script = " of menu ".join(
        [f'menu item "{i}"' for i in reversed(items)]
    )
    script = f"""
    tell application "{app_name}"
        activate
    end tell
    tell application "System Events"
        tell process "{app_name}"
            click {item_script} of menu "{menu_name}" of menu bar 1
        end tell
    end tell
    """
    _osascript(script)


# ── pywinauto helpers (Windows) ───────────────────────────────────────────────

def _win_find_app(title_fragment: str):
    from pywinauto import Application, findwindows
    handles = findwindows.find_windows(title_re=f".*{title_fragment}.*")
    if not handles:
        raise RuntimeError(f"Window not found: '{title_fragment}'")
    app = Application().connect(handle=handles[0])
    return app.window(handle=handles[0])


# ── Backend ───────────────────────────────────────────────────────────────────

class SemanticUIBackend(Backend):
    """Drive applications through semantic (accessibility) controls."""

    name = "semantic_ui"
    priority = 50

    def execute(self, step: MacroStep, params: dict, context: BackendContext) -> StepResult:
        t0 = time.time()
        action = step.action
        p = substitute(step.params, params)

        if context.dry_run:
            return StepResult(
                success=True,
                output={"dry_run": True, "action": action},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        dispatch = {
            "shortcut":        self._shortcut,
            "type_text":       self._type_text,
            "menu_click":      self._menu_click,
            "button_click":    self._button_click,
            "wait_for_window": self._wait_for_window,
            "focus_window":    self._focus_window,
            "get_controls":    self._get_controls,
        }

        handler = dispatch.get(action)
        if handler is None:
            return StepResult(
                success=False,
                error=f"SemanticUIBackend: unknown action '{action}'. "
                      f"Available: {sorted(dispatch)}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

        try:
            output = handler(p, context)
            return StepResult(
                success=True,
                output=output or {},
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )
        except Exception as exc:
            return StepResult(
                success=False,
                error=f"SemanticUIBackend.{action}: {exc}",
                backend_used=self.name,
                duration_ms=(time.time() - t0) * 1000,
            )

    def is_available(self) -> bool:
        if _SYSTEM == "Linux":
            return _atspi_available() or bool(shutil.which("xdotool"))
        elif _SYSTEM == "Darwin":
            return bool(shutil.which("osascript"))
        elif _SYSTEM == "Windows":
            try:
                import pywinauto  # noqa: F401
                return True
            except ImportError:
                return False
        return False

    # ── shortcut ─────────────────────────────────────────────────────────────

    def _shortcut(self, p: dict, context: BackendContext) -> dict:
        keys: str = p.get("keys", "")
        if not keys:
            raise ValueError("shortcut requires 'keys' param.")

        if _SYSTEM == "Linux":
            _xdotool_key(keys)
            return {"keys": keys, "method": "xdotool"}

        elif _SYSTEM == "Darwin":
            # Use pynput (cross-platform) or AppleScript key code
            from cli_anything.macrocli.backends.visual_anchor import VisualAnchorBackend
            from cli_anything.macrocli.core.macro_model import MacroStep as MS
            va = VisualAnchorBackend()
            step = MS(id="x", backend="visual_anchor", action="hotkey", params={"keys": keys})
            result = va._hotkey({"keys": keys}, context)
            return result

        elif _SYSTEM == "Windows":
            import pywinauto.keyboard as kb
            # Convert ctrl+s → {VK_CONTROL}s
            kb.send_keys(keys.replace("+", ""))
            return {"keys": keys, "method": "pywinauto"}

        raise NotImplementedError(f"shortcut not implemented for {_SYSTEM}")

    # ── type_text ─────────────────────────────────────────────────────────────

    def _type_text(self, p: dict, context: BackendContext) -> dict:
        text: str = p.get("text", "")
        if not text:
            raise ValueError("type_text requires 'text' param.")

        if _SYSTEM == "Linux":
            _xdotool_type(text)
            return {"typed": len(text), "method": "xdotool"}

        # macOS / Windows: fall through to visual_anchor type_text
        from cli_anything.macrocli.backends.visual_anchor import VisualAnchorBackend
        va = VisualAnchorBackend()
        return va._type_text(p, context)

    # ── menu_click ────────────────────────────────────────────────────────────

    def _menu_click(self, p: dict, context: BackendContext) -> dict:
        menu_path: list = p.get("menu_path", [])
        app_name: str = p.get("app_name", "")

        if not menu_path:
            raise ValueError("menu_click requires 'menu_path' param (list of strings).")

        if _SYSTEM == "Linux":
            if _atspi_available():
                if not app_name:
                    raise ValueError(
                        "menu_click on Linux AT-SPI requires 'app_name' param."
                    )
                app = _atspi_find_app(app_name)
                if app is None:
                    raise RuntimeError(f"AT-SPI: application '{app_name}' not found.")
                _atspi_menu_path(app, menu_path)
                return {"menu_path": menu_path, "method": "at-spi"}
            else:
                raise RuntimeError(
                    "menu_click on Linux requires AT-SPI.\n"
                    "  apt install python3-pyatspi\n"
                    "  Or use visual_anchor backend instead."
                )

        elif _SYSTEM == "Darwin":
            if not app_name:
                raise ValueError("menu_click on macOS requires 'app_name' param.")
            _macos_menu_click(app_name, menu_path)
            return {"menu_path": menu_path, "method": "osascript"}

        elif _SYSTEM == "Windows":
            if not app_name:
                raise ValueError("menu_click on Windows requires 'app_name' param.")
            win = _win_find_app(app_name)
            # pywinauto menu navigation
            menu = win.menu()
            for item in menu_path:
                menu = menu.item_by_path(item)
            menu.click_input()
            return {"menu_path": menu_path, "method": "pywinauto"}

        raise NotImplementedError(f"menu_click not implemented for {_SYSTEM}")

    # ── button_click ──────────────────────────────────────────────────────────

    def _button_click(self, p: dict, context: BackendContext) -> dict:
        label: str = p.get("label", "")
        app_name: str = p.get("app_name", "")

        if not label:
            raise ValueError("button_click requires 'label' param.")

        if _SYSTEM == "Linux" and _atspi_available():
            if not app_name:
                raise ValueError("button_click on Linux AT-SPI requires 'app_name'.")
            app = _atspi_find_app(app_name)
            if app is None:
                raise RuntimeError(f"AT-SPI: application '{app_name}' not found.")
            btn = _atspi_find_control(app, "button", label)
            if btn is None:
                raise RuntimeError(f"AT-SPI: button '{label}' not found in '{app_name}'.")
            action = btn.queryAction()
            for i in range(action.nActions):
                if action.getName(i).lower() == "click":
                    action.doAction(i)
                    return {"clicked": label, "method": "at-spi"}
            raise RuntimeError(f"AT-SPI: no click action on button '{label}'.")

        elif _SYSTEM == "Darwin":
            script = f"""
            tell application "System Events"
                click button "{label}" of front window of (first process whose frontmost is true)
            end tell
            """
            _osascript(script)
            return {"clicked": label, "method": "osascript"}

        elif _SYSTEM == "Windows":
            win = _win_find_app(app_name or "")
            win.child_window(title=label, control_type="Button").click_input()
            return {"clicked": label, "method": "pywinauto"}

        raise NotImplementedError(
            f"button_click not fully implemented for {_SYSTEM} without AT-SPI.\n"
            "Use visual_anchor backend as fallback."
        )

    # ── wait_for_window ───────────────────────────────────────────────────────

    def _wait_for_window(self, p: dict, context: BackendContext) -> dict:
        title: str = p.get("title_contains", "")
        timeout_ms: int = int(p.get("timeout_ms", 5000))
        poll_ms: int = int(p.get("poll_ms", 300))

        if not title:
            raise ValueError("wait_for_window requires 'title_contains' param.")

        deadline = time.time() + timeout_ms / 1000.0

        if _SYSTEM == "Linux":
            while time.time() < deadline:
                if shutil.which("wmctrl"):
                    r = subprocess.run(
                        ["wmctrl", "-l"], capture_output=True, text=True, env=_x_env()
                    )
                    if title.lower() in r.stdout.lower():
                        return {"found": title, "method": "wmctrl"}
                elif shutil.which("xdotool"):
                    r = subprocess.run(
                        ["xdotool", "search", "--name", title],
                        capture_output=True, text=True, env=_x_env(),
                    )
                    if r.returncode == 0 and r.stdout.strip():
                        return {"found": title, "method": "xdotool"}
                time.sleep(poll_ms / 1000.0)

        elif _SYSTEM == "Darwin":
            while time.time() < deadline:
                script = f"""
                tell application "System Events"
                    set ws to name of every window of every process
                    set found to false
                    repeat with wlist in ws
                        repeat with wname in wlist
                            if "{title}" is in (wname as text) then
                                set found to true
                            end if
                        end repeat
                    end repeat
                    return found
                end tell
                """
                result = _osascript(script)
                if result.lower() == "true":
                    return {"found": title, "method": "osascript"}
                time.sleep(poll_ms / 1000.0)

        elif _SYSTEM == "Windows":
            import pywinauto.findwindows as fw
            while time.time() < deadline:
                try:
                    handles = fw.find_windows(title_re=f".*{title}.*")
                    if handles:
                        return {"found": title, "method": "pywinauto"}
                except Exception:
                    pass
                time.sleep(poll_ms / 1000.0)

        raise TimeoutError(
            f"wait_for_window: window containing '{title}' did not appear "
            f"within {timeout_ms}ms."
        )

    # ── focus_window ──────────────────────────────────────────────────────────

    def _focus_window(self, p: dict, context: BackendContext) -> dict:
        title: str = p.get("title_contains", "")
        if not title:
            raise ValueError("focus_window requires 'title_contains' param.")

        if _SYSTEM == "Linux":
            if shutil.which("wmctrl"):
                subprocess.run(["wmctrl", "-a", title], check=True, env=_x_env())
                return {"focused": title, "method": "wmctrl"}
            _xdotool_focus(title)
            return {"focused": title, "method": "xdotool"}

        elif _SYSTEM == "Darwin":
            _osascript(f'tell application "{title}" to activate')
            return {"focused": title, "method": "osascript"}

        elif _SYSTEM == "Windows":
            win = _win_find_app(title)
            win.set_focus()
            return {"focused": title, "method": "pywinauto"}

        raise NotImplementedError(f"focus_window not implemented for {_SYSTEM}")

    # ── get_controls ──────────────────────────────────────────────────────────

    def _get_controls(self, p: dict, context: BackendContext) -> dict:
        """List interactive controls in a window (for macro authoring / discovery)."""
        window_title: str = p.get("window_title", "")
        max_depth: int = int(p.get("max_depth", 5))

        if _SYSTEM == "Linux" and _atspi_available():
            import pyatspi
            app = _atspi_find_app(window_title) if window_title else None
            root = app or pyatspi.Registry.getDesktop(0)
            controls = []
            interactive_roles = {
                pyatspi.ROLE_PUSH_BUTTON,
                pyatspi.ROLE_MENU,
                pyatspi.ROLE_MENU_ITEM,
                pyatspi.ROLE_TEXT,
                pyatspi.ROLE_COMBO_BOX,
                pyatspi.ROLE_CHECK_BOX,
                pyatspi.ROLE_RADIO_BUTTON,
                pyatspi.ROLE_TOGGLE_BUTTON,
            }
            from collections import deque
            queue = deque([(root, 0)])
            while queue:
                node, depth = queue.popleft()
                if depth > max_depth:
                    continue
                try:
                    if node.getRole() in interactive_roles:
                        controls.append({
                            "role": node.getRoleName(),
                            "name": node.name,
                        })
                    for i in range(node.childCount):
                        child = node.getChildAtIndex(i)
                        if child:
                            queue.append((child, depth + 1))
                except Exception:
                    continue
            return {"controls": controls, "count": len(controls)}

        elif _SYSTEM == "Windows":
            win = _win_find_app(window_title)
            controls = []
            for ctrl in win.descendants():
                try:
                    controls.append({
                        "role": ctrl.element_info.control_type,
                        "name": ctrl.element_info.name,
                    })
                except Exception:
                    pass
            return {"controls": controls, "count": len(controls)}

        raise NotImplementedError(
            f"get_controls not implemented for {_SYSTEM} without AT-SPI / pywinauto."
        )