# Test Case: <title>

## Metadata
- App / Build:
- Windows:
- Date:
- Tester:
- Type: Smoke / Regression / Exploratory

## Preconditions
- ...

## Steps / Expected / Assertions
1) Action:
   Expected:
   Assertion (tool + control):
2) ...

## Artifacts
- Screenshot before:
- Screenshot after:

## Verdict
PASS / FAIL

## Stability Notes
- Controls used (ids + names):
- Any coordinate fallback + why:

# QA Workflows

## Table of Contents
- [Smoke Test (Login Example)](#smoke-test-login-example)
- [Locator Strategy](#locator-strategy)
- [Common Patterns](#common-patterns)

## Smoke Test (Login Example)

```
1) qa_refresh_and_list_windows()
2) select_application_window(id, name)
3) capture_window_screenshot()
4) get_app_window_controls_info(field_list=["label","control_text","control_type","automation_id","control_rect"])
5) set_edit_text(id, name, "username_value")
6) set_edit_text(id, name, "password_value")
7) click_input(id, name)   # login button
8) qa_wait_for_text_contains(id, name, "Welcome", timeout_s=10)
9) capture_window_screenshot()
```

## Locator Strategy

1. Prefer control IDs from `get_app_window_controls_info`
2. Use `automation_id` as a human-readable secondary anchor when available
3. Re-collect controls after navigation or dialog open (control tree changes)
4. Use coordinate actions only as last resort, with a comment explaining why

## Common Patterns

**Form fill + submit**: collect controls -> set_edit_text for each field -> click_input on submit -> assert result text

**Navigation**: click menu/tab -> re-collect controls -> verify new view via screenshot + texts()

**Dialog handling**: after triggering dialog, re-collect controls (new control tree) -> interact -> close dialog -> re-collect parent window controls

**Data grid verification**: collect controls -> find grid cells by control_text -> compare against expected values

# Setup: UFO + MCP Server

## Table of Contents
- [Install UFO](#install-ufo)
- [Configure MCP in Claude Code](#configure-mcp-in-claude-code)
- [Verify](#verify)
- [Backend Selection](#backend-selection)

## Install UFO

Prerequisites: Windows 11 or Windows Server 2025 with Desktop Experience, Python 3.10.x, and Git. The pinned UFO requirements currently need Python 3.10 on Windows: `faiss-cpu==1.8.0` is unavailable for Python 3.13 and `pandas==1.4.3` is unavailable for Python 3.11.

```powershell
cd $env:USERPROFILE
git clone https://github.com/microsoft/UFO.git
cd UFO
py -3.10 -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
```

Quick check:
```powershell
python -c "from ufo.client.mcp.local_servers import load_all_servers; load_all_servers(); print('OK')"
python -c "from ufo.client.mcp.mcp_registry import MCPRegistry; print(MCPRegistry.list())"
```

Expected: UICollector, HostUIExecutor, AppUIExecutor registered.

## Configure MCP in Claude Code

Add to your project `.mcp.json`:

```json
{
  "mcpServers": {
    "ufo-windows-qa": {
      "type": "stdio",
      "command": "C:\\Users\\<you>\\UFO\\.venv\\Scripts\\python.exe",
      "args": [
        "C:\\path\\to\\windows-qa-engineer\\scripts\\ufo_windows_qa_mcp_server.py"
      ],
      "env": {
        "UFO_ROOT": "C:\\Users\\<you>\\UFO",
        "PYTHONPATH": "C:\\Users\\<you>\\UFO",
        "CONTROL_BACKEND": "uia",
        "MAXIMIZE_WINDOW": "false",
        "SHOW_VISUAL_OUTLINE_ON_SCREEN": "true",
        "RUN_CONFIGS": "true"
      }
    }
  }
}
```

Replace `<you>` and the skill path with absolute paths, restart the MCP client, and confirm that the tools are available.

## Verify

Run the doctor script from any directory; it selects UFO's venv and working directory automatically:
```powershell
.\.claude\skills\windows-qa-engineer\scripts\doctor.ps1
```

Or check in Claude Code that these tools appear:
`get_desktop_app_info`, `select_application_window`, `get_app_window_controls_info`,
`click_input`, `set_edit_text`, `texts`, `capture_window_screenshot`,
`qa_refresh_and_list_windows`, `qa_refresh_controls`, `qa_wait_for_text_contains`.

The doctor verifies imports and MCP registration but cannot create an interactive desktop. For an end-to-end check, run the MCP client and the application under test in the same logged-in Windows session, call `qa_refresh_and_list_windows()`, select a visible app such as Notepad, and capture its screenshot. An empty window list from SSH or a service session is expected and does not verify desktop automation.

The bundled smoke test performs that complete Notepad workflow and writes a PNG plus JSON report under the user's temporary directory:

```powershell
& "$env:USERPROFILE\UFO\.venv\Scripts\python.exe" `
  "<skill-dir>\scripts\interactive_smoke_test.py" `
  --project-dir "<project-root>"
```

Run it from a terminal inside the same interactive desktop session, not through SSH or a service.

## Backend Selection

- `CONTROL_BACKEND=uia` (default, recommended for WinForms/WPF stability)
- `CONTROL_BACKEND=win32` (fallback if UIA fails for a specific SUT)

Set via the `env` block in `.mcp.json`.

#!/usr/bin/env python3
"""Exercise UFO discovery, window selection, and capture against Windows Notepad."""

from __future__ import annotations

import argparse
import asyncio
import base64
import json
import subprocess
import tempfile
from enum import Enum
from pathlib import Path
from typing import Any

from fastmcp import Client


def result_data(result: Any) -> Any:
    structured = getattr(result, "structured_content", None)
    if isinstance(structured, dict) and set(structured) == {"result"}:
        return structured["result"]
    if structured is not None:
        return structured
    return getattr(result, "data", None)


def to_plain(value: Any) -> Any:
    if value is None or isinstance(value, (bool, int, float, str)):
        return value
    if isinstance(value, Enum):
        return to_plain(value.value)
    if isinstance(value, Path):
        return str(value)
    root = getattr(value, "root", None)
    if root is not None:
        return to_plain(root)
    if hasattr(value, "model_dump"):
        return to_plain(value.model_dump())
    if isinstance(value, dict):
        return {str(key): to_plain(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [to_plain(item) for item in value]
    if isinstance(value, type):
        return f"{value.__module__}.{value.__qualname__}"
    attributes = getattr(value, "__dict__", None)
    if isinstance(attributes, dict):
        public_attributes = {
            str(key): to_plain(item)
            for key, item in attributes.items()
            if not str(key).startswith("_")
        }
        if public_attributes:
            return public_attributes
    return str(value)


async def run(project_dir: Path, output_dir: Path) -> dict[str, Any]:
    config_path = project_dir / ".mcp.json"
    if not config_path.exists():
        raise FileNotFoundError(f"MCP configuration not found: {config_path}")

    output_dir.mkdir(parents=True, exist_ok=True)
    screenshot_path = output_dir / "notepad.png"
    report_path = output_dir / "report.json"
    notepad = subprocess.Popen(["notepad.exe"])
    report: dict[str, Any] = {"success": False}

    try:
        config = json.loads(config_path.read_text(encoding="utf-8"))
        async with Client(config, timeout=45) as client:
            await asyncio.sleep(2)
            tools = await client.list_tools()
            report["tool_count"] = len(tools)

            discovery = await client.call_tool("qa_refresh_and_list_windows", {})
            windows = to_plain(result_data(discovery) or [])
            report["windows"] = windows
            match = next(
                (
                    window
                    for window in windows
                    if "notepad" in json.dumps(window, ensure_ascii=False).lower()
                ),
                None,
            )
            if not isinstance(match, dict):
                raise RuntimeError("Notepad was not found in the interactive window list")

            window_id = str(match.get("id", ""))
            window_name = str(match.get("name", ""))
            if not window_id or not window_name:
                raise RuntimeError(f"Unsupported window record: {match!r}")

            await client.call_tool(
                "select_application_window", {"id": window_id, "name": window_name}
            )
            capture = await client.call_tool("capture_window_screenshot", {})
            encoded = to_plain(result_data(capture))
            if not isinstance(encoded, str):
                raise RuntimeError("UFO returned no screenshot string")
            if encoded.startswith("data:image/"):
                encoded = encoded.split(",", 1)[1]
            screenshot_path.write_bytes(base64.b64decode(encoded, validate=True))

            report.update(
                {
                    "success": True,
                    "window": {"id": window_id, "name": window_name},
                    "screenshot": str(screenshot_path),
                }
            )
            return report
    except Exception as exc:
        report["error"] = str(exc)
        raise
    finally:
        report_path.write_text(json.dumps(report, indent=2), encoding="utf-8")
        notepad.terminate()


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--project-dir", type=Path, default=Path.cwd())
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path(tempfile.gettempdir()) / "windows-qa-engineer-smoke",
    )
    args = parser.parse_args()
    report = asyncio.run(run(args.project_dir.resolve(), args.output_dir.resolve()))
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
skill_installer.py — General-purpose installer for agent skills.

Reads install.yaml from a skill directory and executes the declared steps:
platform check, python check, git clone, venv, pip install, MCP config, verify.

Usage:
    python skill_installer.py [--skill-dir PATH] [--project-dir PATH]

Options:
    --skill-dir   Path to the skill root (default: parent of this script)
    --project-dir Path to the project root for .mcp.json (default: cwd)
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any


def load_manifest(skill_dir: Path) -> dict[str, Any]:
    """Load and parse install.yaml from the skill directory."""
    manifest_path = skill_dir / "install.yaml"
    if not manifest_path.exists():
        fail(f"install.yaml not found at {manifest_path}")

    try:
        import yaml
    except ImportError:
        # Fall back to a minimal YAML parser for simple manifests
        return _parse_simple_yaml(manifest_path)

    with open(manifest_path) as f:
        return yaml.safe_load(f)


def _parse_simple_yaml(path: Path) -> dict[str, Any]:
    """Minimal YAML parser for install.yaml without PyYAML dependency.

    Handles the subset of YAML used by install.yaml: scalars, lists,
    nested mappings (one level of indent), block scalars (|).
    """
    import re

    result: dict[str, Any] = {}
    current_key: str | None = None
    current_list: list | None = None
    current_map: dict | None = None
    current_map_key: str | None = None
    block_scalar_key: str | None = None
    block_scalar_lines: list[str] = []
    block_scalar_indent: int = 0
    list_of_maps: bool = False
    list_maps: list[dict] = []
    current_list_map: dict | None = None
    # Track nesting: top-level key that owns a map value
    map_owner: str | None = None

    lines = path.read_text().splitlines()
    i = 0
    while i < len(lines):
        line = lines[i]
        stripped = line.strip()

        # Skip comments and blank lines (unless in block scalar)
        if block_scalar_key:
            indent = len(line) - len(line.lstrip())
            if stripped == "" or indent >= block_scalar_indent:
                block_scalar_lines.append(line[block_scalar_indent:] if indent >= block_scalar_indent else "")
                i += 1
                continue
            else:
                # Block scalar ended
                text = "\n".join(block_scalar_lines).rstrip("\n") + "\n"
                if map_owner and current_map is not None:
                    current_map[block_scalar_key] = text
                else:
                    result[block_scalar_key] = text
                block_scalar_key = None
                block_scalar_lines = []
                # Fall through to process current line

        if stripped == "" or stripped.startswith("#"):
            i += 1
            continue

        indent = len(line) - len(line.lstrip())

        # List item inside a list-of-maps (e.g., repos entries)
        if list_of_maps and indent >= 2 and stripped.startswith("- "):
            # Save previous map
            if current_list_map is not None:
                list_maps.append(current_list_map)
            current_list_map = {}
            # Parse key: value from "- key: value"
            item = stripped[2:].strip()
            if ":" in item:
                k, v = item.split(":", 1)
                current_list_map[k.strip()] = _yaml_val(v.strip())
            i += 1
            continue

        if list_of_maps and current_list_map is not None and indent >= 4 and ":" in stripped:
            k, v = stripped.split(":", 1)
            v = v.strip()
            if v == "|":
                block_scalar_key = k.strip()
                block_scalar_indent = indent + 2
                block_scalar_lines = []
                map_owner = None  # block scalar target is current_list_map handled separately
                # Actually for list maps we need special handling
                # For simplicity, read ahead
                i += 1
                while i < len(lines):
                    bl = lines[i]
                    bi = len(bl) - len(bl.lstrip())
                    if bl.strip() == "" or bi >= block_scalar_indent:
                        block_scalar_lines.append(bl[block_scalar_indent:] if bi >= block_scalar_indent else "")
                        i += 1
                    else:
                        break
                current_list_map[k.strip()] = "\n".join(block_scalar_lines).rstrip("\n") + "\n"
                block_scalar_key = None
                block_scalar_lines = []
                continue
            else:
                current_list_map[k.strip()] = _yaml_val(v)
            i += 1
            continue

        # End list-of-maps context on dedent
        if list_of_maps and indent == 0:
            if current_list_map is not None:
                list_maps.append(current_list_map)
                current_list_map = None
            result[current_key] = list_maps
            list_of_maps = False
            list_maps = []
            current_key = None
            # Fall through

        # End current map context on dedent
        if map_owner and indent == 0:
            result[map_owner] = current_map
            current_map = None
            map_owner = None
            current_key = None

        # End simple list on dedent
        if current_list is not None and indent == 0 and not stripped.startswith("-"):
            result[current_key] = current_list
            current_list = None
            current_key = None

        # Simple list item (e.g., "  - fastmcp")
        if current_list is not None and stripped.startswith("- "):
            current_list.append(_yaml_val(stripped[2:].strip()))
            i += 1
            continue

        # Map sub-key (indent >= 2, inside a map owner)
        if map_owner and current_map is not None and indent >= 2 and ":" in stripped:
            k, v = stripped.split(":", 1)
            v = v.strip()
            if v == "|":
                i += 1
                scalar_lines = []
                scalar_indent = indent + 2
                while i < len(lines):
                    scalar_line = lines[i]
                    scalar_line_indent = len(scalar_line) - len(scalar_line.lstrip())
                    if scalar_line.strip() == "" or scalar_line_indent >= scalar_indent:
                        scalar_lines.append(
                            scalar_line[scalar_indent:]
                            if scalar_line_indent >= scalar_indent
                            else ""
                        )
                        i += 1
                    else:
                        break
                current_map[k.strip()] = "\n".join(scalar_lines).rstrip("\n") + "\n"
                continue
            if v == "" or v.startswith("#"):
                i += 1
                nested_map = {}
                nested_indent = indent + 2
                while i < len(lines):
                    nested_line = lines[i]
                    nested_stripped = nested_line.strip()
                    nested_line_indent = len(nested_line) - len(nested_line.lstrip())
                    if nested_stripped == "" or nested_stripped.startswith("#"):
                        i += 1
                        continue
                    if nested_line_indent < nested_indent:
                        break
                    if nested_line_indent == nested_indent and ":" in nested_stripped:
                        nested_key, nested_value = nested_stripped.split(":", 1)
                        nested_map[nested_key.strip()] = _yaml_val(nested_value.strip())
                    i += 1
                current_map[k.strip()] = nested_map
                continue
            current_map[k.strip()] = _yaml_val(v)
            i += 1
            continue

        # Top-level key
        if indent == 0 and ":" in stripped and not stripped.startswith("-"):
            k, v = stripped.split(":", 1)
            k = k.strip()
            v = v.strip()

            if v == "" or v.startswith("#"):
                # Next lines define a nested structure — peek ahead
                current_key = k
                j = i + 1
                while j < len(lines) and (lines[j].strip() == "" or lines[j].strip().startswith("#")):
                    j += 1
                if j < len(lines):
                    next_stripped = lines[j].strip()
                    if next_stripped.startswith("- url:") or next_stripped.startswith("- url :"):
                        list_of_maps = True
                        list_maps = []
                        current_list_map = None
                    elif next_stripped.startswith("- "):
                        current_list = []
                    else:
                        # Nested map
                        map_owner = k
                        current_map = {}
            elif v == "|":
                block_scalar_key = k
                block_scalar_indent = 2
                block_scalar_lines = []
            else:
                result[k] = _yaml_val(v)
            i += 1
            continue

        i += 1

    # Flush any remaining context
    if block_scalar_key:
        text = "\n".join(block_scalar_lines).rstrip("\n") + "\n"
        result[block_scalar_key] = text
    if list_of_maps:
        if current_list_map is not None:
            list_maps.append(current_list_map)
        result[current_key] = list_maps
    if current_list is not None:
        result[current_key] = current_list
    if map_owner and current_map is not None:
        result[map_owner] = current_map

    return result


def _yaml_val(s: str) -> Any:
    """Convert a YAML scalar string to a Python value."""
    # Strip inline comments
    if "  #" in s:
        s = s[: s.index("  #")].strip()
    if s in ("true", "True"):
        return True
    if s in ("false", "False"):
        return False
    if s in ("null", "None", "~"):
        return None
    # Strip quotes
    if (s.startswith('"') and s.endswith('"')) or (s.startswith("'") and s.endswith("'")):
        return s[1:-1]
    try:
        return int(s)
    except ValueError:
        pass
    try:
        return float(s)
    except ValueError:
        pass
    return s


# ---------------------------------------------------------------------------
# Steps
# ---------------------------------------------------------------------------

def check_platform(manifest: dict) -> dict:
    """Check that current platform matches the manifest requirement."""
    required = manifest.get("platform", "any")
    if required == "any":
        return {"step": "platform", "status": "ok", "detail": platform.platform()}

    current = sys.platform  # win32, darwin, linux
    mapping = {"windows": "win32", "macos": "darwin", "linux": "linux"}
    expected = mapping.get(required, required)

    if current == expected:
        return {"step": "platform", "status": "ok", "detail": platform.platform()}

    return {
        "step": "platform",
        "status": "failed",
        "detail": f"Requires {required} but running on {platform.platform()}",
    }


def check_python(manifest: dict) -> dict:
    """Check that current Python meets version constraint."""
    constraint = manifest.get("python")
    if not constraint:
        return {"step": "python", "status": "ok", "detail": platform.python_version()}

    import re

    for clause in (part.strip() for part in constraint.split(",")):
        match = re.fullmatch(r"([><=!]+)([\d.]+)", clause)
        if not match:
            return {
                "step": "python",
                "status": "failed",
                "detail": f"Unsupported Python constraint: {constraint}",
            }

        op, ver_str = match.groups()
        required = tuple(int(x) for x in ver_str.split("."))
        current = sys.version_info[:len(required)]
        comparisons = {
            ">=": current >= required,
            ">": current > required,
            "<=": current <= required,
            "<": current < required,
            "==": current == required,
            "!=": current != required,
        }
        if op not in comparisons or not comparisons[op]:
            return {
                "step": "python",
                "status": "failed",
                "detail": (
                    f"Requires Python {constraint} but running "
                    f"{platform.python_version()}"
                ),
            }

    return {"step": "python", "status": "ok", "detail": platform.python_version()}


def process_repos(manifest: dict) -> tuple[list[dict], Path | None]:
    """Clone repos, create venvs, install requirements. Returns (steps, venv_python)."""
    repos = manifest.get("repos", [])
    steps = []
    venv_python: Path | None = None

    for repo in repos:
        url = repo["url"]
        target = Path(os.path.expanduser(repo["target"]))

        # Clone
        if target.exists():
            steps.append({"step": "clone", "status": "skipped", "detail": f"{target} already exists"})
        else:
            run(["git", "clone", url, str(target)])
            steps.append({"step": "clone", "status": "ok", "detail": str(target)})

        # Venv
        if repo.get("venv"):
            venv_dir = target / ".venv"
            if venv_dir.exists():
                steps.append({"step": "venv", "status": "skipped", "detail": f"{venv_dir} already exists"})
            else:
                run([sys.executable, "-m", "venv", str(venv_dir)])
                steps.append({"step": "venv", "status": "ok", "detail": str(venv_dir)})

            # Resolve venv python
            if sys.platform == "win32":
                venv_python = venv_dir / "Scripts" / "python.exe"
            else:
                venv_python = venv_dir / "bin" / "python"

        # Requirements
        if repo.get("requirements"):
            req_file = target / repo["requirements"]
            pip_exe = venv_python or Path(sys.executable)
            run([str(pip_exe), "-m", "pip", "install", "-r", str(req_file)])
            steps.append({"step": "deps", "status": "ok", "detail": str(req_file)})

    return steps, venv_python


def install_pip_packages(manifest: dict, venv_python: Path | None) -> dict | None:
    """Install additional pip packages."""
    packages = manifest.get("pip", [])
    if not packages:
        return None

    pip_exe = venv_python or Path(sys.executable)
    run([str(pip_exe), "-m", "pip", "install"] + packages)
    return {"step": "pip", "status": "ok", "detail": ", ".join(packages)}


def configure_mcp(
    manifest: dict,
    skill_dir: Path,
    project_dir: Path,
    venv_python: Path | None,
) -> dict | None:
    """Register the MCP server in .mcp.json."""
    mcp_conf = manifest.get("mcp")
    if not mcp_conf:
        return None

    mcp_json_path = project_dir / ".mcp.json"

    # Read existing
    existing: dict[str, Any] = {}
    if mcp_json_path.exists():
        with open(mcp_json_path) as f:
            existing = json.load(f)

    servers = existing.setdefault("mcpServers", {})

    # Build command — use venv python if available
    python_cmd = str(venv_python) if venv_python else "python"
    script_path = str(skill_dir / mcp_conf["script"])

    # Build env
    env: dict[str, str] = {}

    # PYTHONPATH from repos with pythonpath: true
    pythonpaths = []
    for repo in manifest.get("repos", []):
        if repo.get("pythonpath"):
            pythonpaths.append(str(Path(os.path.expanduser(repo["target"]))))
    if pythonpaths:
        env["PYTHONPATH"] = os.pathsep.join(pythonpaths)

    # Merge env from manifest
    if mcp_conf.get("env"):
        env.update(
            {
                key: os.path.expanduser(str(value))
                for key, value in mcp_conf["env"].items()
            }
        )

    # Build server entry
    entry: dict[str, Any] = {
        "type": "stdio",
        "command": python_cmd,
        "args": [script_path],
    }
    if env:
        entry["env"] = env

    servers[mcp_conf["name"]] = entry

    # Write atomically
    tmp_fd, tmp_path = tempfile.mkstemp(dir=project_dir, suffix=".mcp.json.tmp")
    try:
        with os.fdopen(tmp_fd, "w") as f:
            json.dump(existing, f, indent=2)
            f.write("\n")
        os.replace(tmp_path, mcp_json_path)
    except Exception:
        os.unlink(tmp_path)
        raise

    return {"step": "mcp", "status": "ok", "detail": f"{mcp_json_path} updated"}


def run_verify(manifest: dict, venv_python: Path | None) -> dict | None:
    """Run the verify script."""
    verify = manifest.get("verify")
    if not verify or not verify.get("script"):
        return None

    python_cmd = str(venv_python) if venv_python else sys.executable

    # Build PYTHONPATH
    pythonpaths = []
    for repo in manifest.get("repos", []):
        if repo.get("pythonpath"):
            pythonpaths.append(str(Path(os.path.expanduser(repo["target"]))))

    env = os.environ.copy()
    if pythonpaths:
        existing_pp = env.get("PYTHONPATH", "")
        new_pp = os.pathsep.join(pythonpaths)
        env["PYTHONPATH"] = f"{new_pp}{os.pathsep}{existing_pp}" if existing_pp else new_pp

    result = subprocess.run(
        [python_cmd, "-c", verify["script"]],
        capture_output=True,
        text=True,
        env=env,
        cwd=pythonpaths[0] if pythonpaths else None,
    )

    if result.returncode == 0:
        detail = result.stdout.strip() or "passed"
        return {"step": "verify", "status": "ok", "detail": detail}
    else:
        detail = result.stderr.strip() or result.stdout.strip() or "verification failed"
        return {"step": "verify", "status": "failed", "detail": detail}


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def run(cmd: list[str]) -> subprocess.CompletedProcess:
    """Run a subprocess, raising on failure."""
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        stderr = result.stderr.strip()
        stdout = result.stdout.strip()
        detail = stderr or stdout or f"Command failed with exit code {result.returncode}"
        raise RuntimeError(f"Command failed: {' '.join(cmd)}\n{detail}")
    return result


def fail(message: str) -> None:
    """Print error JSON and exit."""
    report = {"success": False, "error": message, "steps": []}
    print(json.dumps(report, indent=2))
    sys.exit(1)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(description="Install skill dependencies from install.yaml")
    parser.add_argument(
        "--skill-dir",
        type=Path,
        default=Path(__file__).resolve().parent.parent,
        help="Path to the skill root (default: parent of scripts/)",
    )
    parser.add_argument(
        "--project-dir",
        type=Path,
        default=Path.cwd(),
        help="Project root for .mcp.json placement (default: cwd)",
    )
    args = parser.parse_args()

    skill_dir = args.skill_dir.resolve()
    project_dir = args.project_dir.resolve()

    manifest = load_manifest(skill_dir)
    steps: list[dict] = []
    success = True

    # 1. Platform check
    step = check_platform(manifest)
    steps.append(step)
    if step["status"] == "failed":
        success = False
        report = {"success": False, "steps": steps, "mcp_configured": False, "restart_required": False}
        print(json.dumps(report, indent=2))
        sys.exit(1)

    # 2. Python check
    step = check_python(manifest)
    steps.append(step)
    if step["status"] == "failed":
        success = False
        report = {"success": False, "steps": steps, "mcp_configured": False, "restart_required": False}
        print(json.dumps(report, indent=2))
        sys.exit(1)

    # 3. Repos (clone, venv, requirements)
    try:
        repo_steps, venv_python = process_repos(manifest)
        steps.extend(repo_steps)
    except RuntimeError as e:
        steps.append({"step": "repos", "status": "failed", "detail": str(e)})
        success = False
        report = {"success": False, "steps": steps, "mcp_configured": False, "restart_required": False}
        print(json.dumps(report, indent=2))
        sys.exit(1)

    # 4. Additional pip packages
    try:
        pip_step = install_pip_packages(manifest, venv_python)
        if pip_step:
            steps.append(pip_step)
    except RuntimeError as e:
        steps.append({"step": "pip", "status": "failed", "detail": str(e)})
        success = False
        report = {"success": False, "steps": steps, "mcp_configured": False, "restart_required": False}
        print(json.dumps(report, indent=2))
        sys.exit(1)

    # 5. MCP config
    mcp_configured = False
    try:
        mcp_step = configure_mcp(manifest, skill_dir, project_dir, venv_python)
        if mcp_step:
            steps.append(mcp_step)
            mcp_configured = True
    except Exception as e:
        steps.append({"step": "mcp", "status": "failed", "detail": str(e)})
        success = False

    # 6. Verify
    if success:
        verify_step = run_verify(manifest, venv_python)
        if verify_step:
            steps.append(verify_step)
            if verify_step["status"] == "failed":
                success = False

    # 7. Report
    report = {
        "success": success,
        "steps": steps,
        "mcp_configured": mcp_configured,
        "restart_required": mcp_configured and success,
    }
    print(json.dumps(report, indent=2))

    if not success:
        sys.exit(1)


if __name__ == "__main__":
    main()

"""
ufo_windows_qa_mcp_server.py

Stdio MCP server that exposes UFO's real Windows automation tools to an MCP client.

Composes UFO's UICollector, HostUIExecutor, AppUIExecutor into ONE server
via FastMCP.mount(). No mocks, no re-implementation.

Requires: UFO installed (pip install from repo), fastmcp, pydantic
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import sys
import time
from typing import Annotated, Any, Dict, List, Optional

ufo_root = os.environ.get("UFO_ROOT")
if ufo_root:
    os.chdir(ufo_root)
    if ufo_root not in sys.path:
        sys.path.insert(0, ufo_root)

from fastmcp import FastMCP
from pydantic import Field

from ufo.client.mcp.mcp_registry import MCPRegistry
from ufo.client.mcp.local_servers import load_all_servers

logger = logging.getLogger(__name__)


def _get_ufo_server(namespace: str) -> FastMCP:
    """Load all UFO servers, then fetch by namespace."""
    load_all_servers()
    if not MCPRegistry.is_registered(namespace):
        raise RuntimeError(
            f"UFO MCP server '{namespace}' not registered. "
            f"Available: {MCPRegistry.list()}"
        )
    return MCPRegistry.get(namespace)


# Compose into one server
mcp = FastMCP("UFO Windows QA (UIA/Win32)")

mcp.mount(_get_ufo_server("UICollector"))
mcp.mount(_get_ufo_server("HostUIExecutor"))
mcp.mount(_get_ufo_server("AppUIExecutor"))


# QA helper tools (thin wrappers around UFO tools)

def _parse_tool_result(result: Any) -> Any:
    """Return structured FastMCP results when possible."""
    def to_plain(value: Any) -> Any:
        root = getattr(value, "root", None)
        if root is not None:
            return to_plain(root)
        if hasattr(value, "model_dump"):
            return to_plain(value.model_dump())
        if isinstance(value, dict):
            return {str(key): to_plain(item) for key, item in value.items()}
        if isinstance(value, (list, tuple)):
            return [to_plain(item) for item in value]
        return value

    structured = getattr(result, "structured_content", None)
    if structured is not None:
        if isinstance(structured, dict) and set(structured.keys()) == {"result"}:
            return to_plain(structured["result"])
        return to_plain(structured)

    content = getattr(result, "content", None)
    if content:
        text = getattr(content[0], "text", None)
        if text is not None:
            try:
                return to_plain(json.loads(text))
            except json.JSONDecodeError:
                return text

    return to_plain(result)


async def _call_mounted_tool(name: str, arguments: Dict[str, Any]) -> Any:
    """Call one of the mounted UFO tools through FastMCP's public Tool API."""
    tool = await mcp.get_tool(name)
    return await tool.run(arguments)


@mcp.tool()
async def qa_refresh_and_list_windows(
    remove_empty: Annotated[bool, Field(description="Drop empty/ghost windows.")] = True
) -> Annotated[List[Dict[str, Any]], Field(description="Window list.")]:
    """Refresh + list windows in one call. Wraps UICollector.get_desktop_app_info."""
    result = await _call_mounted_tool(
        "get_desktop_app_info",
        {"remove_empty": remove_empty, "refresh_app_windows": True},
    )
    parsed = _parse_tool_result(result)
    return parsed if isinstance(parsed, list) else []


@mcp.tool()
async def qa_refresh_controls(
    field_list: Annotated[List[str], Field(description="Fields to fetch per control.")],
) -> Annotated[List[Dict[str, Any]], Field(description="Controls for selected window.")]:
    """Refresh control map for the selected window. Wraps UICollector.get_app_window_controls_info."""
    try:
        result = await _call_mounted_tool(
            "get_app_window_controls_info", {"field_list": field_list}
        )
        parsed = _parse_tool_result(result)
        return parsed if isinstance(parsed, list) else []
    except Exception as exc:
        logger.warning("qa_refresh_controls failed: %s", exc)
        return [{"error": str(exc), "source": "ufo_qa"}]


@mcp.tool()
async def qa_wait_for_text_contains(
    id: Annotated[str, Field(description="Control id.")],
    name: Annotated[str, Field(description="Control name.")],
    expected_substring: Annotated[str, Field(description="Substring that must appear.")],
    timeout_s: Annotated[float, Field(description="Max wait seconds.")] = 10.0,
    poll_s: Annotated[float, Field(description="Poll interval seconds.")] = 0.5,
) -> Annotated[Dict[str, Any], Field(description="Result with ok flag and observed text.")]:
    """Poll texts(id,name) until expected_substring appears or timeout. Avoids arbitrary sleeps."""
    deadline = time.time() + max(0.1, timeout_s)
    last_text: Optional[str] = None

    while time.time() < deadline:
        raw = await _call_mounted_tool("texts", {"id": id, "name": name})
        res = _parse_tool_result(raw)
        last_text = res if isinstance(res, str) else str(res)
        if expected_substring in last_text:
            return {"ok": True, "text": last_text, "matched": expected_substring}
        await asyncio.sleep(max(0.05, poll_s))

    return {
        "ok": False,
        "text": last_text,
        "matched": expected_substring,
        "timeout_s": timeout_s,
    }


def main() -> None:
    mcp.run()


if __name__ == "__main__":
    main()