from __future__ import annotations

import json
import os
import re
import shutil
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
import xml.etree.ElementTree as ET
from pathlib import Path

from .envfile import load_env_file
from .maven import find_maven_command
from .versioning import (
    infer_ghidra_install_meta,
    infer_ghidra_version_from_path,
    read_pom_versions,
)


REQUIRED_GHIDRA_JARS: tuple[tuple[str, str], ...] = (
    ("Base", "Ghidra/Features/Base/lib/Base.jar"),
    ("Decompiler", "Ghidra/Features/Decompiler/lib/Decompiler.jar"),
    ("Docking", "Ghidra/Framework/Docking/lib/Docking.jar"),
    ("Generic", "Ghidra/Framework/Generic/lib/Generic.jar"),
    ("Project", "Ghidra/Framework/Project/lib/Project.jar"),
    ("SoftwareModeling", "Ghidra/Framework/SoftwareModeling/lib/SoftwareModeling.jar"),
    ("Utility", "Ghidra/Framework/Utility/lib/Utility.jar"),
    ("Gui", "Ghidra/Framework/Gui/lib/Gui.jar"),
    ("FileSystem", "Ghidra/Framework/FileSystem/lib/FileSystem.jar"),
    ("Graph", "Ghidra/Framework/Graph/lib/Graph.jar"),
    ("DB", "Ghidra/Framework/DB/lib/DB.jar"),
    ("Emulation", "Ghidra/Framework/Emulation/lib/Emulation.jar"),
    ("PDB", "Ghidra/Features/PDB/lib/PDB.jar"),
    ("FunctionID", "Ghidra/Features/FunctionID/lib/FunctionID.jar"),
    ("Help", "Ghidra/Framework/Help/lib/Help.jar"),
    ("Debugger-api", "Ghidra/Debug/Debugger-api/lib/Debugger-api.jar"),
    (
        "Framework-TraceModeling",
        "Ghidra/Debug/Framework-TraceModeling/lib/Framework-TraceModeling.jar",
    ),
    (
        "Debugger-rmi-trace",
        "Ghidra/Debug/Debugger-rmi-trace/lib/Debugger-rmi-trace.jar",
    ),
)

PLUGIN_CLASS = "com.xebyte.GhidraMCPPlugin"
PLUGIN_EXTENSION_NAME = "GhidraMCP"
DEFAULT_MCP_URL = "http://127.0.0.1:8089"
DEFAULT_MCP_WAIT_SECONDS = 120
DEFAULT_GHIDRA_EXIT_WAIT_SECONDS = 15
DEFAULT_BENCHMARK_DLL = Path("fun-doc") / "benchmark" / "build" / "Benchmark.dll"
DEFAULT_BENCHMARK_DEBUG_EXE = Path("fun-doc") / "benchmark" / "build" / "BenchmarkDebug.exe"
LEGACY_BENCHMARK_PROGRAM = "/benchmark/Benchmark.dll"
DEFAULT_BENCHMARK_FOLDER = "/testing/benchmark"
DEFAULT_BENCHMARK_PROGRAM = f"{DEFAULT_BENCHMARK_FOLDER}/Benchmark.dll"
DEFAULT_BENCHMARK_DEBUG_PROGRAM = f"{DEFAULT_BENCHMARK_FOLDER}/BenchmarkDebug.exe"
DEFAULT_BENCHMARK_FUNCTION = "calc_crc16"
# Max seconds to wait for post-import benchmark analysis to settle before the
# YAML regression asserts. Sized for a cold install (freshly-created Ghidra
# user-config dir), where the first analysis pass runs well past a minute.
BENCHMARK_ANALYSIS_TIMEOUT_S = 240
BENCHMARK_DEPLOY_TEST_MODES = {
    "benchmark-read",
    "benchmark-write",
    "release",
    "debugger-live",
    "multi-program",
}
SMOKE_REQUIRED_TOOLS = {
    "decompile_function",
    "get_function_variables",
    "analyze_function_completeness",
    "batch_set_comments",
    "set_local_variable_type",
    "rename_variables",
    "prompt_policy",
    "save_program",
    "save_all_programs",
    "set_function_prototype",
    "rename_function_by_address",
    "search_data_types",
    "create_struct",
    "get_struct_layout",
    "list_open_programs",
    "debugger/launch",
}
RELEASE_CONTRACT_TOOLS = SMOKE_REQUIRED_TOOLS | {
    "analysis_status",
    "create_folder",
    "delete_file",
    "import_file",
    "list_project_files",
    "list_functions",
    "search_functions",
    "get_address_spaces",
    "list_imports",
    "list_exports",
    "list_strings",
    "debugger/launch",
    "debugger/status",
    "debugger/modules",
}


def ghidra_user_base_dir() -> Path:
    if sys.platform == "darwin":
        return Path.home() / "Library" / "ghidra"
    if os.name == "nt":
        appdata = os.environ.get("APPDATA")
        if appdata:
            return Path(appdata) / "ghidra"
        return Path.home() / "AppData" / "Roaming" / "ghidra"

    xdg_config_home = os.environ.get("XDG_CONFIG_HOME")
    if xdg_config_home:
        return Path(xdg_config_home) / "ghidra"
    return Path.home() / ".config" / "ghidra"


def _version_sort_key(name: str) -> tuple[int, int, int, int]:
    """Sort key for Ghidra user-config dir names.

    Returns ``(major, minor, patch, explicit_patch)``. The trailing
    flag is 1 when the dir name carried an explicit patch component
    (e.g. ``ghidra_12.1.0_PUBLIC``) and 0 when it didn't
    (``ghidra_12.1_PUBLIC``), so a dir with an explicit patch beats
    an otherwise-equal shorter dir name. Without this tiebreaker the
    sort was non-deterministic across filesystems: Windows' alpha
    glob order put ``ghidra_12.1.0_PUBLIC`` before ``ghidra_12.1_PUBLIC``
    and the test passed; Linux's creation-order glob produced the
    opposite outcome and CI failed.
    """
    match = re.search(r"ghidra_(\d+)\.(\d+)(?:\.(\d+))?", name)
    if not match:
        return (0, 0, 0, 0)
    explicit_patch = 1 if match.group(3) is not None else 0
    return (
        int(match.group(1)),
        int(match.group(2)),
        int(match.group(3) or 0),
        explicit_patch,
    )


def resolve_ghidra_user_dir(
    ghidra_path: Path, user_base_dir: Path | None = None
) -> Path:
    """Resolve the user-config dir matching a Ghidra install.

    Ghidra writes its per-user state under
    ``%APPDATA%\\ghidra\\ghidra_<version>_<layout>\\``. The dir is
    created lazily on first launch, so for a freshly-installed Ghidra
    it may not exist yet. We therefore prefer to *construct* the
    expected dir name from the install path rather than enumerating
    existing siblings — see #217, where a v5.10→v5.11 deploy targeting
    a freshly-installed ``F:\\ghidra_12.1_PUBLIC`` quietly resolved to
    a leftover ``ghidra_12.1_DEV`` user dir and installed the
    extension where the running Ghidra never looked for it.
    """
    user_base_dir = user_base_dir or ghidra_user_base_dir()
    target_version, target_layout = infer_ghidra_install_meta(ghidra_path)

    # When both version and layout are recoverable from the install
    # path, return the explicit dir unconditionally. The dir does not
    # need to already exist — Ghidra will create it on first launch.
    if target_version and target_layout:
        return user_base_dir / f"ghidra_{target_version}_{target_layout}"

    # Version known but layout couldn't be inferred (e.g. a custom
    # install path with application.properties present). Prefer an
    # existing matching dir, then PUBLIC, then a constructed PUBLIC
    # default.
    if target_version:
        if user_base_dir.is_dir():
            matching_dirs = sorted(
                path
                for path in user_base_dir.glob(f"ghidra_{target_version}*")
                if path.is_dir() and "_location_" not in path.name
            )
            if matching_dirs:
                public_dir = next(
                    (path for path in matching_dirs if "PUBLIC" in path.name), None
                )
                return public_dir or matching_dirs[0]
        return user_base_dir / f"ghidra_{target_version}_PUBLIC"

    # No version metadata at all — last-resort fallback to the
    # newest-looking existing dir so a totally custom install still
    # gets *some* answer instead of an exception.
    if user_base_dir.is_dir():
        version_dirs = sorted(
            (path for path in user_base_dir.glob("ghidra_*") if path.is_dir()),
            key=lambda path: _version_sort_key(path.name),
            reverse=True,
        )
        if version_dirs:
            return version_dirs[0]

    return user_base_dir / "ghidra_unknown_PUBLIC"


def patch_frontend_tool_config(content: str) -> tuple[str, bool]:
    original = content
    updated = content

    for package_name in ("Developer", "GhidraMCP"):
        updated = re.sub(
            rf"\s*<PACKAGE NAME=\"{re.escape(package_name)}\"\s*/>\s*",
            "\n",
            updated,
        )
        updated = re.sub(
            rf"(?s)\s*<PACKAGE NAME=\"{re.escape(package_name)}\">\s*.*?</PACKAGE>\s*",
            "\n",
            updated,
        )

    if PLUGIN_CLASS in updated:
        updated = mark_extension_known_in_tool_config(updated, PLUGIN_EXTENSION_NAME)
        return updated, updated != original

    utility_self_closing = '<PACKAGE NAME="Utility" />'
    if utility_self_closing in updated:
        replacement = (
            '<PACKAGE NAME="Utility">\n'
            f'                <INCLUDE CLASS="{PLUGIN_CLASS}" />\n'
            "            </PACKAGE>"
        )
        updated = updated.replace(utility_self_closing, replacement, 1)
        updated = mark_extension_known_in_tool_config(updated, PLUGIN_EXTENSION_NAME)
        return updated, True

    utility_block = '<PACKAGE NAME="Utility">'
    if utility_block in updated:
        replacement = (
            '<PACKAGE NAME="Utility">\n'
            f'                <INCLUDE CLASS="{PLUGIN_CLASS}" />'
        )
        updated = updated.replace(utility_block, replacement, 1)
        updated = mark_extension_known_in_tool_config(updated, PLUGIN_EXTENSION_NAME)
        return updated, True

    root_node = "<ROOT_NODE"
    if root_node in updated:
        insertion = (
            '<PACKAGE NAME="Utility">\n'
            f'                <INCLUDE CLASS="{PLUGIN_CLASS}" />\n'
            "            </PACKAGE>\n"
            "<ROOT_NODE"
        )
        updated = updated.replace(root_node, insertion, 1)
        updated = mark_extension_known_in_tool_config(updated, PLUGIN_EXTENSION_NAME)
        return updated, True

    updated = mark_extension_known_in_tool_config(updated, PLUGIN_EXTENSION_NAME)
    return updated, updated != original


def mark_extension_known_in_tool_config(content: str, extension_name: str) -> str:
    """Record an installed extension as known to suppress Ghidra's first-run plugin dialog."""
    if re.search(
        rf'<EXTENSION\s+(?:[^>]*\s)?NAME="{re.escape(extension_name)}"',
        content,
    ):
        return content

    extension_entry = f'            <EXTENSION NAME="{extension_name}" />\n'
    empty_extensions = re.compile(r"(?m)^([ \t]*)<EXTENSIONS\s*/>\s*$")
    if empty_extensions.search(content):
        return empty_extensions.sub(
            rf"\1<EXTENSIONS>\n{extension_entry}\1</EXTENSIONS>",
            content,
            count=1,
        )

    extensions_open = re.compile(r"(?m)^([ \t]*)<EXTENSIONS>\s*$")
    match = extensions_open.search(content)
    if match:
        insert_at = match.end()
        return content[:insert_at] + "\n" + extension_entry + content[insert_at:]

    if "</TOOL>" not in content:
        return content
    return content.replace(
        "</TOOL>",
        f"        <EXTENSIONS>\n{extension_entry}        </EXTENSIONS>\n    </TOOL>",
        1,
    )


def patch_tool_tcd(content: str) -> tuple[str, bool]:
    original = content
    updated = re.sub(
        rf'\s*<PACKAGE NAME="GhidraMCP">\s*<INCLUDE CLASS="{re.escape(PLUGIN_CLASS)}"\s*/>\s*</PACKAGE>',
        "",
        content,
    )
    updated = mark_extension_known_in_tool_config(updated, PLUGIN_EXTENSION_NAME)
    return updated, updated != original


def patch_codebrowser_tcd(content: str) -> tuple[str, bool]:
    return patch_tool_tcd(content)


def _write_text_file(path: Path, content: str) -> None:
    path.write_text(content, encoding="utf-8", newline="")


def patch_ghidra_user_configs(
    user_base_dir: Path,
    target_user_dir: Path | None = None,
    *,
    dry_run: bool = False,
) -> None:
    """Patch FrontEndTool.xml + tool tcd files under the Ghidra user dir.

    When ``target_user_dir`` is provided, only files inside that directory
    are patched. This is the recommended call shape from
    :func:`deploy_to_ghidra` — a Ghidra 12.1 deploy must NOT touch the
    user-config dirs left over from older Ghidra installs (12.0.4,
    11.4.2, …), because those dirs reference extensions from those older
    Ghidras. Stamping the new plugin's INCLUDE into a sibling version's
    FrontEndTool.xml is exactly the #217 bug: the deploy log this morning
    showed ``Patched FrontEnd config …/ghidra_12.0.4_PUBLIC/…`` even
    though we were targeting 12.1.

    When ``target_user_dir`` is None, falls back to globbing every
    version subdirectory under ``user_base_dir``. Kept for backward
    compatibility (existing tests pass without changes); production
    deploys should always supply ``target_user_dir``.
    """
    if not user_base_dir.is_dir():
        return

    if target_user_dir is not None:
        # #217 fix: restrict the glob to a single subdirectory.
        if not target_user_dir.is_dir():
            return
        front_end_files = sorted(target_user_dir.glob("FrontEndTool.xml"))
        tcd_files = sorted(target_user_dir.glob("tools/*.tcd"))
    else:
        front_end_files = sorted(user_base_dir.glob("*/FrontEndTool.xml"))
        tcd_files = sorted(user_base_dir.glob("*/tools/*.tcd"))

    for front_end_file in front_end_files:
        updated, modified = patch_frontend_tool_config(
            front_end_file.read_text(encoding="utf-8")
        )
        if not modified:
            continue
        if dry_run:
            print(f"DRY RUN: patch {front_end_file}")
            continue
        _write_text_file(front_end_file, updated)
        print(f"Patched FrontEnd config {front_end_file}")

    for tcd_file in tcd_files:
        updated, modified = patch_tool_tcd(tcd_file.read_text(encoding="utf-8"))
        if not modified:
            continue
        if dry_run:
            print(f"DRY RUN: patch {tcd_file}")
            continue
        _write_text_file(tcd_file, updated)
        print(f"Patched tool config {tcd_file}")


def _find_plugin_jar(repo_root: Path) -> Path | None:
    target_dir = repo_root / "target"
    version = read_pom_versions(repo_root).project_version
    candidates = [
        target_dir / "GhidraMCP.jar",
        target_dir / f"GhidraMCP-{version}.jar",
    ]
    for candidate in candidates:
        if candidate.is_file():
            return candidate

    jars = sorted(
        target_dir.glob("GhidraMCP*.jar"),
        key=lambda path: path.stat().st_mtime,
        reverse=True,
    )
    return jars[0] if jars else None


def install_user_extension(
    repo_root: Path, ghidra_path: Path, archive_path: Path, *, dry_run: bool = False
) -> Path:
    user_base_dir = ghidra_user_base_dir()
    user_version_dir = resolve_ghidra_user_dir(ghidra_path, user_base_dir)
    user_extensions_base = user_version_dir / "Extensions"
    user_extension_dir = user_extensions_base / "GhidraMCP"
    user_lib_dir = user_extension_dir / "lib"

    if dry_run:
        print(f"DRY RUN: ensure directory {user_extensions_base}")
        print(f"DRY RUN: remove stale jars matching {user_lib_dir / 'GhidraMCP*.jar'}")
        print(f"DRY RUN: extract {archive_path} -> {user_extensions_base}")
        return user_extension_dir

    user_extensions_base.mkdir(parents=True, exist_ok=True)
    user_lib_dir.mkdir(parents=True, exist_ok=True)
    for stale_jar in user_lib_dir.glob("GhidraMCP*.jar"):
        for attempt in range(10):
            try:
                stale_jar.unlink(missing_ok=True)
                break
            except PermissionError:
                if attempt == 9:
                    raise
                time.sleep(1)
        print(f"Removed stale plugin jar {stale_jar}")

    try:
        with zipfile.ZipFile(archive_path) as archive:
            archive.extractall(user_extensions_base)
        print(f"Installed user extension to {user_extension_dir}")
        return user_extension_dir
    except Exception as exc:
        plugin_jar = _find_plugin_jar(repo_root)
        if plugin_jar is None:
            raise RuntimeError(
                "Extension extraction failed and no fallback plugin jar was found"
            ) from exc

        fallback_destination = user_lib_dir / "GhidraMCP.jar"
        shutil.copy2(plugin_jar, fallback_destination)
        print(f"Fell back to jar-only install at {fallback_destination}")
        return user_extension_dir


def find_ghidra_executable(ghidra_path: Path) -> Path:
    # Ghidra release zips ship BOTH ghidraRun (shell script) and
    # ghidraRun.bat (Windows batch) regardless of host OS, so picking the
    # right one requires a platform check rather than first-match-found.
    # On Linux/macOS, returning ghidraRun.bat made subprocess.Popen try to
    # exec cmd.exe and fail with FileNotFoundError. See #191.
    if sys.platform == "win32":
        candidates = [
            ghidra_path / "ghidraRun.bat",
            ghidra_path / "ghidraRun",
            ghidra_path / "ghidra",
        ]
    else:
        candidates = [
            ghidra_path / "ghidraRun",
            ghidra_path / "ghidra",
            ghidra_path / "ghidraRun.bat",
        ]
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    raise FileNotFoundError(f"Unable to find Ghidra launcher under {ghidra_path}")


def find_plugin_archive(repo_root: Path) -> Path:
    version = read_pom_versions(repo_root).project_version
    # Check Gradle output first, then Maven target/ for backward compatibility during transition.
    candidates = [
        repo_root / "build" / "distributions" / f"GhidraMCP-{version}.zip",
        repo_root / "target" / f"GhidraMCP-{version}.zip",
        repo_root / "target" / "GhidraMCP.zip",
    ]
    for candidate in candidates:
        if candidate.is_file():
            return candidate

    for search_dir in [repo_root / "build" / "distributions", repo_root / "target"]:
        archives = sorted(
            search_dir.glob("GhidraMCP*.zip"),
            key=lambda path: path.stat().st_mtime,
            reverse=True,
        )
        if archives:
            return archives[0]

    raise FileNotFoundError(
        "No GhidraMCP plugin archive found in build/distributions/ or target/"
    )


def print_command(command: list[str]) -> None:
    print(" ".join(command))


def resolve_mcp_url(repo_root: Path) -> str:
    env_values = load_env_file(repo_root / ".env")
    if env_values.get("GHIDRA_MCP_URL"):
        return env_values["GHIDRA_MCP_URL"].rstrip("/")
    port = env_values.get("GHIDRA_MCP_PORT", "8089").strip() or "8089"
    bind = env_values.get("GHIDRA_MCP_BIND_ADDRESS", "127.0.0.1").strip()
    if not bind or bind in {"0.0.0.0", "::"}:
        bind = "127.0.0.1"
    return f"http://{bind}:{port}".rstrip("/")


def resolve_deploy_test_modes(repo_root: Path, cli_modes: list[str] | None) -> list[str]:
    modes = list(cli_modes or [])
    env_values = load_env_file(repo_root / ".env")
    raw_modes = env_values.get("GHIDRA_MCP_DEPLOY_TESTS", "").strip()
    if raw_modes and raw_modes.lower() not in {"0", "false", "no", "none", "off"}:
        modes.extend(
            mode.strip()
            for mode in re.split(r"[,;\s]+", raw_modes)
            if mode.strip()
        )
    return list(dict.fromkeys(modes))


def _mcp_headers(repo_root: Path) -> dict[str, str]:
    env_values = load_env_file(repo_root / ".env")
    token = env_values.get("GHIDRA_MCP_AUTH_TOKEN", "").strip()
    return {"Authorization": f"Bearer {token}"} if token else {}


def _mcp_request(
    repo_root: Path,
    mcp_url: str,
    path: str,
    *,
    method: str = "GET",
    data: dict | None = None,
    params: dict | None = None,
    timeout: int = 10,
) -> tuple[int, object]:
    body = None
    headers = _mcp_headers(repo_root)
    if data is not None:
        body = json.dumps(data).encode("utf-8")
        headers["Content-Type"] = "application/json"
    url = f"{mcp_url}{path}"
    if params:
        url = f"{url}?{urllib.parse.urlencode(params)}"
    request = urllib.request.Request(url, data=body, headers=headers, method=method)
    with urllib.request.urlopen(request, timeout=timeout) as response:
        text = response.read().decode("utf-8", errors="replace")
        try:
            parsed: object = json.loads(text)
        except ValueError:
            parsed = text
        return response.status, parsed


def _ensure_mcp_ok(path: str, payload: object) -> None:
    if isinstance(payload, dict) and payload.get("error"):
        raise RuntimeError(f"{path} failed: {payload['error']}")
    if isinstance(payload, str) and payload.lower().startswith("failed"):
        raise RuntimeError(f"{path} failed: {payload}")


def _mcp_error_message(payload: object) -> str:
    if isinstance(payload, dict):
        error = payload.get("error")
        if error is not None:
            return str(error)
    if isinstance(payload, str):
        return payload
    return ""


def _expect_mcp_error(path: str, payload: object, required_terms: tuple[str, ...]) -> None:
    message = _mcp_error_message(payload)
    if not message:
        raise RuntimeError(f"{path} was expected to fail but returned: {payload}")
    lowered = message.lower()
    missing = [term for term in required_terms if term.lower() not in lowered]
    if missing:
        raise RuntimeError(
            f"{path} error was not actionable enough; missing {missing}. Error: {message}"
        )


def _enumerate_ghidra_processes() -> list[dict[str, object]]:
    """Return every running Ghidra process on this machine, install-agnostic.

    Each entry is {pid, name, command}. The earlier
    _find_matching_ghidra_processes filtered by install path on the same
    pass, which silently missed Ghidras running from a *different*
    install during a version-changing deploy — see the v5.10→v5.11
    Ghidra-12.1 deploy where an old 12.0.4 was still up but went
    undetected. This helper does the cross-platform process scan once;
    callers filter by path themselves.
    """
    if os.name == "nt":
        command = [
            "powershell",
            "-NoProfile",
            "-Command",
            (
                "Get-CimInstance Win32_Process | "
                "Where-Object { $_.Name -match '^(javaw?|ghidra).*' } | "
                "Select-Object ProcessId,Name,ExecutablePath,CommandLine | "
                "ConvertTo-Json -Compress"
            ),
        ]
        completed = subprocess.run(command, capture_output=True, text=True, check=False)
        if completed.returncode != 0 or not completed.stdout.strip():
            return []
        raw = json.loads(completed.stdout)
        rows = raw if isinstance(raw, list) else [raw]
        out: list[dict[str, object]] = []
        for row in rows:
            cmd = str(row.get("CommandLine") or "")
            name = str(row.get("Name") or "").lower()
            cmd_lower = cmd.lower()
            is_ghidra = (
                name in {"java.exe", "javaw.exe", "ghidrarun.bat", "ghidrarun"}
                and ("ghidra.ghidra" in cmd_lower or "ghidrarun" in cmd_lower)
            )
            if is_ghidra:
                out.append(
                    {
                        "pid": int(row["ProcessId"]),
                        "name": row.get("Name", ""),
                        "command": cmd,
                    }
                )
        return out
    ps = subprocess.run(["ps", "-eo", "pid=,args="], capture_output=True, text=True, check=False)
    out = []
    for line in ps.stdout.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        pid_text, _, command = stripped.partition(" ")
        command_lower = command.lower()
        if "ghidra.ghidra" in command_lower or "ghidrarun" in command_lower:
            out.append({"pid": int(pid_text), "name": "process", "command": command})
    return out


def _find_matching_ghidra_processes(ghidra_path: Path) -> list[dict[str, object]]:
    """Ghidra processes whose command-line includes ``ghidra_path``.

    Used by the deploy flow to identify the install we're targeting so
    it can be gracefully shut down before extension replacement. For
    processes that match *other* Ghidra installs, see
    ``_find_mismatched_ghidra_processes`` — those would be warned about
    rather than auto-shut-down, because they may belong to unrelated
    work the operator hasn't agreed to close.
    """
    target = str(ghidra_path.resolve()).lower()
    return [
        proc for proc in _enumerate_ghidra_processes()
        if target in str(proc["command"]).lower()
    ]


def _find_mismatched_ghidra_processes(ghidra_path: Path) -> list[dict[str, object]]:
    """Ghidra processes from a DIFFERENT install than ``ghidra_path``.

    Surfaced as a warning at deploy time so a version-mixing scenario
    is visible: an old Ghidra still bound to MCP port 8089 will respond
    to the deploy's post-start smoke checks instead of the just-deployed
    new version, producing confusing "wrong version" failures.
    """
    target = str(ghidra_path.resolve()).lower()
    return [
        proc for proc in _enumerate_ghidra_processes()
        if target not in str(proc["command"]).lower()
    ]


def _terminate_process(pid: int) -> None:
    if os.name == "nt":
        subprocess.run(["taskkill", "/PID", str(pid), "/F"], check=False)
    else:
        os.kill(pid, signal.SIGKILL)


def _terminate_processes_by_name(process_name: str) -> None:
    if os.name == "nt":
        subprocess.run(["taskkill", "/IM", process_name, "/F"], check=False)
        return
    subprocess.run(["pkill", "-f", process_name], check=False)


def _project_state_path_from_gpr(project_path: str) -> Path | None:
    if not project_path:
        return None
    gpr = Path(project_path)
    if gpr.suffix.lower() != ".gpr":
        return None
    return gpr.with_suffix(".rep") / "projectState"


def _deploy_tests_use_benchmark(test_modes: list[str]) -> bool:
    return any(mode in BENCHMARK_DEPLOY_TEST_MODES for mode in test_modes)


def clear_restored_benchmark_tools(repo_root: Path, *, dry_run: bool = False) -> int:
    env_values = load_env_file(repo_root / ".env")
    project_state = _project_state_path_from_gpr(env_values.get("GHIDRA_PROJECT_PATH", "").strip())
    if project_state is None or not project_state.is_file():
        return 0

    try:
        tree = ET.parse(project_state)
    except ET.ParseError as exc:
        print(f"WARNING: Could not parse Ghidra project state {project_state}: {exc}")
        return 0

    root = tree.getroot()
    parent_by_child = {child: parent for parent in root.iter() for child in parent}
    removed = 0
    benchmark_state_markers = (
        f'VALUE="{DEFAULT_BENCHMARK_PROGRAM}"',
        f'VALUE="diablo2:{DEFAULT_BENCHMARK_PROGRAM}"',
        f'VALUE="{DEFAULT_BENCHMARK_DEBUG_PROGRAM}"',
        f'VALUE="diablo2:{DEFAULT_BENCHMARK_DEBUG_PROGRAM}"',
        f'VALUE="{LEGACY_BENCHMARK_PROGRAM}"',
        "/testing/benchmark/",
        "/New Traces/pydbg/BenchmarkDebug.exe",
    )
    for tool in list(root.iter("RUNNING_TOOL")):
        if tool.attrib.get("TOOL_NAME") not in {"CodeBrowser", "Debugger"}:
            continue
        tool_xml = ET.tostring(tool, encoding="unicode")
        if not any(marker in tool_xml for marker in benchmark_state_markers):
            continue
        parent = parent_by_child.get(tool)
        if parent is None:
            continue
        if dry_run:
            removed += 1
            continue
        parent.remove(tool)
        removed += 1

    if removed == 0:
        return 0
    if dry_run:
        print(f"DRY RUN: remove {removed} restored benchmark CodeBrowser tool(s) from {project_state}")
        return removed

    backup_path = project_state.with_name(project_state.name + ".GhidraMCP.bak")
    shutil.copy2(project_state, backup_path)
    tree.write(project_state, encoding="utf-8", xml_declaration=True)
    print(f"Removed {removed} restored benchmark CodeBrowser tool(s) from {project_state}")
    print(f"Backed up previous project state to {backup_path}")
    return removed


def close_running_ghidra_for_deploy(
    repo_root: Path,
    ghidra_path: Path,
    *,
    mcp_url: str,
    dry_run: bool = False,
    wait_seconds: int = DEFAULT_GHIDRA_EXIT_WAIT_SECONDS,
) -> bool:
    # Warn about Ghidras running from a DIFFERENT install. We don't
    # touch them (they may belong to unrelated work the operator hasn't
    # agreed to close), but surfacing them keeps a version-mixing
    # scenario from going undetected: an old Ghidra still bound to MCP
    # port 8089 will respond to the deploy's post-start smoke checks
    # instead of the just-deployed new version, producing confusing
    # "wrong version" failures. This was the v5.10→v5.11 deploy gap
    # the user flagged after the Ghidra 12.0.4 → 12.1 cutover.
    mismatched = _find_mismatched_ghidra_processes(ghidra_path)
    if mismatched:
        print(
            f"WARNING: {len(mismatched)} Ghidra process(es) running from a "
            f"DIFFERENT install than deploy target {ghidra_path}:"
        )
        for proc in mismatched:
            print(f"  PID {proc['pid']}: {proc['command']}")
        print(
            "  These may bind MCP port 8089 and intercept the post-deploy "
            "smoke checks intended for the new install. If the deploy's "
            "version probe reports the wrong version, close the other "
            "Ghidra(s) (save work first) and re-run."
        )

    matches = _find_matching_ghidra_processes(ghidra_path)
    if not matches:
        print("No matching running Ghidra process detected.")
        return False
    for proc in matches:
        print(f"Detected running Ghidra PID {proc['pid']}: {proc['command']}")
    if dry_run:
        print(f"DRY RUN: save all open programs via {mcp_url}/save_all_programs")
        print(f"DRY RUN: graceful exit via {mcp_url}/exit_ghidra")
        for proc in matches:
            print(f"DRY RUN: force-kill PID {proc['pid']} if still running")
        return True

    try:
        _mcp_request(repo_root, mcp_url, "/save_all_programs", timeout=60)
        print("Requested save for all open Ghidra programs.")
    except Exception as exc:
        print(f"WARNING: save_all_programs failed before deploy: {exc}")
        try:
            _mcp_request(repo_root, mcp_url, "/save_program", timeout=60)
            print("Requested fallback Ghidra program save.")
        except Exception as fallback_exc:
            print(f"WARNING: fallback save_program failed before deploy: {fallback_exc}")
    try:
        _mcp_request(repo_root, mcp_url, "/exit_ghidra", timeout=10)
        print("Requested graceful Ghidra exit.")
    except Exception as exc:
        print(f"WARNING: exit_ghidra failed before deploy: {exc}")

    deadline = time.monotonic() + wait_seconds
    while time.monotonic() < deadline:
        if not _find_matching_ghidra_processes(ghidra_path):
            print("Ghidra exited cleanly.")
            return True
        time.sleep(1)
    for proc in _find_matching_ghidra_processes(ghidra_path):
        print(f"Force-killing Ghidra PID {proc['pid']}.")
        _terminate_process(int(proc["pid"]))
    return True


def wait_for_mcp(
    repo_root: Path,
    mcp_url: str,
    *,
    timeout_seconds: int = DEFAULT_MCP_WAIT_SECONDS,
) -> None:
    deadline = time.monotonic() + timeout_seconds
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        for path in ("/mcp/health", "/health", "/check_connection"):
            try:
                status, _payload = _mcp_request(repo_root, mcp_url, path, timeout=5)
                if status == 200:
                    print(f"MCP ready at {mcp_url} ({path}).")
                    return
            except Exception as exc:
                last_error = exc
        time.sleep(2)
    raise RuntimeError(f"MCP did not become ready at {mcp_url}: {last_error}")


def wait_for_project(
    repo_root: Path,
    mcp_url: str,
    *,
    timeout_seconds: int = DEFAULT_MCP_WAIT_SECONDS,
) -> None:
    deadline = time.monotonic() + timeout_seconds
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        try:
            _status, payload = _mcp_request(
                repo_root,
                mcp_url,
                "/list_project_files",
                params={"folder": "/"},
                timeout=5,
            )
            if isinstance(payload, dict) and "error" not in payload:
                print("Ghidra project is ready.")
                return
            last_error = RuntimeError(
                payload.get("error", str(payload)) if isinstance(payload, dict) else str(payload)
            )
        except Exception as exc:
            last_error = exc
        time.sleep(2)
    raise RuntimeError(f"Ghidra project did not become ready: {last_error}")


def _schema_tools(schema: object) -> set[str]:
    if not isinstance(schema, dict):
        return set()
    tools = schema.get("tools") or []
    names = set()
    for tool in tools:
        if not isinstance(tool, dict):
            continue
        name = tool.get("name")
        path = tool.get("path")
        if name:
            names.add(str(name))
        if path:
            names.add(str(path).lstrip("/"))
    return names


def _schema_tool_map(schema: object) -> dict[str, dict]:
    if not isinstance(schema, dict):
        return {}
    result: dict[str, dict] = {}
    for tool in schema.get("tools") or []:
        if not isinstance(tool, dict):
            continue
        keys = []
        if tool.get("name"):
            keys.append(str(tool["name"]))
        if tool.get("path"):
            keys.append(str(tool["path"]).lstrip("/"))
        for key in keys:
            result[key] = tool
    return result


def run_default_smoke_test(repo_root: Path, mcp_url: str) -> None:
    _status, schema = _mcp_request(repo_root, mcp_url, "/mcp/schema", timeout=20)
    tools = _schema_tools(schema)
    missing = sorted(SMOKE_REQUIRED_TOOLS - tools)
    if missing:
        raise RuntimeError(f"MCP schema missing required tools: {', '.join(missing)}")
    print(f"MCP smoke passed: schema exposes {len(tools)} tools.")


def _close_and_delete_project_file(repo_root: Path, mcp_url: str, program_path: str) -> None:
    deadline = time.monotonic() + 90
    last_error = ""
    while time.monotonic() < deadline:
        try:
            _mcp_request(
                repo_root,
                mcp_url,
                "/close_program",
                data={"name": program_path},
                method="POST",
                timeout=30,
            )
            _status, payload = _mcp_request(
                repo_root,
                mcp_url,
                "/delete_file",
                data={"filePath": program_path},
                method="POST",
                timeout=30,
            )
            _ensure_mcp_ok("/delete_file", payload)
            return
        except Exception as exc:
            last_error = str(exc)
            if "in use" not in last_error.lower() and "background" not in last_error.lower():
                raise
            time.sleep(3)
    raise RuntimeError(f"Timed out deleting {program_path}: {last_error}")


def reset_benchmark_fixture(repo_root: Path, mcp_url: str) -> None:
    benchmark_dll = repo_root / DEFAULT_BENCHMARK_DLL
    benchmark_debug_exe = repo_root / DEFAULT_BENCHMARK_DEBUG_EXE
    _terminate_processes_by_name("BenchmarkDebug.exe")
    if not benchmark_dll.is_file() or not benchmark_debug_exe.is_file():
        print("Benchmark binary output missing; building it now.")
        subprocess.run(
            [sys.executable, str(repo_root / "fun-doc" / "benchmark" / "build.py")],
            cwd=repo_root,
            check=True,
        )
    for program_path in (
        LEGACY_BENCHMARK_PROGRAM,
        DEFAULT_BENCHMARK_PROGRAM,
        DEFAULT_BENCHMARK_DEBUG_PROGRAM,
    ):
        _close_and_delete_project_file(repo_root, mcp_url, program_path)
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/create_folder",
        data={"path": DEFAULT_BENCHMARK_FOLDER},
        method="POST",
        timeout=30,
    )
    _ensure_mcp_ok("/create_folder", payload)
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/import_file",
        data={
            "file_path": str(benchmark_dll),
            "project_folder": DEFAULT_BENCHMARK_FOLDER,
            "auto_analyze": True,
        },
        method="POST",
        timeout=120,
    )
    _ensure_mcp_ok("/import_file", payload)
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/import_file",
        data={
            "file_path": str(benchmark_debug_exe),
            "project_folder": DEFAULT_BENCHMARK_FOLDER,
            "auto_analyze": True,
        },
        method="POST",
        timeout=120,
    )
    _ensure_mcp_ok("/import_file", payload)
    deadline = time.monotonic() + 90
    while time.monotonic() < deadline:
        try:
            _status, status = _mcp_request(
                repo_root,
                mcp_url,
                "/analysis_status",
                params={"program": DEFAULT_BENCHMARK_PROGRAM},
                timeout=10,
            )
            _ensure_mcp_ok("/analysis_status", status)
            _status, exe_status = _mcp_request(
                repo_root,
                mcp_url,
                "/analysis_status",
                params={"program": DEFAULT_BENCHMARK_DEBUG_PROGRAM},
                timeout=10,
            )
            _ensure_mcp_ok("/analysis_status", exe_status)
            state = (status.get("state") or status.get("status")) if isinstance(status, dict) else None
            exe_state = (
                (exe_status.get("state") or exe_status.get("status"))
                if isinstance(exe_status, dict)
                else None
            )
            is_idle = isinstance(status, dict) and status.get("analyzing") is False
            exe_idle = isinstance(exe_status, dict) and exe_status.get("analyzing") is False
            if (is_idle or state in {"complete", "done", "idle", "finished"}) and (
                exe_idle or exe_state in {"complete", "done", "idle", "finished"}
            ):
                print(f"Benchmark fixture reset at {DEFAULT_BENCHMARK_PROGRAM}.")
                return
        except Exception:
            pass
        time.sleep(2)
    print("WARNING: Benchmark analysis did not report complete within 90s; continuing.")


def _list_benchmark_functions(repo_root: Path, mcp_url: str) -> list[tuple[str, str]]:
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/list_functions",
        params={"program": DEFAULT_BENCHMARK_PROGRAM},
        timeout=60,
    )
    _ensure_mcp_ok("/list_functions", payload)
    functions: list[tuple[str, str]] = []
    if isinstance(payload, dict):
        raw_functions = payload.get("functions") or payload.get("results") or []
        for function in raw_functions:
            if not isinstance(function, dict):
                continue
            name = str(function.get("name") or "")
            address = function.get("address") or function.get("entry_point")
            if name and address:
                functions.append((name, str(address)))
    elif isinstance(payload, str):
        for line in payload.splitlines():
            match = re.match(r"(.+?)\s+at\s+([0-9a-fA-Fx]+)\s*$", line.strip())
            if match:
                functions.append((match.group(1), match.group(2)))
    return functions


def _list_benchmark_exports(repo_root: Path, mcp_url: str) -> list[tuple[str, str]]:
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/list_exports",
        params={"program": DEFAULT_BENCHMARK_PROGRAM},
        timeout=60,
    )
    _ensure_mcp_ok("/list_exports", payload)
    exports: list[tuple[str, str]] = []
    if isinstance(payload, str):
        for line in payload.splitlines():
            match = re.match(r"(.+?)\s+->\s+([0-9a-fA-Fx]+)\s*$", line.strip())
            if match:
                exports.append((match.group(1), match.group(2)))
    return exports


def _ensure_benchmark_function(repo_root: Path, mcp_url: str, address: str, name: str) -> None:
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/get_function_by_address",
        params={"program": DEFAULT_BENCHMARK_PROGRAM, "address": address},
        timeout=30,
    )
    if isinstance(payload, dict) and "error" not in payload:
        return
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/create_function",
        params={"program": DEFAULT_BENCHMARK_PROGRAM},
        data={
            "address": address,
            "name": re.sub(r"[^A-Za-z0-9_]", "_", name).strip("_") or "BenchmarkFunction",
            "disassemble_first": True,
        },
        method="POST",
        timeout=60,
    )
    _ensure_mcp_ok("/create_function", payload)


def _has_editable_variable(repo_root: Path, mcp_url: str, address: str) -> bool:
    _status, decompile_payload = _mcp_request(
        repo_root,
        mcp_url,
        "/decompile_function",
        params={"program": DEFAULT_BENCHMARK_PROGRAM, "address": address},
        timeout=60,
    )
    _ensure_mcp_ok("/decompile_function", decompile_payload)
    _status, variables = _mcp_request(
        repo_root,
        mcp_url,
        "/get_function_variables",
        params={"program": DEFAULT_BENCHMARK_PROGRAM, "address": address},
        timeout=30,
    )
    _ensure_mcp_ok("/get_function_variables", variables)
    if not isinstance(variables, dict):
        return False
    for variable in (variables.get("locals") or []) + (variables.get("parameters") or []):
        if isinstance(variable, dict) and variable.get("name") and not variable.get("is_phantom"):
            return True
    return False


def _find_benchmark_function(repo_root: Path, mcp_url: str, *, require_variable: bool = False) -> str:
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/search_functions",
        params={
            "program": DEFAULT_BENCHMARK_PROGRAM,
            "name_pattern": DEFAULT_BENCHMARK_FUNCTION,
            "limit": 10,
        },
        timeout=30,
    )
    functions = []
    if isinstance(payload, dict):
        _ensure_mcp_ok("/search_functions", payload)
        functions = payload.get("results") or payload.get("functions") or []
    for function in functions:
        if isinstance(function, dict) and DEFAULT_BENCHMARK_FUNCTION in str(function.get("name") or ""):
            address = function.get("address") or function.get("entry_point")
            if address and (not require_variable or _has_editable_variable(repo_root, mcp_url, str(address))):
                return str(address)

    fallback_functions = _list_benchmark_functions(repo_root, mcp_url)
    if require_variable:
        for _name, address in fallback_functions:
            if _has_editable_variable(repo_root, mcp_url, address):
                return address
    elif fallback_functions:
        return fallback_functions[0][1]

    for name, address in _list_benchmark_exports(repo_root, mcp_url):
        if name.startswith("Ordinal_") or name == "entry":
            continue
        if DEFAULT_BENCHMARK_FUNCTION not in name and not fallback_functions:
            continue
        _ensure_benchmark_function(repo_root, mcp_url, address, name)
        if not require_variable:
            return address
        for _ in range(5):
            if _has_editable_variable(repo_root, mcp_url, address):
                return address
            time.sleep(1)

    suffix = " with an editable variable" if require_variable else ""
    raise RuntimeError(f"Could not find a benchmark function{suffix} in {DEFAULT_BENCHMARK_PROGRAM}")


def run_benchmark_read_test(repo_root: Path, mcp_url: str) -> None:
    address = _find_benchmark_function(repo_root, mcp_url)
    read_calls = [
        ("/list_open_programs", {"program": DEFAULT_BENCHMARK_PROGRAM}),
        ("/search_data_types", {"program": DEFAULT_BENCHMARK_PROGRAM, "pattern": "int", "limit": 5}),
        ("/decompile_function", {"program": DEFAULT_BENCHMARK_PROGRAM, "address": address}),
        ("/get_function_variables", {"program": DEFAULT_BENCHMARK_PROGRAM, "address": address}),
        ("/analyze_function_completeness", {"program": DEFAULT_BENCHMARK_PROGRAM, "function_address": address}),
        ("/get_plate_comment", {"program": DEFAULT_BENCHMARK_PROGRAM, "address": address}),
        ("/save_program", {"program": DEFAULT_BENCHMARK_PROGRAM}),
    ]
    for path, params in read_calls:
        _status, payload = _mcp_request(repo_root, mcp_url, path, params=params, timeout=60)
        _ensure_mcp_ok(path, payload)
    struct_name = f"DeploySmokeStruct_{int(time.time())}"
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/create_struct",
        params={"program": DEFAULT_BENCHMARK_PROGRAM},
        data={
            "name": struct_name,
            "fields": [{"name": "dwValue", "type": "uint", "offset": 0}],
        },
        method="POST",
        timeout=60,
    )
    _ensure_mcp_ok("/create_struct", payload)
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/get_struct_layout",
        params={"program": DEFAULT_BENCHMARK_PROGRAM, "struct_name": struct_name},
        timeout=30,
    )
    _ensure_mcp_ok("/get_struct_layout", payload)
    print(f"Benchmark read/create test passed on benchmark function @ {address}.")


def run_benchmark_extended_read_test(repo_root: Path, mcp_url: str) -> None:
    address = _find_benchmark_function(repo_root, mcp_url)
    read_calls = [
        ("/list_project_files", {"folder": DEFAULT_BENCHMARK_FOLDER}),
        ("/analysis_status", {"program": DEFAULT_BENCHMARK_PROGRAM}),
        ("/list_functions", {"program": DEFAULT_BENCHMARK_PROGRAM}),
        (
            "/search_functions",
            {"program": DEFAULT_BENCHMARK_PROGRAM, "name_pattern": "FUN_", "limit": 10},
        ),
        ("/get_address_spaces", {"program": DEFAULT_BENCHMARK_PROGRAM}),
        ("/list_imports", {"program": DEFAULT_BENCHMARK_PROGRAM}),
        ("/list_exports", {"program": DEFAULT_BENCHMARK_PROGRAM}),
        ("/list_strings", {"program": DEFAULT_BENCHMARK_PROGRAM, "limit": 10}),
        ("/decompile_function", {"program": DEFAULT_BENCHMARK_PROGRAM, "address": address}),
    ]
    for path, params in read_calls:
        _status, payload = _mcp_request(repo_root, mcp_url, path, params=params, timeout=60)
        _ensure_mcp_ok(path, payload)
    print(f"Benchmark extended read test passed on benchmark function @ {address}.")


def run_benchmark_write_test(repo_root: Path, mcp_url: str) -> None:
    address = _find_benchmark_function(repo_root, mcp_url, require_variable=True)
    _status, variables = _mcp_request(
        repo_root,
        mcp_url,
        "/get_function_variables",
        params={"program": DEFAULT_BENCHMARK_PROGRAM, "address": address},
        timeout=30,
    )
    _ensure_mcp_ok("/get_function_variables", variables)
    variable_name = None
    if isinstance(variables, dict):
        for variable in (variables.get("locals") or []) + (variables.get("parameters") or []):
            if isinstance(variable, dict) and variable.get("name") and not variable.get("is_phantom"):
                variable_name = str(variable["name"])
                break
    if not variable_name:
        raise RuntimeError("No benchmark editable variable available for write smoke")
    write_calls = [
        (
            "/batch_set_comments",
            {
                "address": address,
                "plate_comment": "GhidraMCP deploy benchmark write probe",
                "disassembly_comments": [{"address": address, "comment": "deploy smoke"}],
            },
        ),
        (
            "/set_local_variable_type",
            {
                "function_address": address,
                "variable_name": variable_name,
                "new_type": "uint",
            },
        ),
        (
            "/rename_variables",
            {
                "function_address": address,
                "variable_renames": {variable_name: "dwDeploySmoke"},
                "force_individual": True,
            },
        ),
        (
            "/rename_function_by_address",
            {
                "function_address": address,
                "new_name": "DeploySmokeCalcCrc16",
            },
        ),
        (
            "/set_function_prototype",
            {
                "function_address": address,
                "prototype": "ushort DeploySmokeCalcCrc16(uchar * data, uint length)",
                "calling_convention": "__stdcall",
            },
        ),
    ]
    for path, data in write_calls:
        _status, payload = _mcp_request(
            repo_root,
            mcp_url,
            path,
            params={"program": DEFAULT_BENCHMARK_PROGRAM},
            data=data,
            method="POST",
            timeout=60,
        )
        _ensure_mcp_ok(path, payload)
    print(f"Benchmark write test passed on benchmark function @ {address}.")


def run_negative_contract_test(repo_root: Path, mcp_url: str) -> None:
    address = _find_benchmark_function(repo_root, mcp_url, require_variable=True)
    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/get_function_variables",
        params={"program": "/testing/benchmark/Missing.dll", "address": address},
        timeout=30,
    )
    _expect_mcp_error("/get_function_variables", payload, ("program not found", "available"))

    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/decompile_function",
        params={"program": DEFAULT_BENCHMARK_PROGRAM, "address": "not-an-address"},
        timeout=30,
    )
    _expect_mcp_error("/decompile_function", payload, ("address",))

    _status, payload = _mcp_request(
        repo_root,
        mcp_url,
        "/set_local_variable_type",
        params={"program": DEFAULT_BENCHMARK_PROGRAM},
        data={
            "function_address": address,
            "variable_name": "definitely_missing_local",
            "new_type": "uint",
        },
        method="POST",
        timeout=60,
    )
    _expect_mcp_error(
        "/set_local_variable_type",
        payload,
        ("definitely_missing_local", "available variables"),
    )
    print("Negative/error-shape contract test passed.")


def run_multi_program_targeting_test(repo_root: Path, mcp_url: str) -> None:
    address = _find_benchmark_function(repo_root, mcp_url)
    _status, programs = _mcp_request(repo_root, mcp_url, "/list_open_programs", timeout=30)
    _ensure_mcp_ok("/list_open_programs", programs)
    if not isinstance(programs, dict):
        raise RuntimeError("/list_open_programs returned an unexpected payload")
    open_programs = programs.get("programs") or []
    paths = {
        str(program.get("path"))
        for program in open_programs
        if isinstance(program, dict) and program.get("path")
    }
    if DEFAULT_BENCHMARK_PROGRAM not in paths:
        raise RuntimeError(f"{DEFAULT_BENCHMARK_PROGRAM} is not open; open paths: {sorted(paths)}")

    _status, by_path = _mcp_request(
        repo_root,
        mcp_url,
        "/get_function_variables",
        params={"program": DEFAULT_BENCHMARK_PROGRAM, "address": address},
        timeout=30,
    )
    _ensure_mcp_ok("/get_function_variables", by_path)
    if not isinstance(by_path, dict) or by_path.get("function_address") != address:
        raise RuntimeError("Program path targeting returned the wrong benchmark function")

    _status, by_name = _mcp_request(
        repo_root,
        mcp_url,
        "/analysis_status",
        params={"program": "Benchmark.dll"},
        timeout=30,
    )
    _ensure_mcp_ok("/analysis_status", by_name)
    _status, by_project_path = _mcp_request(
        repo_root,
        mcp_url,
        "/analysis_status",
        params={"program": DEFAULT_BENCHMARK_PROGRAM},
        timeout=30,
    )
    _ensure_mcp_ok("/analysis_status", by_project_path)
    print("Multi-program targeting test passed.")


class DebuggerLiveTestSkipped(Exception):
    """Raised by run_debugger_live_test when the test cannot run on this
    machine because a prerequisite is missing (Windows-only,
    BenchmarkDebug.exe absent, dbgeng backend unavailable, etc.).

    Distinct from RuntimeError so the release regression tier can
    distinguish "this test couldn't run" from "this test ran and
    failed." The deploy script catches it and prints a SKIPPED line
    rather than failing the whole release gate.
    """


# Substrings in a /debugger/launch error payload that indicate an
# environmental setup gap (missing WDK, dbgeng wiring, etc.) rather
# than a real regression. Each is observed in production logs:
#
#   * "dbgeng (.bat)': null"        — backend never came up
#   * "ghidratrace"                 — Python TraceRmi package mismatch
#   * "BenchmarkDebug.exe"          — taskkill stub couldn't find a
#                                     prior instance; means dbgeng never
#                                     launched anything in the first place
#   * "Could not load dbgeng"       — WDK not installed
_DEBUGGER_LAUNCH_SKIP_HINTS = (
    "dbgeng (.bat)': null",
    "ghidratrace",
    "could not load dbgeng",
    "dbgeng.dll",
    "no debugger backend",
)


def run_debugger_live_test(repo_root: Path, mcp_url: str) -> None:
    if os.name != "nt":
        raise DebuggerLiveTestSkipped(
            "Debugger live regression is currently Windows-only."
        )
    benchmark_debug_exe = repo_root / DEFAULT_BENCHMARK_DEBUG_EXE
    if not benchmark_debug_exe.is_file():
        raise DebuggerLiveTestSkipped(
            f"BenchmarkDebug.exe not found at {benchmark_debug_exe}. "
            "Build the benchmark fixture or set up the WDK toolchain."
        )

    env_values = load_env_file(repo_root / ".env")
    python_executable = (
        os.environ.get("GHIDRA_DEBUGGER_PYTHON", "").strip()
        or env_values.get("GHIDRA_DEBUGGER_PYTHON", "").strip()
    )
    launch_data: dict[str, object] = {
        "program": DEFAULT_BENCHMARK_DEBUG_PROGRAM,
        "executable_path": str(benchmark_debug_exe),
        "args": "--seconds 180",
        "cwd": str(benchmark_debug_exe.parent),
        "timeout_seconds": 90,
        "offer": "BATCH_FILE:local-dbgeng.bat",
    }
    if python_executable:
        launch_data["python_executable"] = python_executable

    try:
        _status, launch = _mcp_request(
            repo_root,
            mcp_url,
            "/debugger/launch",
            data=launch_data,
            method="POST",
            timeout=120,
        )
        try:
            _ensure_mcp_ok("/debugger/launch", launch)
        except RuntimeError as launch_err:
            # Classify the launch failure: a known-environmental cause
            # (no WDK, ghidratrace version mismatch, dbgeng backend
            # missing) becomes a skip; anything else is a real test
            # failure that should bubble up.
            msg = str(launch_err).lower()
            if any(hint in msg for hint in _DEBUGGER_LAUNCH_SKIP_HINTS):
                raise DebuggerLiveTestSkipped(
                    f"Debugger backend unavailable on this machine: "
                    f"{launch_err}. Install the Windows Debugger Toolkit "
                    "(WDK) and ensure the matching ghidratrace wheel is "
                    "installed against the active Python (see the "
                    "`debugger` dependency group: `uv sync --group debugger`) "
                    "to enable this test."
                ) from launch_err
            raise

        deadline = time.monotonic() + 45
        status_payload: object = {}
        while time.monotonic() < deadline:
            _status, status_payload = _mcp_request(
                repo_root,
                mcp_url,
                "/debugger/status",
                timeout=20,
            )
            _ensure_mcp_ok("/debugger/status", status_payload)
            if (
                isinstance(status_payload, dict)
                and status_payload.get("trace_active") is True
                and status_payload.get("target_connected") is True
                and status_payload.get("thread")
            ):
                break
            time.sleep(2)
        else:
            raise RuntimeError(f"Debugger did not report an active target: {status_payload}")

        for path, params in (
            ("/debugger/traces", {}),
            ("/debugger/modules", {}),
            ("/debugger/registers", {}),
            ("/debugger/stack_trace", {"depth": 8}),
        ):
            _status, payload = _mcp_request(
                repo_root,
                mcp_url,
                path,
                params=params,
                timeout=30,
            )
            _ensure_mcp_ok(path, payload)
        print("Debugger live test passed: launched BenchmarkDebug.exe and read trace state.")
    finally:
        _terminate_processes_by_name("BenchmarkDebug.exe")


def run_endpoint_catalog_test(repo_root: Path, mcp_url: str) -> None:
    _status, schema = _mcp_request(repo_root, mcp_url, "/mcp/schema", timeout=20)
    live_tools = _schema_tools(schema)
    catalog = json.loads((repo_root / "tests" / "endpoints.json").read_text(encoding="utf-8"))
    endpoints = catalog.get("endpoints", []) if isinstance(catalog, dict) else catalog
    expected = {
        str(endpoint.get("path", "")).lstrip("/")
        for endpoint in endpoints
        if isinstance(endpoint, dict) and endpoint.get("path")
    }
    missing = sorted(expected - live_tools)
    if missing:
        raise RuntimeError(f"Live schema missing {len(missing)} catalog endpoint(s): {', '.join(missing[:20])}")
    print(f"Endpoint catalog test passed: {len(expected)} catalog endpoints present.")


def run_selected_endpoint_contract_test(repo_root: Path, mcp_url: str) -> None:
    _status, schema = _mcp_request(repo_root, mcp_url, "/mcp/schema", timeout=20)
    tools = _schema_tool_map(schema)
    missing_tools = sorted(RELEASE_CONTRACT_TOOLS - set(tools))
    if missing_tools:
        raise RuntimeError(
            f"Release schema missing selected endpoint contract tool(s): {', '.join(missing_tools)}"
        )

    catalog = json.loads((repo_root / "tests" / "endpoints.json").read_text(encoding="utf-8"))
    endpoints = catalog.get("endpoints", []) if isinstance(catalog, dict) else catalog
    catalog_by_name = {
        str(endpoint.get("path", "")).lstrip("/"): endpoint
        for endpoint in endpoints
        if isinstance(endpoint, dict) and endpoint.get("path")
    }
    contract_errors: list[str] = []
    for name in sorted(RELEASE_CONTRACT_TOOLS):
        schema_tool = tools[name]
        catalog_tool = catalog_by_name.get(name)
        if catalog_tool is None:
            contract_errors.append(f"{name}: missing from tests/endpoints.json")
            continue
        schema_method = str(schema_tool.get("method") or "GET").upper()
        catalog_method = str(catalog_tool.get("method") or "GET").upper()
        if schema_method != catalog_method:
            contract_errors.append(f"{name}: method schema={schema_method} catalog={catalog_method}")
        schema_params = {
            str(param.get("name"))
            for param in schema_tool.get("params") or []
            if isinstance(param, dict) and param.get("name")
        }
        catalog_params = {str(param) for param in catalog_tool.get("params") or []}
        missing_params = sorted(catalog_params - schema_params)
        if missing_params:
            contract_errors.append(f"{name}: schema missing catalog params {missing_params}")
    if contract_errors:
        raise RuntimeError("Selected endpoint contract failed: " + "; ".join(contract_errors))
    print(f"Selected endpoint contract test passed for {len(RELEASE_CONTRACT_TOOLS)} tools.")


def _benchmark_regression_dir(repo_root: Path) -> Path:
    return repo_root / "fun-doc" / "benchmark" / "regression"


def _bench_get(repo_root: Path, mcp_url: str, path: str, params: dict | None = None,
               *, timeout: int = 30) -> tuple[int, object]:
    """GET wrapper for benchmark regression runner. Returns (status, parsed)."""
    return _mcp_request(repo_root, mcp_url, path, params=params, timeout=timeout)


def _bench_post(repo_root: Path, mcp_url: str, path: str, body: dict,
                *, timeout: int = 30) -> tuple[int, object]:
    return _mcp_request(repo_root, mcp_url, path, data=body, method="POST", timeout=timeout)


def _bench_text(parsed: object) -> str:
    """Coerce an MCP response into a single text blob for substring matching."""
    if isinstance(parsed, str):
        return parsed
    if isinstance(parsed, dict) and "result" in parsed and isinstance(parsed["result"], str):
        return parsed["result"]
    if isinstance(parsed, dict):
        return json.dumps(parsed)
    return str(parsed)


def _bench_lines(parsed: object) -> list[str]:
    text = _bench_text(parsed)
    return [line for line in text.splitlines() if line.strip()]


def _bench_assert_program_block(repo_root: Path, mcp_url: str, program_path: str,
                                 prog: dict, failures: list[str]) -> None:
    """Assert binary-level fields against /get_metadata, /list_segments etc."""
    p_query = {"program": program_path}

    _, meta = _bench_get(repo_root, mcp_url, "/get_metadata", p_query)
    meta_text = _bench_text(meta)
    if "architecture" in prog and f"Architecture: {prog['architecture']}" not in meta_text:
        failures.append(f"program.architecture: expected 'Architecture: {prog['architecture']}' in /get_metadata; got snippet: {meta_text[:120]!r}")
    if "language" in prog and prog["language"] not in meta_text:
        failures.append(f"program.language: expected '{prog['language']}' in /get_metadata; got snippet: {meta_text[:120]!r}")
    if "compiler" in prog and f"Compiler: {prog['compiler']}" not in meta_text:
        failures.append(f"program.compiler: expected 'Compiler: {prog['compiler']}' in /get_metadata")

    if "function_count_min" in prog:
        _, fc = _bench_get(repo_root, mcp_url, "/get_function_count", p_query)
        actual = fc.get("function_count") if isinstance(fc, dict) else None
        if not isinstance(actual, (int, float)) or actual < prog["function_count_min"]:
            failures.append(f"program.function_count_min: expected >={prog['function_count_min']}; got {actual}")

    if "string_count_min" in prog:
        _, strs = _bench_get(repo_root, mcp_url, "/list_strings", p_query)
        n = len(_bench_lines(strs))
        if n < prog["string_count_min"]:
            failures.append(f"program.string_count_min: expected >={prog['string_count_min']}; got {n}")

    if "segments" in prog:
        _, segs = _bench_get(repo_root, mcp_url, "/list_segments", p_query)
        seg_text = _bench_text(segs)
        for spec in prog["segments"]:
            need = spec["name"] + ":"
            if need not in seg_text:
                failures.append(f"program.segments: expected '{need}' in /list_segments")

    if "must_contain_strings" in prog:
        _, strs = _bench_get(repo_root, mcp_url, "/list_strings", p_query)
        strs_text = _bench_text(strs)
        for needle in prog["must_contain_strings"]:
            if needle not in strs_text:
                failures.append(f"program.must_contain_strings: expected {needle!r} in /list_strings")


def _bench_assert_function(repo_root: Path, mcp_url: str, program_path: str,
                            entry: dict, failures: list[str]) -> None:
    addr = entry["address"]
    p_query = {"program": program_path, "address": addr}

    # /get_function_by_address gives us the canonical name + thunk flag in text form.
    _, by_addr = _bench_get(repo_root, mcp_url, "/get_function_by_address", p_query)
    by_addr_text = _bench_text(by_addr)
    if "name" in entry:
        needle = f"Function: {entry['name']} at"
        if needle not in by_addr_text:
            failures.append(f"function@{addr}.name: expected '{needle}' in /get_function_by_address; got: {by_addr_text[:160]!r}")

    # /get_function_signature returns JSON with structural fields.
    _, sig = _bench_get(repo_root, mcp_url, "/get_function_signature", p_query)
    if not isinstance(sig, dict):
        failures.append(f"function@{addr}.signature: /get_function_signature did not return JSON")
        return

    if "param_count" in entry and sig.get("param_count") != entry["param_count"]:
        failures.append(f"function@{addr}.param_count: expected {entry['param_count']}; got {sig.get('param_count')}")
    if "basic_block_count" in entry and sig.get("basic_block_count") != entry["basic_block_count"]:
        failures.append(f"function@{addr}.basic_block_count: expected {entry['basic_block_count']}; got {sig.get('basic_block_count')}")
    if "cyclomatic_complexity" in entry and sig.get("cyclomatic_complexity") != entry["cyclomatic_complexity"]:
        failures.append(f"function@{addr}.cyclomatic_complexity: expected {entry['cyclomatic_complexity']}; got {sig.get('cyclomatic_complexity')}")
    if "instruction_count_min" in entry:
        ic = sig.get("instruction_count")
        if not isinstance(ic, int) or ic < entry["instruction_count_min"]:
            failures.append(f"function@{addr}.instruction_count_min: expected >={entry['instruction_count_min']}; got {ic}")
    if "immediate_values_contains" in entry:
        actual_imm = set(sig.get("immediate_values") or [])
        for v in entry["immediate_values_contains"]:
            if v not in actual_imm:
                failures.append(f"function@{addr}.immediate_values_contains: expected {v} (0x{v:x}) in /get_function_signature.immediate_values; got {sorted(actual_imm)}")
    if "string_constants_contains" in entry:
        actual = set(sig.get("string_constants") or [])
        for s in entry["string_constants_contains"]:
            if s not in actual:
                failures.append(f"function@{addr}.string_constants_contains: expected {s!r} in /get_function_signature.string_constants")
    if "callee_names_contains" in entry:
        actual = set(sig.get("callee_names") or [])
        for s in entry["callee_names_contains"]:
            if s not in actual:
                failures.append(f"function@{addr}.callee_names_contains: expected {s!r} in /get_function_signature.callee_names")
    if "return_type_contains" in entry:
        # Return type appears in the by_addr "Signature:" line.
        sig_line = ""
        for line in by_addr_text.splitlines():
            if line.strip().startswith("Signature:"):
                sig_line = line
                break
        if entry["return_type_contains"] not in sig_line:
            failures.append(f"function@{addr}.return_type_contains: expected {entry['return_type_contains']!r} in 'Signature:' line; got {sig_line!r}")
    if "is_thunk" in entry:
        # /get_function_by_address doesn't expose thunk explicitly in text;
        # rely on the signature presence as a proxy. Treat is_thunk:false
        # as "we got a signature" — sufficient for the user-authored functions.
        if entry["is_thunk"] is False and not by_addr_text.strip().startswith("Function:"):
            failures.append(f"function@{addr}.is_thunk=false: function did not resolve via /get_function_by_address")
    if "signature_contains" in entry:
        for needle in entry["signature_contains"]:
            if needle not in by_addr_text:
                failures.append(f"function@{addr}.signature_contains: expected {needle!r} in /get_function_by_address")

    if entry.get("xref_count_to_min", 0) > 0:
        _, xrefs = _bench_get(repo_root, mcp_url, "/get_xrefs_to", p_query)
        n = len(_bench_lines(xrefs))
        if n < entry["xref_count_to_min"]:
            failures.append(f"function@{addr}.xref_count_to_min: expected >={entry['xref_count_to_min']}; got {n}")

    if entry.get("decompile_must_be_nonempty") or entry.get("decompile_contains"):
        _, dec = _bench_get(repo_root, mcp_url, "/decompile_function", p_query, timeout=60)
        dec_text = _bench_text(dec)
        if entry.get("decompile_must_be_nonempty") and not dec_text.strip():
            failures.append(f"function@{addr}.decompile_must_be_nonempty: /decompile_function returned empty")
        for needle in entry.get("decompile_contains", []):
            if needle not in dec_text:
                failures.append(f"function@{addr}.decompile_contains: expected {needle!r} in /decompile_function output")


def _bench_assert_endpoint_smoke(repo_root: Path, mcp_url: str, program_path: str,
                                  entry: dict, failures: list[str]) -> None:
    endpoint = entry["endpoint"]
    method = entry.get("method", "GET").upper()
    params = dict(entry.get("params") or {})
    body = entry.get("body")
    assertion = entry.get("assert") or {}
    a_type = assertion.get("type", "nonempty")

    # Auto-add program= for endpoints that take a target program (most do).
    # Skip for genuinely program-less endpoints.
    program_less = {"/check_connection", "/list_open_programs", "/list_calling_conventions",
                    "/list_scripts", "/check_tools", "/list_data_type_categories"}
    if endpoint not in program_less and "program" not in params:
        params["program"] = program_path

    try:
        if method == "POST":
            payload = body or {}
            # POST endpoints take program= as query param per project convention.
            url_params = {"program": program_path} if endpoint not in program_less else {}
            _, parsed = _mcp_request(repo_root, mcp_url, endpoint,
                                      data=payload, method="POST",
                                      params=url_params or None, timeout=60)
        else:
            _, parsed = _bench_get(repo_root, mcp_url, endpoint, params or None)
    except Exception as exc:
        failures.append(f"endpoint_smoke {endpoint}: request failed: {exc}")
        return

    # Surface MCP-level error envelope explicitly.
    if isinstance(parsed, dict) and parsed.get("error"):
        failures.append(f"endpoint_smoke {endpoint}: MCP error: {parsed['error']}")
        return

    text = _bench_text(parsed)
    lines = _bench_lines(parsed)

    if a_type == "nonempty":
        if not text.strip():
            failures.append(f"endpoint_smoke {endpoint}: expected nonempty response")
    elif a_type == "lines":
        if "min_lines" in assertion and len(lines) < assertion["min_lines"]:
            failures.append(f"endpoint_smoke {endpoint}: expected >={assertion['min_lines']} lines; got {len(lines)}")
        if "max_lines" in assertion and len(lines) > assertion["max_lines"]:
            failures.append(f"endpoint_smoke {endpoint}: expected <={assertion['max_lines']} lines; got {len(lines)}")
        for needle in assertion.get("contains", []):
            if not any(needle in line for line in lines):
                failures.append(f"endpoint_smoke {endpoint}: expected {needle!r} in some line")
    elif a_type == "text":
        for needle in assertion.get("contains", []):
            if needle not in text:
                failures.append(f"endpoint_smoke {endpoint}: expected {needle!r} in response")
    elif a_type == "json":
        if not isinstance(parsed, dict):
            failures.append(f"endpoint_smoke {endpoint}: expected JSON object response")
        else:
            for key in assertion.get("contains_keys", []):
                if key not in parsed:
                    failures.append(f"endpoint_smoke {endpoint}: expected key {key!r} in response")
    else:
        failures.append(f"endpoint_smoke {endpoint}: unknown assertion type {a_type!r}")


def _bench_ensure_full_analysis(repo_root: Path, mcp_url: str, program_path: str) -> None:
    """Re-run analysis on a freshly-imported binary to surface exported functions.

    AutoImporter's auto_analyze=true runs only the default fast analyzers,
    leaving export-table entries unbound to Function objects. Calling
    /run_analysis after import promotes them and runs the richer "Decompiler
    Switch Analysis" / "Aggressive Instruction Finder" passes that the
    CodeBrowser's interactive auto-analyze runs by default.

    /run_analysis returns in milliseconds — analysis happens on a background
    thread. Without polling for completion the YAML asserts race the analyzer
    and see partial state. Poll /analysis_status.analyzing until it flips
    back to false (or BENCHMARK_ANALYSIS_TIMEOUT_S timeout).

    The timeout must cover a *cold* install: on a freshly-created Ghidra
    user-config dir (e.g. the first deploy to a newly-installed Ghidra patch
    release) the initial analysis of Benchmark.dll + BenchmarkDebug.exe can run
    well past a minute, so a 60s cap raced the analyzer and produced spurious
    `param_count: 0` / `undefined` signature failures.
    """
    try:
        _, _ = _mcp_request(repo_root, mcp_url, "/run_analysis",
                             params={"program": program_path},
                             method="POST", timeout=120)
    except Exception as exc:
        print(f"WARNING: /run_analysis on {program_path} failed: {exc}")
        return

    deadline = time.monotonic() + BENCHMARK_ANALYSIS_TIMEOUT_S
    while time.monotonic() < deadline:
        try:
            _, status = _mcp_request(repo_root, mcp_url, "/analysis_status",
                                      params={"program": program_path}, timeout=10)
        except Exception:
            time.sleep(1)
            continue
        if isinstance(status, dict) and status.get("analyzing") is False:
            return
        time.sleep(2)
    print(f"WARNING: /analysis_status on {program_path} still busy after "
          f"{BENCHMARK_ANALYSIS_TIMEOUT_S}s; proceeding anyway")


def run_benchmark_yaml_regression(repo_root: Path, mcp_url: str) -> None:
    """Run YAML-driven assertions against the imported benchmark binaries.

    Reads every fun-doc/benchmark/regression/*.yaml file and verifies its
    contents end-to-end against the live MCP server. Failures are collected
    across the whole pass and raised as a single RuntimeError so a single
    deploy run reports every regression at once.
    """
    try:
        import yaml  # type: ignore
    except ImportError as exc:
        raise RuntimeError("PyYAML required for benchmark YAML regression") from exc

    reg_dir = _benchmark_regression_dir(repo_root)
    if not reg_dir.is_dir():
        print(f"benchmark YAML regression: {reg_dir} not found, skipping")
        return

    yaml_files = sorted(p for p in reg_dir.glob("*.yaml"))
    if not yaml_files:
        print(f"benchmark YAML regression: no *.yaml files in {reg_dir}, skipping")
        return

    total_assertions = 0
    total_skipped = 0
    failures: list[str] = []

    for yaml_path in yaml_files:
        spec = yaml.safe_load(yaml_path.read_text(encoding="utf-8")) or {}
        prog_block = spec.get("program") or {}
        program_path = prog_block.get("path")
        if not program_path:
            failures.append(f"{yaml_path.name}: missing program.path")
            continue

        # Ensure the binary has full-analysis state before asserting against it.
        # AutoImporter's auto_analyze=true runs only fast analyzers; the richer
        # passes (which surface exports as Function objects + run Decompiler
        # Switch Analysis) need an explicit /run_analysis call.
        _bench_ensure_full_analysis(repo_root, mcp_url, program_path)

        binary_failures_before = len(failures)

        if prog_block:
            _bench_assert_program_block(repo_root, mcp_url, program_path, prog_block, failures)

        for fn in spec.get("functions") or []:
            total_assertions += 1
            _bench_assert_function(repo_root, mcp_url, program_path, fn, failures)

        for smoke in spec.get("endpoint_smoke") or []:
            total_assertions += 1
            _bench_assert_endpoint_smoke(repo_root, mcp_url, program_path, smoke, failures)

        for skip in spec.get("skipped") or []:
            total_skipped += 1

        new_fails = len(failures) - binary_failures_before
        status = "OK" if new_fails == 0 else f"{new_fails} FAILURES"
        print(f"  {yaml_path.name}: {status}")

    if failures:
        msg = (f"Benchmark YAML regression failed ({len(failures)} assertion(s)):\n  - "
               + "\n  - ".join(failures))
        raise RuntimeError(msg)

    print(f"Benchmark YAML regression passed: {total_assertions} assertion-blocks, {total_skipped} explicit skip(s).")


def run_release_regression_tests(repo_root: Path, mcp_url: str) -> None:
    reset_benchmark_fixture(repo_root, mcp_url)
    run_selected_endpoint_contract_test(repo_root, mcp_url)
    run_benchmark_extended_read_test(repo_root, mcp_url)
    run_benchmark_yaml_regression(repo_root, mcp_url)
    run_multi_program_targeting_test(repo_root, mcp_url)
    run_negative_contract_test(repo_root, mcp_url)
    try:
        run_debugger_live_test(repo_root, mcp_url)
    except DebuggerLiveTestSkipped as skip:
        # Environmental prerequisite missing — don't fail the release
        # gate. Surface the reason so the operator can decide whether to
        # set up the toolchain locally before the next release cut.
        print(f"SKIPPED debugger live test: {skip}")
    print("Release regression tier passed.")


def run_deploy_tests(repo_root: Path, mcp_url: str, test_modes: list[str]) -> None:
    run_default_smoke_test(repo_root, mcp_url)
    if _deploy_tests_use_benchmark(test_modes):
        _mcp_request(
            repo_root,
            mcp_url,
            "/prompt_policy",
            data={"action": "enable", "reason": "deploy_tests", "seconds": 300},
            method="POST",
            timeout=10,
        )
    for mode in test_modes:
        if mode == "endpoint-catalog":
            run_endpoint_catalog_test(repo_root, mcp_url)
        elif mode == "benchmark-read":
            reset_benchmark_fixture(repo_root, mcp_url)
            run_benchmark_extended_read_test(repo_root, mcp_url)
        elif mode == "benchmark-write":
            reset_benchmark_fixture(repo_root, mcp_url)
            run_benchmark_write_test(repo_root, mcp_url)
        elif mode == "negative-contract":
            reset_benchmark_fixture(repo_root, mcp_url)
            run_negative_contract_test(repo_root, mcp_url)
        elif mode == "multi-program":
            reset_benchmark_fixture(repo_root, mcp_url)
            run_multi_program_targeting_test(repo_root, mcp_url)
        elif mode == "selected-contract":
            run_selected_endpoint_contract_test(repo_root, mcp_url)
        elif mode == "debugger-live":
            reset_benchmark_fixture(repo_root, mcp_url)
            try:
                run_debugger_live_test(repo_root, mcp_url)
            except DebuggerLiveTestSkipped as skip:
                print(f"SKIPPED debugger live test: {skip}")
        elif mode == "release":
            run_release_regression_tests(repo_root, mcp_url)


def _resolve_debugger_python(repo_root: Path) -> Path | None:
    """Find the Python interpreter Ghidra's debugger launchers actually use.

    The Ghidra dbgeng / gdb / lldb launcher .bat / .sh scripts run
    ``"%OPT_PYTHON_EXE%" ...`` (defaulting to ``python``), and the
    GhidraMCP plugin propagates ``GHIDRA_DEBUGGER_PYTHON`` from .env into
    that variable when running the debugger live test. So the
    interpreter to install ``ghidratrace`` into is:

      1. ``GHIDRA_DEBUGGER_PYTHON`` from the environment, if set
      2. ``GHIDRA_DEBUGGER_PYTHON`` from ``<repo>/.env``, if set
      3. ``shutil.which("python")`` as the system default

    Returns ``None`` only when no resolvable interpreter is found —
    rare on Windows but possible in headless CI containers.
    """
    candidate = os.environ.get("GHIDRA_DEBUGGER_PYTHON", "").strip()
    if not candidate:
        env_values = load_env_file(repo_root / ".env")
        candidate = env_values.get("GHIDRA_DEBUGGER_PYTHON", "").strip()
    if candidate:
        path = Path(candidate)
        if path.is_file():
            return path
    fallback = shutil.which("python")
    return Path(fallback) if fallback else None


def install_ghidratrace_for_debugger(
    repo_root: Path,
    ghidra_path: Path,
    *,
    dry_run: bool = False,
) -> int:
    """Install the matching ``ghidratrace`` wheel into the launcher Python.

    Why this exists: when Ghidra is upgraded (e.g., 12.0.4 → 12.1), the
    wheel that ships at ``<ghidra>/Ghidra/Debug/Debugger-rmi-trace/pypkg/dist``
    bumps version too. If a stale 12.0 ``ghidratrace`` is still
    pip-installed in the launcher's Python, TraceRmi negotiation fails
    with ``VersionMismatchError: Front-end: 12.1, back-end: 12.0`` —
    observed twice in this release cycle. The wheel lives inside the
    Ghidra install (not on PyPI), so a plain ``uv sync --group debugger``
    can't cover it.

    Returns 0 on success / no-op, nonzero on installer failure.
    """
    wheel_dir = ghidra_path / "Ghidra" / "Debug" / "Debugger-rmi-trace" / "pypkg" / "dist"
    wheels = sorted(wheel_dir.glob("ghidratrace-*-py3-none-any.whl"))
    if not wheels:
        print(f"  No ghidratrace wheel found under {wheel_dir} — skipping debugger Python sync")
        return 0
    wheel = wheels[-1]

    debugger_python = _resolve_debugger_python(repo_root)
    if debugger_python is None:
        print("  Could not resolve a debugger Python (set GHIDRA_DEBUGGER_PYTHON) — skipping")
        return 0

    if dry_run:
        print(f"DRY RUN: {debugger_python} -m pip install --force-reinstall {wheel}")
        print(f"DRY RUN: {debugger_python} -m pip install --upgrade 'protobuf>=6.31.0'")
        return 0

    # protobuf>=6.31.0 is gated separately by ghidratrace.setuputils — install
    # it before the wheel so the post-install setuputils check doesn't trip.
    pb = subprocess.run(
        [str(debugger_python), "-m", "pip", "install", "--upgrade", "protobuf>=6.31.0"],
        check=False, capture_output=True, text=True,
    )
    if pb.returncode != 0:
        print(f"  protobuf install failed: {pb.stderr.strip()[:200]}")
        return pb.returncode

    gt = subprocess.run(
        [str(debugger_python), "-m", "pip", "install", "--force-reinstall", str(wheel)],
        check=False, capture_output=True, text=True,
    )
    if gt.returncode != 0:
        print(f"  ghidratrace install failed: {gt.stderr.strip()[:200]}")
        return gt.returncode

    print(f"  Installed {wheel.name} into {debugger_python}")
    return 0


def install_ghidra_dependencies(
    repo_root: Path,
    ghidra_path: Path,
    *,
    force: bool = False,
    dry_run: bool = False,
) -> int:
    maven_command = str(find_maven_command())
    ghidra_version = read_pom_versions(repo_root).ghidra_version
    m2_root = Path.home() / ".m2" / "repository" / "ghidra"

    for artifact_id, relative_path in REQUIRED_GHIDRA_JARS:
        jar_path = ghidra_path / relative_path
        if not jar_path.is_file():
            raise FileNotFoundError(f"Missing required Ghidra jar: {jar_path}")

        cached_jar = (
            m2_root
            / artifact_id
            / ghidra_version
            / f"{artifact_id}-{ghidra_version}.jar"
        )
        if cached_jar.is_file() and not force:
            print(f"Skipping already installed dependency: {artifact_id}")
            continue

        command = [
            maven_command,
            "install:install-file",
            f"-Dfile={jar_path}",
            "-DgroupId=ghidra",
            f"-DartifactId={artifact_id}",
            f"-Dversion={ghidra_version}",
            "-Dpackaging=jar",
            "-DgeneratePom=true",
        ]
        if dry_run:
            print("DRY RUN:", end=" ")
            print_command(command)
            continue

        completed = subprocess.run(command, cwd=repo_root, check=False)
        if completed.returncode != 0:
            return completed.returncode

    # Keep the debugger launcher's Python in sync with the installed
    # Ghidra version's ghidratrace wheel. Without this, a Ghidra version
    # bump leaves a stale ghidratrace pip-installed in the launcher's
    # Python and TraceRmi negotiation fails with the back-end reporting
    # the old version. Best-effort: a failure here does NOT block the
    # main JAR-install dependency setup since most users don't use the
    # live debugger.
    install_ghidratrace_for_debugger(repo_root, ghidra_path, dry_run=dry_run)

    return 0


def test_write_access(path_to_test: Path) -> bool:
    try:
        path_to_test.mkdir(parents=True, exist_ok=True)
        probe = path_to_test / ".ghidra-mcp-write-test"
        probe.write_text("ok", encoding="utf-8")
        probe.unlink()
        return True
    except OSError:
        return False


def _has_dependency_group(pyproject: Path, group: str) -> bool:
    """Return True if ``pyproject.toml`` defines ``group`` under
    ``[dependency-groups]``.

    A plain substring scan is too loose — the word could appear in a comment or
    an unrelated section — and too strict, since it wouldn't confirm the entry
    is one ``uv sync --group <group>`` can actually resolve. Parse the TOML and
    look for the real key.
    """
    if not pyproject.is_file():
        return False

    try:
        import tomllib  # Python 3.11+
    except ModuleNotFoundError:
        try:
            import tomli as tomllib  # type: ignore[no-redef]
        except ModuleNotFoundError:
            tomllib = None  # type: ignore[assignment]

    if tomllib is not None:
        try:
            with pyproject.open("rb") as handle:
                data = tomllib.load(handle)
        except (OSError, ValueError):
            return False
        groups = data.get("dependency-groups")
        return isinstance(groups, dict) and group in groups

    # Python 3.10 without tomli: fall back to a section-scoped scan so the word
    # only counts when it's a key inside [dependency-groups].
    try:
        text = pyproject.read_text(encoding="utf-8")
    except OSError:
        return False
    in_section = False
    key_re = re.compile(rf"^\s*(?:{re.escape(group)}|[\"']{re.escape(group)}[\"'])\s*=")
    for raw in text.splitlines():
        line = raw.split("#", 1)[0]
        stripped = line.strip()
        if stripped.startswith("[") and stripped.endswith("]"):
            in_section = stripped == "[dependency-groups]"
            continue
        if in_section and key_re.match(line):
            return True
    return False


def collect_preflight_issues(
    repo_root: Path,
    ghidra_path: Path,
    python_executable: Path,
    *,
    install_debugger: bool,
    strict: bool = False,
    user_base_dir: Path | None = None,
) -> list[str]:
    from .requirements import ensure_uv_available

    issues: list[str] = []

    try:
        ensure_uv_available()
    except FileNotFoundError as exc:
        issues.append(str(exc))

    if shutil.which("java") is None:
        issues.append("Java not found on PATH (JDK 21 recommended).")

    try:
        find_ghidra_executable(ghidra_path)
    except FileNotFoundError:
        issues.append(f"Ghidra executable not found at: {ghidra_path}")
        return issues

    for _artifact_id, relative_path in REQUIRED_GHIDRA_JARS:
        jar_path = ghidra_path / relative_path
        if not jar_path.is_file():
            issues.append(f"Missing required Ghidra dependency: {jar_path}")

    if install_debugger:
        pyproject = repo_root / "pyproject.toml"
        if not _has_dependency_group(pyproject, "debugger"):
            issues.append(
                "Debugger dependency group not found in pyproject.toml "
                "(expected a [dependency-groups] 'debugger' entry)"
            )

    extensions_dir = ghidra_path / "Extensions" / "Ghidra"
    if not test_write_access(extensions_dir):
        issues.append(
            f"No write access to Ghidra extensions directory: {extensions_dir}"
        )

    user_extension_dir = (
        resolve_ghidra_user_dir(ghidra_path, user_base_dir) / "Extensions"
    )
    if not test_write_access(user_extension_dir):
        issues.append(
            f"No write access to user extension directory: {user_extension_dir}"
        )

    if strict:
        for url in ("https://repo.maven.apache.org", "https://pypi.org"):
            try:
                request = urllib.request.Request(url, method="HEAD")
                with urllib.request.urlopen(request, timeout=10):
                    pass
            except Exception:
                issues.append(f"Network check failed: {url}")

    return issues


def build_bridge_wheel(repo_root: Path, *, dry_run: bool = False) -> Path | None:
    """Build the bridge wheel with ``uv build`` and return its path.

    The Python bridge ships as a wheel (``ghidra_mcp_bridge-*.whl``) rather than
    a loose ``bridge_mcp_ghidra.py`` script. Returns the newest built wheel, or
    None on a dry run / when no wheel is produced.
    """
    from .requirements import ensure_uv_available

    dist_dir = repo_root / "dist"
    if dry_run:
        print(f"DRY RUN: uv build --wheel (-> {dist_dir})")
        return None
    uv = ensure_uv_available()
    subprocess.run([uv, "build", "--wheel"], check=True, cwd=str(repo_root))
    wheels = sorted(
        dist_dir.glob("ghidra_mcp_bridge-*.whl"), key=lambda p: p.stat().st_mtime
    )
    return wheels[-1] if wheels else None


def deploy_to_ghidra(
    repo_root: Path,
    ghidra_path: Path,
    *,
    dry_run: bool = False,
    test_modes: list[str] | None = None,
) -> int:
    archive_path = find_plugin_archive(repo_root)
    extensions_dir = ghidra_path / "Extensions" / "Ghidra"
    destination_archive = extensions_dir / archive_path.name
    dotenv_source = repo_root / ".env"
    user_base_dir = ghidra_user_base_dir()
    mcp_url = resolve_mcp_url(repo_root)
    test_modes = resolve_deploy_test_modes(repo_root, test_modes)

    close_running_ghidra_for_deploy(
        repo_root, ghidra_path, mcp_url=mcp_url, dry_run=dry_run
    )

    if dry_run:
        print(f"DRY RUN: ensure directory {extensions_dir}")
        print(
            f"DRY RUN: remove existing archives matching {extensions_dir / 'GhidraMCP*.zip'}"
        )
        print(f"DRY RUN: copy {archive_path} -> {destination_archive}")
        build_bridge_wheel(repo_root, dry_run=True)
        print(f"DRY RUN: copy built bridge wheel -> {ghidra_path}")
        if dotenv_source.is_file():
            print(
                f"DRY RUN: copy {dotenv_source} -> {ghidra_path / dotenv_source.name}"
            )
        install_user_extension(repo_root, ghidra_path, archive_path, dry_run=True)
        target_user_dir = resolve_ghidra_user_dir(ghidra_path, user_base_dir)
        patch_ghidra_user_configs(user_base_dir, target_user_dir, dry_run=True)
        if _deploy_tests_use_benchmark(test_modes):
            clear_restored_benchmark_tools(repo_root, dry_run=True)
        start_ghidra(ghidra_path, repo_root=repo_root, dry_run=True)
        print(f"DRY RUN: wait up to {DEFAULT_MCP_WAIT_SECONDS}s for MCP at {mcp_url}")
        print(f"DRY RUN: wait up to {DEFAULT_MCP_WAIT_SECONDS}s for active project")
        print("DRY RUN: run default MCP smoke test")
        for mode in test_modes:
            print(f"DRY RUN: run deploy test {mode}")
        return 0

    extensions_dir.mkdir(parents=True, exist_ok=True)
    for existing_archive in extensions_dir.glob("GhidraMCP*.zip"):
        existing_archive.unlink()

    shutil.copy2(archive_path, destination_archive)
    print(f"Installed plugin archive to {destination_archive}")

    bridge_wheel = build_bridge_wheel(repo_root)
    if bridge_wheel is not None and bridge_wheel.is_file():
        wheel_destination = ghidra_path / bridge_wheel.name
        wheel_destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(bridge_wheel, wheel_destination)
        print(f"Copied bridge wheel to {wheel_destination}")

    if dotenv_source.is_file():
        dotenv_destination = ghidra_path / dotenv_source.name
        shutil.copy2(dotenv_source, dotenv_destination)
        print(f"Copied .env to {dotenv_destination}")

    install_user_extension(repo_root, ghidra_path, archive_path)
    target_user_dir = resolve_ghidra_user_dir(ghidra_path, user_base_dir)
    patch_ghidra_user_configs(user_base_dir, target_user_dir)
    if _deploy_tests_use_benchmark(test_modes):
        clear_restored_benchmark_tools(repo_root)
    start_ghidra(ghidra_path, repo_root=repo_root)
    wait_for_mcp(repo_root, mcp_url, timeout_seconds=DEFAULT_MCP_WAIT_SECONDS)
    wait_for_project(repo_root, mcp_url, timeout_seconds=DEFAULT_MCP_WAIT_SECONDS)
    run_deploy_tests(repo_root, mcp_url, test_modes)

    return 0


def start_ghidra(ghidra_path: Path, *, repo_root: Path | None = None, dry_run: bool = False) -> int:
    executable = find_ghidra_executable(ghidra_path)
    env_root = repo_root if repo_root is not None else Path.cwd()
    env_values = load_env_file(env_root / ".env")
    project_path = env_values.get("GHIDRA_PROJECT_PATH", "").strip()
    if executable.suffix.lower() in {".bat", ".cmd"}:
        command = [os.environ.get("COMSPEC", "cmd.exe"), "/c", str(executable)]
    else:
        command = [str(executable)]
    if project_path:
        command.append(project_path)

    if dry_run:
        print("DRY RUN:", end=" ")
        print_command(command)
        return 0

    subprocess.Popen(command, cwd=ghidra_path, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    print(f"Started Ghidra from {executable}")
    return 0


def clean_all(repo_root: Path, *, dry_run: bool = False) -> int:
    paths_to_remove = [
        repo_root / "target",
        repo_root / ".pytest_cache",
        repo_root / "__pycache__",
    ]

    log_dir = repo_root / "logs"
    log_files = sorted(log_dir.glob("*.log")) if log_dir.is_dir() else []

    for path in paths_to_remove:
        if not path.exists():
            continue
        if dry_run:
            print(f"DRY RUN: remove {path}")
            continue
        if path.is_dir():
            shutil.rmtree(path, ignore_errors=True)
        else:
            path.unlink(missing_ok=True)

    for log_file in log_files:
        if dry_run:
            print(f"DRY RUN: remove {log_file}")
            continue
        log_file.unlink(missing_ok=True)

    print("Cleanup completed.")
    return 0
