#!/usr/bin/env python3
"""Prepare and run the local interactive self-update smoke harness.

This harness creates a disposable Godot project with a physical
addons/godot_ai directory, patches that fixture only, and stages a synthetic
v(N+1) plugin ZIP that adds a typed Dict/Array hot-reload trigger plus a new
class_name dependency chain. The operator still performs the one step that
matters: click Update in the dock.
"""

from __future__ import annotations

import argparse
import os
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.request
import zipfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
PLUGIN_ROOT = ROOT / "plugin" / "addons" / "godot_ai"
DEFAULT_PROJECT_DIR = Path(tempfile.gettempdir()) / "godot-ai-self-update-smoke"
PROJECT_HEADER = "; Generated by script/local-self-update-smoke."
SMOKE_DIR = "self_update_smoke"
SMOKE_ZIP_NAME = "godot-ai-plugin-vnext.zip"
SMOKE_ZIP_RES_PATH = f"res://{SMOKE_DIR}/{SMOKE_ZIP_NAME}"
SMOKE_DOWNLOAD_URL = "smoke://local-prestaged"
DEFAULT_HTTP_PORT = 18000
DEFAULT_WS_PORT = 19500
SMOKE_TRIGGER_LOG = "MCP | [self-update-smoke vnext _exit_tree]"


class HarnessError(RuntimeError):
    pass


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Prepare a disposable Godot project for the interactive "
            "self-update smoke, optionally launch Godot, then verify state "
            "after the editor exits."
        )
    )
    parser.add_argument(
        "--project-dir",
        type=Path,
        default=DEFAULT_PROJECT_DIR,
        help=f"Disposable smoke project path (default: {DEFAULT_PROJECT_DIR})",
    )
    parser.add_argument(
        "--godot",
        default=os.environ.get(
            "GODOT_BIN",
            "/Applications/Godot_mono.app/Contents/MacOS/Godot"
            if platform.system() == "Darwin"
            else "godot",
        ),
        help="Godot executable to launch (default: GODOT_BIN or platform default)",
    )
    parser.add_argument(
        "--base-version",
        default="",
        help="Override the fixture v(N) plugin.cfg version",
    )
    parser.add_argument(
        "--base-from-release-tag",
        default="",
        help=(
            "Source v(N) from the GitHub release with this tag (e.g. 'v2.4.0') "
            "instead of current source. Validates the actual upgrade transition "
            "users experience between two real releases."
        ),
    )
    parser.add_argument(
        "--base-from-zip",
        type=Path,
        default=None,
        help=(
            "Source v(N) from this local godot-ai-plugin.zip (offline alternative "
            "to --base-from-release-tag). Path must point at a release-shaped zip."
        ),
    )
    parser.add_argument(
        "--next-version",
        default="",
        help="Override the synthetic v(N+1) plugin.cfg version",
    )
    parser.add_argument(
        "--server-version",
        default="",
        help=(
            "Published godot-ai Python package version for the base smoke fixture "
            "(default: base version)."
        ),
    )
    parser.add_argument(
        "--next-server-version",
        default="",
        help=(
            "Published godot-ai Python package version for the vNext fixture "
            "(default: --server-version). Use with --base-version/--next-version "
            "to smoke old-to-current server replacement."
        ),
    )
    parser.add_argument(
        "--http-port",
        type=int,
        default=DEFAULT_HTTP_PORT,
        help=f"Fixture-only MCP HTTP port (default: {DEFAULT_HTTP_PORT})",
    )
    parser.add_argument(
        "--ws-port",
        type=int,
        default=DEFAULT_WS_PORT,
        help=f"Fixture-only MCP websocket port (default: {DEFAULT_WS_PORT})",
    )
    parser.add_argument(
        "--no-launch",
        action="store_true",
        help="Prepare and print instructions without launching Godot",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="Replace project-dir even if it is not a prior smoke fixture",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        project_dir = args.project_dir.expanduser().resolve()
        if args.base_from_release_tag and args.base_from_zip:
            raise HarnessError("Pass at most one of --base-from-release-tag / --base-from-zip")
        base_zip: Path | None = None
        if args.base_from_zip:
            base_zip = args.base_from_zip.expanduser().resolve()
            if not base_zip.is_file():
                raise HarnessError(f"--base-from-zip not found: {base_zip}")
        elif args.base_from_release_tag:
            base_zip = ensure_release_zip_cached(args.base_from_release_tag)
        default_base_version = (
            release_tag_to_version(args.base_from_release_tag)
            if args.base_from_release_tag
            else read_plugin_version(PLUGIN_ROOT / "plugin.cfg")
        )
        base_version = args.base_version or default_base_version
        next_version = args.next_version or read_plugin_version(PLUGIN_ROOT / "plugin.cfg")
        if next_version == base_version:
            next_version = bump_patch_version(base_version)
        server_version = args.server_version or base_version
        next_server_version = args.next_server_version or server_version
        baseline = diagnostic_reports_snapshot()

        prepare_project(
            project_dir=project_dir,
            base_version=base_version,
            next_version=next_version,
            server_version=server_version,
            next_server_version=next_server_version,
            http_port=args.http_port,
            ws_port=args.ws_port,
            force=args.force,
            base_zip=base_zip,
        )
        write_baseline(project_dir, baseline)
        print_instructions(
            project_dir,
            args.godot,
            base_version,
            next_version,
            server_version,
            next_server_version,
            baseline,
        )

        if args.no_launch:
            return 0

        return launch_and_verify(project_dir, args.godot, next_version, baseline)
    except HarnessError as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        return 1
    except Exception as exc:
        print(f"FAIL: Unexpected {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1


def read_plugin_version(plugin_cfg: Path) -> str:
    match = re.search(r'^version="([^"]+)"$', plugin_cfg.read_text(), re.MULTILINE)
    if not match:
        raise HarnessError(f"Could not read plugin version from {plugin_cfg}")
    return match.group(1)


def release_tag_to_version(tag: str) -> str:
    return tag[1:] if tag.startswith("v") else tag


def ensure_release_zip_cached(tag: str) -> Path:
    cache_dir = Path(tempfile.gettempdir()) / "godot-ai-self-update-smoke-cache"
    cache_dir.mkdir(parents=True, exist_ok=True)
    zip_path = cache_dir / f"godot-ai-plugin-{tag}.zip"
    if zip_path.is_file() and zip_path.stat().st_size > 0:
        return zip_path
    url = f"https://github.com/hi-godot/godot-ai/releases/download/{tag}/godot-ai-plugin.zip"
    print(f"Fetching release zip: {url}")
    try:
        with urllib.request.urlopen(url) as resp:
            data = resp.read()
    except Exception as exc:
        raise HarnessError(f"Failed to download {url}: {exc}") from exc
    zip_path.write_bytes(data)
    return zip_path


def extract_release_zip_to_addon(zip_path: Path, addon_dir: Path) -> None:
    if addon_dir.exists():
        shutil.rmtree(addon_dir)
    addon_dir.mkdir(parents=True)
    with zipfile.ZipFile(zip_path) as zf:
        for member in zf.infolist():
            name = member.filename
            if not name.startswith("addons/godot_ai/"):
                continue
            rel = name[len("addons/godot_ai/") :]
            if not rel:
                continue
            dest = addon_dir / rel
            if member.is_dir():
                dest.mkdir(parents=True, exist_ok=True)
                continue
            dest.parent.mkdir(parents=True, exist_ok=True)
            with zf.open(member) as src, open(dest, "wb") as out:
                shutil.copyfileobj(src, out)
    if not (addon_dir / "plugin.cfg").is_file():
        raise HarnessError(f"Extracted zip {zip_path} did not contain addons/godot_ai/plugin.cfg")


def bump_patch_version(version: str) -> str:
    match = re.match(r"^(\d+)\.(\d+)\.(\d+)", version)
    if not match:
        return f"{version}-self-update-smoke"
    major, minor, patch = (int(part) for part in match.groups())
    return f"{major}.{minor}.{patch + 1}"


def prepare_project(
    *,
    project_dir: Path,
    base_version: str,
    next_version: str,
    server_version: str,
    next_server_version: str,
    http_port: int,
    ws_port: int,
    force: bool,
    base_zip: Path | None = None,
) -> None:
    ensure_safe_project_dir(project_dir, force)
    project_dir.mkdir(parents=True)
    marker_dir = project_dir / ".godot-ai-self-update-smoke"
    marker_dir.mkdir()
    (marker_dir / "marker.txt").write_text(
        "Generated by script/local-self-update-smoke. Safe to delete.\n",
        encoding="utf-8",
    )

    write_project_files(project_dir)

    base_addon = project_dir / "addons" / "godot_ai"
    if base_zip is not None:
        extract_release_zip_to_addon(base_zip, base_addon)
    else:
        shutil.copytree(PLUGIN_ROOT, base_addon, ignore=copy_ignore)
    patch_fixture_plugin(
        base_addon,
        version=base_version,
        server_version=server_version,
        http_port=http_port,
        ws_port=ws_port,
        force_local_update=True,
        next_version=next_version,
    )

    vnext_root = project_dir / ".godot-ai-self-update-smoke" / "vnext" / "addons" / "godot_ai"
    shutil.copytree(PLUGIN_ROOT, vnext_root, ignore=copy_ignore)
    patch_fixture_plugin(
        vnext_root,
        version=next_version,
        server_version=next_server_version,
        http_port=http_port,
        ws_port=ws_port,
        force_local_update=False,
        next_version=next_version,
    )
    patch_vnext_hot_reload_trigger(vnext_root / "mcp_dock.gd")
    patch_vnext_topology_change(vnext_root)

    zip_path = project_dir / SMOKE_DIR / SMOKE_ZIP_NAME
    zip_path.parent.mkdir()
    create_plugin_zip(vnext_root, zip_path)


def ensure_safe_project_dir(project_dir: Path, force: bool) -> None:
    if not project_dir.exists():
        return
    if project_dir.is_symlink():
        raise HarnessError(f"Refusing to remove symlinked project dir: {project_dir}")
    marker = project_dir / ".godot-ai-self-update-smoke" / "marker.txt"
    generated = marker.exists() and is_generated_smoke_project(project_dir)
    if marker.exists() and not generated and not force:
        raise HarnessError(
            f"{project_dir} has a smoke marker but does not look generated. "
            "Pass --project-dir elsewhere or --force."
        )
    if force or generated:
        shutil.rmtree(project_dir)
        return
    raise HarnessError(
        f"{project_dir} already exists and is not marked as a smoke fixture. "
        "Pass --project-dir elsewhere or --force."
    )


def is_generated_smoke_project(project_dir: Path) -> bool:
    project_file = project_dir / "project.godot"
    if not project_file.is_file():
        return False
    try:
        return project_file.read_text(encoding="utf-8").startswith(PROJECT_HEADER)
    except OSError:
        return False


def copy_ignore(_dir: str, names: list[str]) -> set[str]:
    return {
        name
        for name in names
        if name in {".godot", ".import"}
        or name.endswith(".tmp")
        or name.endswith(".bak")
        or name.startswith("godot_ai.backup.")
    }


def write_project_files(project_dir: Path) -> None:
    (project_dir / "project.godot").write_text(
        f"""{PROJECT_HEADER}
config_version=5

[application]
config/name="Godot AI Self Update Smoke"
run/main_scene="res://empty.tscn"
config/features=PackedStringArray("4.6")

[editor_plugins]
enabled=PackedStringArray("res://addons/godot_ai/plugin.cfg")
""",
        encoding="utf-8",
    )
    (project_dir / "empty.tscn").write_text(
        """[gd_scene format=3]

[node name="Main" type="Node3D"]
""",
        encoding="utf-8",
    )


def patch_fixture_plugin(
    addon_dir: Path,
    *,
    version: str,
    server_version: str,
    http_port: int,
    ws_port: int,
    force_local_update: bool,
    next_version: str,
) -> None:
    # `server_lifecycle.gd` and `utils/update_manager.gd` were both added in
    # v2.4.0 (extractions from plugin.gd and mcp_dock.gd respectively). The
    # `--base-from-release-tag` flag accepts older tags, but the harness's
    # patching here targets the v2.4.0+ extracted shape. For pre-v2.4.0
    # bases, raise a clear error pointing at the integration test that DOES
    # cover that path via a different mechanism (extract zip directly,
    # invoke runner.start() on the base's own runner, assert parse errors
    # fire as the documented historical constraint).
    _require_v240_plus_addon_shape(addon_dir, version)
    patch_plugin_cfg(addon_dir / "plugin.cfg", version)
    patch_port_isolation(
        addon_dir / "client_configurator.gd",
        addon_dir / "utils" / "settings.gd",
        http_port,
        ws_port,
    )
    patch_server_package_pin(addon_dir / "client_configurator.gd", server_version)
    patch_managed_record_isolation(addon_dir / "plugin.gd")
    patch_lifecycle_expected_server_version(
        addon_dir / "utils" / "server_lifecycle.gd", server_version
    )
    if force_local_update:
        # Update flow lives on McpUpdateManager; the dock keeps only
        # the visible banner UI.
        patch_local_update_banner(addon_dir / "utils" / "update_manager.gd", next_version)


def _require_v240_plus_addon_shape(addon_dir: Path, version: str) -> None:
    """Check that the addon tree has the v2.4.0+ extracted files this harness patches.

    `server_lifecycle.gd` and `update_manager.gd` were both added in v2.4.0;
    pre-v2.4.0 bases (notably v2.3.2) have the same logic embedded in
    `plugin.gd` and `mcp_dock.gd` respectively under different names. The
    harness's patches target the extracted shape and would need separate
    paths to cover the embedded shape, which doesn't justify the
    maintenance cost when the v2.3.2 -> current upgrade transition is
    already covered by `tests/integration/test_self_update_historical_constraint.py`.
    """
    missing = [
        rel
        for rel in ("utils/server_lifecycle.gd", "utils/update_manager.gd")
        if not (addon_dir / rel).is_file()
    ]
    if not missing:
        return
    raise HarnessError(
        f"Base addon at version {version!r} is missing {missing}, which the "
        "local smoke harness patches to override the update flow. Both files "
        "were added in v2.4.0; pre-v2.4.0 release zips (e.g. v2.3.2) don't "
        "have them. Use --base-from-release-tag v2.4.0 or later for the "
        "interactive smoke. The v2.3.2 -> current upgrade transition is "
        "covered by tests/integration/test_self_update_historical_constraint.py "
        "(run with RUN_HISTORICAL_SELF_UPDATE=1 and a cached v2.3.2 zip)."
    )


def patch_plugin_cfg(path: Path, version: str) -> None:
    text = path.read_text(encoding="utf-8")
    text, count = re.subn(
        r'^version="[^"]+"$',
        f'version="{version}"',
        text,
        count=1,
        flags=re.MULTILINE,
    )
    if count != 1:
        raise HarnessError(f"Could not replace plugin.cfg version in {path}")
    path.write_text(text, encoding="utf-8")


def patch_port_isolation(
    configurator_path: Path, settings_path: Path, http_port: int, ws_port: int
) -> None:
    # SETTING_HTTP_PORT and SETTING_EXCLUDED_DOMAINS live in utils/settings.gd.
    stext = settings_path.read_text(encoding="utf-8")
    stext = replace_once(
        settings_path,
        stext,
        'const SETTING_HTTP_PORT := "godot_ai/http_port"',
        'const SETTING_HTTP_PORT := "godot_ai_self_update_smoke/http_port"',
        "SETTING_HTTP_PORT",
    )
    stext = replace_once(
        settings_path,
        stext,
        'const SETTING_EXCLUDED_DOMAINS := "godot_ai/excluded_domains"',
        'const SETTING_EXCLUDED_DOMAINS := "godot_ai_self_update_smoke/excluded_domains"',
        "SETTING_EXCLUDED_DOMAINS",
    )
    settings_path.write_text(stext, encoding="utf-8")

    # DEFAULT_HTTP_PORT, DEFAULT_WS_PORT, SETTING_WS_PORT, and
    # _read_port_setting all remain in client_configurator.gd.
    text = configurator_path.read_text(encoding="utf-8")
    text = subn_once(
        configurator_path,
        text,
        r"const DEFAULT_HTTP_PORT := \d+",
        f"const DEFAULT_HTTP_PORT := {http_port}",
        "DEFAULT_HTTP_PORT",
    )
    text = subn_once(
        configurator_path,
        text,
        r"const DEFAULT_WS_PORT := \d+",
        f"const DEFAULT_WS_PORT := {ws_port}",
        "DEFAULT_WS_PORT",
    )
    text = replace_once(
        configurator_path,
        text,
        'const SETTING_WS_PORT := "godot_ai/ws_port"',
        'const SETTING_WS_PORT := "godot_ai_self_update_smoke/ws_port"',
        "SETTING_WS_PORT",
    )
    text = replace_function(
        text,
        "static func _read_port_setting(key: String, default_port: int) -> int:",
        """static func _read_port_setting(_key: String, default_port: int) -> int:
\t## The smoke fixture must not adopt or kill another editor's server through
\t## global EditorSettings. Keep its ports deterministic and fixture-local.
\treturn default_port""",
    )
    configurator_path.write_text(text, encoding="utf-8")


def patch_server_package_pin(path: Path, server_version: str) -> None:
    text = path.read_text(encoding="utf-8")
    text = replace_once(
        path,
        text,
        'const SERVER_NAME := "godot-ai"\n',
        (
            'const SERVER_NAME := "godot-ai"\n'
            f'const SELF_UPDATE_SMOKE_SERVER_VERSION := "{server_version}"\n'
        ),
        "SELF_UPDATE_SMOKE_SERVER_VERSION",
    )
    text = replace_once(
        path,
        text,
        "var version := get_plugin_version()",
        "var version := SELF_UPDATE_SMOKE_SERVER_VERSION",
        "uvx server version pin",
    )
    path.write_text(text, encoding="utf-8")


def patch_managed_record_isolation(path: Path) -> None:
    text = path.read_text(encoding="utf-8")
    text = replace_once(
        path,
        text,
        'const MANAGED_SERVER_PID_SETTING := "godot_ai/managed_server_pid"',
        'const MANAGED_SERVER_PID_SETTING := "godot_ai_self_update_smoke/managed_server_pid"',
        "MANAGED_SERVER_PID_SETTING",
    )
    text = replace_once(
        path,
        text,
        'const MANAGED_SERVER_VERSION_SETTING := "godot_ai/managed_server_version"',
        (
            "const MANAGED_SERVER_VERSION_SETTING := "
            '"godot_ai_self_update_smoke/managed_server_version"'
        ),
        "MANAGED_SERVER_VERSION_SETTING",
    )
    path.write_text(text, encoding="utf-8")


def patch_lifecycle_expected_server_version(path: Path, server_version: str) -> None:
    text = path.read_text(encoding="utf-8")
    marker = 'const ClientConfigurator := preload("res://addons/godot_ai/client_configurator.gd")'
    line_start = text.find(marker)
    if line_start < 0:
        raise HarnessError(f"Could not find ClientConfigurator preload in {path}")
    line_end = text.find("\n", line_start)
    if line_end < 0:
        raise HarnessError(f"Could not find ClientConfigurator preload line end in {path}")
    text = (
        text[: line_end + 1]
        + f'const SELF_UPDATE_SMOKE_EXPECTED_SERVER_VERSION := "{server_version}"\n'
        + text[line_end + 1 :]
    )
    text = replace_once(
        path,
        text,
        (
            "func _expected_server_version() -> String:\n"
            "\treturn ClientConfigurator.get_plugin_version()"
        ),
        (
            "func _expected_server_version() -> String:\n"
            "\treturn SELF_UPDATE_SMOKE_EXPECTED_SERVER_VERSION"
        ),
        "_expected_server_version",
    )
    path.write_text(text, encoding="utf-8")


def patch_local_update_banner(path: Path, next_version: str) -> None:
    """Override the McpUpdateManager update flow to use a local zip fixture.

    Patches `addons/godot_ai/utils/update_manager.gd`:

      - Inject smoke constants alongside `UPDATE_TEMP_ZIP`.
      - Replace `check_for_updates()` so it skips the GitHub HTTP fetch
        and directly emits `update_check_completed` with the smoke
        payload (the dock listens to that signal and paints the banner).
      - Replace `start_install()` with the smoke branch — copies the
        fixture zip into `UPDATE_TEMP_ZIP` and calls `_install_zip`,
        bypassing the real download.
    """
    text = path.read_text(encoding="utf-8")
    text = replace_once(
        path,
        text,
        'const UPDATE_TEMP_ZIP := "user://godot_ai_update/update.zip"\n',
        (
            'const UPDATE_TEMP_ZIP := "user://godot_ai_update/update.zip"\n'
            f'const SELF_UPDATE_SMOKE_ZIP := "{SMOKE_ZIP_RES_PATH}"\n'
            f'const SELF_UPDATE_SMOKE_DOWNLOAD_URL := "{SMOKE_DOWNLOAD_URL}"\n'
            f'const SELF_UPDATE_SMOKE_NEXT_VERSION := "{next_version}"\n'
        ),
        "SELF_UPDATE_SMOKE constants",
    )
    text = replace_function(
        text,
        "func check_for_updates() -> void:",
        """func check_for_updates() -> void:
\t_latest_download_url = SELF_UPDATE_SMOKE_DOWNLOAD_URL
\tvar smoke_zip_path := ProjectSettings.globalize_path(SELF_UPDATE_SMOKE_ZIP)
\tprint("MCP | self-update smoke: using local zip %s" % smoke_zip_path)
\tupdate_check_completed.emit({
\t\t"has_update": true,
\t\t"version": SELF_UPDATE_SMOKE_NEXT_VERSION,
\t\t"forced": false,
\t\t"label_text": "Smoke update available: v%s" % SELF_UPDATE_SMOKE_NEXT_VERSION,
\t\t"download_url": SELF_UPDATE_SMOKE_DOWNLOAD_URL,
\t})""",
    )
    text = replace_function(
        text,
        "func start_install() -> void:",
        """func start_install() -> void:
\tif _latest_download_url != SELF_UPDATE_SMOKE_DOWNLOAD_URL:
\t\tOS.shell_open(RELEASES_PAGE)
\t\treturn
\tinstall_state_changed.emit({
\t\t"phase": "smoke_installing",
\t\t"button_text": "Installing smoke update...",
\t\t"button_disabled": true,
\t})
\tvar src := ProjectSettings.globalize_path(SELF_UPDATE_SMOKE_ZIP)
\tvar dst_dir := ProjectSettings.globalize_path(UPDATE_TEMP_DIR)
\tvar dst := ProjectSettings.globalize_path(UPDATE_TEMP_ZIP)
\tvar path_record_path := ProjectSettings.globalize_path(
\t\t"res://.godot-ai-self-update-smoke/user-update-path.txt"
\t)
\tvar path_record := FileAccess.open(path_record_path, FileAccess.WRITE)
\tif path_record != null:
\t\tpath_record.store_string(dst_dir)
\t\tpath_record.close()
\tDirAccess.make_dir_recursive_absolute(dst_dir)
\tvar bytes := FileAccess.get_file_as_bytes(src)
\tif bytes.is_empty() and FileAccess.get_open_error() != OK:
\t\tvar read_error := FileAccess.get_open_error()
\t\tprint("MCP | self-update smoke: failed to read local zip %s (error %d)" % [
\t\t\tsrc,
\t\t\tread_error,
\t\t])
\t\tinstall_state_changed.emit({
\t\t\t"phase": "smoke_zip_missing",
\t\t\t"button_text": "Smoke zip missing",
\t\t\t"button_disabled": false,
\t\t})
\t\treturn
\tvar f := FileAccess.open(dst, FileAccess.WRITE)
\tif f == null:
\t\tvar open_error := FileAccess.get_open_error()
\t\tprint("MCP | self-update smoke: failed to stage zip %s (error %d)" % [
\t\t\tdst,
\t\t\topen_error,
\t\t])
\t\tinstall_state_changed.emit({
\t\t\t"phase": "smoke_stage_failed",
\t\t\t"button_text": "Stage failed",
\t\t\t"button_disabled": false,
\t\t})
\t\treturn
\tf.store_buffer(bytes)
\tvar write_error := f.get_error()
\tf.close()
\tif write_error != OK:
\t\tprint("MCP | self-update smoke: write error %d for %s" % [write_error, dst])
\t\tinstall_state_changed.emit({
\t\t\t"phase": "smoke_stage_failed",
\t\t\t"button_text": "Stage failed",
\t\t\t"button_disabled": false,
\t\t})
\t\treturn
\tprint("MCP | self-update smoke: staged local zip %s" % dst)
\t_install_zip.call_deferred()""",
    )
    path.write_text(text, encoding="utf-8")


def patch_vnext_hot_reload_trigger(path: Path) -> None:
    """Inject a typed Dict + Array onto McpDock so the vNext hot-reload
    actually exercises the field-storage crash class (issue #245).

    Anchored to a stable plain field (`_last_connected`) — anchoring to
    a recently-moved field would silently miss the regression rather
    than failing loudly. (`_last_log_count` moved into the LogViewer
    subpanel as part of audit-v2 #360 and is no longer on McpDock.)
    """
    text = path.read_text(encoding="utf-8")
    field_marker = "var _last_connected := false\n"
    if field_marker not in text:
        raise HarnessError(f"Could not find self-update field marker in {path}")
    text = text.replace(
        field_marker,
        (
            field_marker + 'var _self_update_smoke_trigger: Dictionary = {"armed": true}\n'
            'var _self_update_smoke_array_trigger: Array[String] = ["armed"]\n'
        ),
        1,
    )
    exit_marker = "func _exit_tree() -> void:\n"
    if exit_marker not in text:
        raise HarnessError(f"Could not find _exit_tree in {path}")
    text = text.replace(
        exit_marker,
        (
            exit_marker
            + '\tprint("MCP | [self-update-smoke vnext _exit_tree] keys=%d values=%d" % '
            "[_self_update_smoke_trigger.keys().size(), _self_update_smoke_array_trigger.size()])\n"
        ),
        1,
    )
    path.write_text(text, encoding="utf-8")


def patch_vnext_topology_change(addon_dir: Path) -> None:
    utils_dir = addon_dir / "utils"
    base_path = utils_dir / "self_update_smoke_base.gd"
    child_path = utils_dir / "self_update_smoke_child.gd"
    base_path.write_text(
        """@tool
class_name McpSelfUpdateSmokeBase
extends RefCounted


func marker() -> String:
\treturn "base"
""",
        encoding="utf-8",
    )
    child_path.write_text(
        """@tool
class_name McpSelfUpdateSmokeChild
extends McpSelfUpdateSmokeBase


func marker() -> String:
\treturn "child:" + super.marker()
""",
        encoding="utf-8",
    )

    # Anchor the topology-change preload onto a stable line in the dock —
    # `UpdateManagerScript`'s `const … := preload(...)` is one of the
    # first lines and survives unrelated dock refactors.
    dock_path = addon_dir / "mcp_dock.gd"
    text = dock_path.read_text(encoding="utf-8")
    marker = (
        'const UpdateManagerScript := preload("res://addons/godot_ai/utils/update_manager.gd")\n'
    )
    text = replace_once(
        dock_path,
        text,
        marker,
        (
            marker
            + "const SelfUpdateSmokeChild := preload("
            + '"res://addons/godot_ai/utils/self_update_smoke_child.gd")\n'
        ),
        "self-update smoke topology preload",
    )
    dock_path.write_text(text, encoding="utf-8")


def replace_function(text: str, signature: str, replacement: str) -> str:
    lines = text.splitlines(keepends=True)
    start_line = -1
    for i, line in enumerate(lines):
        if line.startswith(signature):
            start_line = i
            break
    if start_line == -1:
        raise HarnessError(f"Could not find function signature: {signature}")

    next_func_line = -1
    for i in range(start_line + 1, len(lines)):
        if lines[i].startswith("func ") or lines[i].startswith("static func "):
            next_func_line = i
            break
    if next_func_line == -1:
        raise HarnessError(f"Could not find end of function: {signature}")

    end_line = next_func_line
    j = next_func_line - 1
    while j > start_line and (lines[j].strip() == "" or lines[j].startswith("##")):
        j -= 1
    if j < next_func_line - 1:
        end_line = j + 1

    before = "".join(lines[:start_line])
    after = "".join(lines[end_line:])
    return before + replacement + after


def replace_once(path: Path, text: str, old: str, new: str, label: str) -> str:
    if old not in text:
        raise HarnessError(f"Could not patch {label} in {path}")
    return text.replace(old, new, 1)


def subn_once(path: Path, text: str, pattern: str, replacement: str, label: str) -> str:
    text, count = re.subn(pattern, replacement, text, count=1)
    if count != 1:
        raise HarnessError(f"Could not patch {label} in {path}")
    return text


def create_plugin_zip(vnext_addon: Path, zip_path: Path) -> None:
    # Include zero-byte directory entries (the shape `zip -r` produces
    # without `-D`, also typical of AssetLib uploads and hand-built
    # archives). Current release.yml strips them via `zip -D`, but the
    # smoke must still exercise the directory-entry path: that's what
    # tripped the 2.2.x/2.3.0 install safety check, and the runner has
    # to keep handling it correctly for older or manually built zips.
    with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("addons/", "")
        zf.writestr("addons/godot_ai/", "")
        for path in sorted(vnext_addon.rglob("*")):
            rel = path.relative_to(vnext_addon)
            arcname = (Path("addons") / "godot_ai" / rel).as_posix()
            if path.is_dir():
                zf.writestr(arcname + "/", "")
            else:
                zf.write(path, arcname)


def diagnostic_reports_snapshot() -> set[Path]:
    if platform.system() != "Darwin":
        return set()
    reports_dir = Path.home() / "Library" / "Logs" / "DiagnosticReports"
    if not reports_dir.exists():
        return set()
    return {path for path in reports_dir.glob("Godot*.ips") if path.is_file()}


def write_baseline(project_dir: Path, baseline: set[Path]) -> None:
    baseline_file = project_dir / ".godot-ai-self-update-smoke" / "diagnostic-baseline.txt"
    baseline_file.write_text(
        "\n".join(str(path) for path in sorted(baseline)) + ("\n" if baseline else ""),
        encoding="utf-8",
    )


def print_instructions(
    project_dir: Path,
    godot_bin: str,
    base_version: str,
    next_version: str,
    server_version: str,
    next_server_version: str,
    baseline: set[Path],
) -> None:
    latest = max(baseline, key=lambda path: path.stat().st_mtime, default=None)
    print("")
    print("Self-update smoke fixture ready")
    print(f"  project:      {project_dir}")
    print(f"  base version: {base_version}")
    print(f"  next version: {next_version}")
    print(f"  base server:  godot-ai=={server_version}")
    print(f"  next server:  godot-ai=={next_server_version}")
    print(f"  local zip:    {project_dir / SMOKE_DIR / SMOKE_ZIP_NAME}")
    if latest is not None:
        print(f"  latest .ips:  {latest}")
    elif platform.system() == "Darwin":
        print("  latest .ips:  none found")
    else:
        print("  crash report: non-macOS; .ips check is skipped")
    print("")
    print("Manual step:")
    print("  1. In the Godot AI dock, click Update.")
    print("  2. Wait for 'MCP | update runner enabling new plugin' and 'MCP | plugin loaded'.")
    print("  3. Confirm the editor stays alive. Do not restart the editor to pass the smoke.")
    print("  4. Close Godot normally when done; this script then checks disk state and .ips files.")
    print("")
    print("Failure signals:")
    print("  - Godot exits or a new Godot*.ips appears.")
    print("  - The update temp dir remains after update.")
    print("  - 'MCP | [self-update-smoke vnext _exit_tree]' prints during the update window.")
    print("")
    print("Launch command:")
    print(f"  {godot_bin} --editor --path {project_dir}")
    print("")


def launch_and_verify(
    project_dir: Path, godot_bin: str, next_version: str, baseline: set[Path]
) -> int:
    godot_path = shutil.which(godot_bin) if os.sep not in godot_bin else godot_bin
    if godot_path is None or not Path(godot_path).exists():
        raise HarnessError(f"Godot executable not found: {godot_bin}")

    started = time.time()
    proc = subprocess.Popen(
        [str(godot_path), "--editor", "--path", str(project_dir)],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
    )
    lines: list[str] = []
    assert proc.stdout is not None
    for line in proc.stdout:
        print(line, end="")
        lines.append(line.rstrip("\n"))
    code = proc.wait()
    if code != 0:
        print(f"FAIL: Godot exited with code {code}")
        print_new_reports(baseline, started)
        return code or 1

    ok = verify_post_run(project_dir, next_version, baseline, started, lines)
    return 0 if ok else 1


def verify_post_run(
    project_dir: Path,
    next_version: str,
    baseline: set[Path],
    started: float,
    lines: list[str],
) -> bool:
    ok = True
    installed_version = read_plugin_version(project_dir / "addons" / "godot_ai" / "plugin.cfg")
    if installed_version == next_version:
        print(f"PASS: plugin version advanced to {installed_version}")
    else:
        print(f"FAIL: plugin version is {installed_version}, expected {next_version}")
        ok = False

    user_update_path_record = project_dir / ".godot-ai-self-update-smoke" / "user-update-path.txt"
    if user_update_path_record.exists():
        user_update_dir = Path(user_update_path_record.read_text(encoding="utf-8").strip())
        if user_update_dir.exists():
            print(f"FAIL: update temp dir still exists: {user_update_dir}")
            ok = False
        else:
            print(f"PASS: update temp dir was consumed: {user_update_dir}")
    else:
        print("FAIL: smoke fixture did not record the update temp dir path")
        ok = False

    if vnext_exit_tree_during_update(lines):
        print(f"FAIL: {SMOKE_TRIGGER_LOG} printed before the new plugin finished loading")
        ok = False
    else:
        print("PASS: vNext _exit_tree trigger did not run during the update window")

    if smoke_adopted_existing_server_before_update(lines):
        print(
            "FAIL: smoke fixture adopted an existing server before update; "
            "clear the smoke ports first"
        )
        ok = False
    elif smoke_started_own_server_before_update(lines):
        print("PASS: smoke fixture started its own server before update")
    else:
        print("FAIL: smoke fixture did not start its own server before update")
        ok = False

    if smoke_stopped_server_during_update(lines):
        print("PASS: self-update stopped the managed smoke server")
    else:
        print("FAIL: self-update did not stop a managed smoke server")
        ok = False

    if smoke_reported_server_version_mismatch(lines):
        print("FAIL: smoke fixture reported a server/plugin version mismatch")
        ok = False
    else:
        print("PASS: no server/plugin version mismatch reported")

    new_reports = new_diagnostic_reports(baseline, started)
    if new_reports:
        print("FAIL: new Godot crash reports:")
        for report in new_reports:
            print(f"  {report}")
        ok = False
    elif platform.system() == "Darwin":
        print("PASS: no new Godot .ips crash report detected")
    else:
        print("SKIP: non-macOS .ips crash report check")

    print("Check the Godot console for:")
    print("  PASS signal: update runner enabling new plugin, then plugin loaded")
    return ok


def vnext_exit_tree_during_update(lines: list[str]) -> bool:
    staged = False
    enabling_new_plugin = False
    new_plugin_loaded = False
    for line in lines:
        if "MCP | self-update smoke: staged local zip" in line:
            staged = True
        if staged and "MCP | update runner enabling new plugin" in line:
            enabling_new_plugin = True
        if enabling_new_plugin and "MCP | plugin loaded" in line:
            new_plugin_loaded = True
        if staged and SMOKE_TRIGGER_LOG in line and not new_plugin_loaded:
            return True
    return False


def smoke_started_own_server_before_update(lines: list[str]) -> bool:
    return any("MCP | started server" in line for line in smoke_lines_before_update(lines))


def smoke_adopted_existing_server_before_update(lines: list[str]) -> bool:
    external_signals = (
        "MCP | adopted external server",
        "MCP | foreign server already running",
        "foreign server already running",
    )
    return any(
        any(signal in line for signal in external_signals)
        for line in smoke_lines_before_update(lines)
    )


def smoke_stopped_server_during_update(lines: list[str]) -> bool:
    staged = False
    for line in lines:
        if "MCP | self-update smoke: staged local zip" in line:
            staged = True
            continue
        if not staged:
            continue
        if "MCP | update runner enabling new plugin" in line:
            return False
        if "MCP | stopped server" in line:
            return True
    return False


def smoke_reported_server_version_mismatch(lines: list[str]) -> bool:
    return any(
        "is occupied by godot-ai server v" in line and "plugin expects v" in line for line in lines
    )


def smoke_lines_before_update(lines: list[str]) -> list[str]:
    before: list[str] = []
    for line in lines:
        if "MCP | self-update smoke: staged local zip" in line:
            break
        before.append(line)
    return before


def print_new_reports(baseline: set[Path], started: float) -> None:
    reports = new_diagnostic_reports(baseline, started)
    if reports:
        print("New Godot crash reports:")
        for report in reports:
            print(f"  {report}")


def new_diagnostic_reports(baseline: set[Path], started: float) -> list[Path]:
    current = diagnostic_reports_snapshot()
    return sorted(
        (
            path
            for path in current - baseline
            if path.exists() and path.stat().st_mtime >= started - 2
        ),
        key=lambda path: path.stat().st_mtime,
    )


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