"""The newly imported escape hatch from an already-running historical updater.

This module must remain stdlib-only: the parent still has the old interpreter,
modules and native extensions. Only the child imports the updated application.
"""
from __future__ import annotations

import dataclasses
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
from typing import Any, NoReturn


_result: int | None = None

# The historical updater entrypoints. `_historical_context` reads capture data
# from these frames; `tools.lazy_deps.install_specs` hands off only inside them.
# Keep in sync with scripts/audit-old-updater-imports.py UPDATE_ENTRYPOINTS: the
# audit list is wider on purpose (helpers run *under* these frames and are found
# by the frame walk).
_HISTORICAL_UPDATE_FRAMES = (
    ("hermes_cli.update_cmd", "hermes_cli.main"),
    ("_cmd_update_impl", "cmd_update"),
)


def _is_historical_update_frame(frame: Any) -> bool:
    modules, names = _HISTORICAL_UPDATE_FRAMES
    return (frame.f_globals.get("__name__") in modules
            and frame.f_code.co_name in names)


def in_historical_update() -> bool:
    """True when an ancestor frame is a historical updater entrypoint.

    The current tree's updater sets ``_hermes_current_updater_frame`` in its
    entrypoints; no historical on-disk version contains that local, so its
    presence proves the caller is current code that must keep running rather
    than hand control to a takeover child.
    """
    frame = sys._getframe(1)
    try:
        while frame is not None:
            if (_is_historical_update_frame(frame)
                    and "_hermes_current_updater_frame" not in frame.f_code.co_varnames):
                return True
            frame = frame.f_back
    finally:
        del frame
    return False


def _historical_context() -> tuple[dict, list[dict], Any]:
    """Carry data already held by old frames, without importing their modules.

    Older callers did not pass a continuation object. Read only the historical
    updater's known local names; do not serialize frames or arbitrary globals.
    """
    names = ("had_desktop_app_before_update", "pre_update_snapshot_id",
             "pre_update_version", "gateway_mode", "assume_yes", "_pre_update_plan")
    found = {}
    resumes = []
    restart_update = None
    frame = sys._getframe(1)
    try:
        while frame is not None:
            # Capture hooks can be imported BEFORE the old updater has pulled.
            # A declared-but-unassigned version in its innermost known frame
            # proves early entry; a None value is still a post-capture value.
            if (restart_update is None and _is_historical_update_frame(frame)
                    and "pre_update_version" in frame.f_code.co_varnames):
                restart_update = "pre_update_version" not in frame.f_locals
            for name in names:
                if name not in found and name in frame.f_locals:
                    found[name] = frame.f_locals[name]
            token = frame.f_locals.get("_windows_gateway_resume")
            if isinstance(token, dict) and not any(token is item for item in resumes):
                resumes.append(token)
            frame = frame.f_back
    finally:
        del frame
    receipt_module = sys.modules.get("hermes_cli.update_receipt")
    receipt_slot = vars(receipt_module).get("_current") if receipt_module else None
    get_current = getattr(receipt_slot, "get", None)
    current = get_current() if get_current is not None else receipt_slot
    receipt = getattr(current, "data", None)
    plan = found.get("_pre_update_plan")
    if dataclasses.is_dataclass(plan) and not isinstance(plan, type):
        plan = dataclasses.asdict(plan)
    elif not isinstance(plan, dict):
        plan = receipt.get("plan") if isinstance(receipt, dict) else None
    return {
        "restart_update": restart_update is True,
        # Some old updaters reach this hook before checking Desktop at all.
        "desktop": found.get("had_desktop_app_before_update"),
        "pre_update_snapshot_id": found.get("pre_update_snapshot_id"),
        "pre_update_version": found.get("pre_update_version"),
        "gateway_mode": bool(found.get("gateway_mode", "--gateway" in sys.argv)),
        "assume_yes": bool(found.get("assume_yes", "--yes" in sys.argv)),
        "windows_resume": resumes[0] if resumes else None,
        "plan": plan,
        "receipt": receipt if isinstance(receipt, dict) else None,
    }, resumes, receipt_slot


def _run_child(request: dict) -> tuple[int, dict]:
    """One JSON exchange; the old process never imports the updated graph."""
    root = Path(__file__).resolve().parents[1]
    home = os.environ.get("HERMES_HOME")
    constants = sys.modules.get("hermes_constants")
    override = vars(constants).get("_HERMES_HOME_OVERRIDE") if constants else None
    if override is not None:
        value = override.get()
        if isinstance(value, (str, Path)) and value:
            home = str(value)
    request.update(root=str(root), home=home, argv=list(sys.argv))
    env = {key: value for key, value in os.environ.items()
           if not key.startswith(("PYTHON", "UV_")) and key != "VIRTUAL_ENV"}
    if home:
        env["HERMES_HOME"] = home
    env.setdefault("HERMES_UPDATE_HANDOFF_PID", str(os.getpid()))
    # Everything the user saw so far came from the OLD updater; say so before the
    # new one (package manager) takes over, so logs show where the switch happened.
    # ASCII only, like every line printed before the child starts: a Windows pipe encodes this
    # process's stdout in the ANSI code page, and a non-cp1252 character raised here, before the
    # takeover ever ran (review S1).
    print("-> Handing off to the new updater (package manager) for the rest of this update...", flush=True)
    with tempfile.TemporaryDirectory(prefix="hermes-update-takeover-") as directory:
        context = Path(directory) / "request.json"
        result = Path(directory) / "result.json"
        context.write_text(json.dumps(request), encoding="utf-8")
        code = _run_in_custody(
            [sys.executable, "-I", "-S", "-B", "-X", "utf8", str(root / "hermes_cli/_update_takeover.py"),
             str(context), str(result)], root, cwd=root, env=env,
        )
        completed = json.loads(result.read_text(encoding="utf-8-sig")) if result.is_file() else {}
        if not isinstance(completed, dict):
            raise ValueError("invalid update takeover acknowledgement")
        return code if code >= 0 else 1, completed


def _run_in_custody(argv: list, root: Path, **kwargs) -> int:
    """Run the takeover (it syncs and builds the checkout) in this updater's checkout custody,
    from the update_lock module ALREADY loaded here (nothing new is imported). Windows: created
    suspended, bound to this updater's kill-on-close job, then resumed; a refused bind still runs
    (the update committed) and the takeover joins the lock holding its own lease. Releases
    without these helpers hold no checkout lock, and the takeover acquires it itself."""
    lock = sys.modules.get("hermes_cli.update_lock")
    bind, resume = (getattr(lock, name, None) for name in ("bind_child_to_update_tree", "resume_suspended_child"))
    if os.name != "nt" or bind is None or resume is None or getattr(lock, "_HELD", None) is None:
        # Unbounded for the same reason as the Windows wait below (review L7): the takeover is
        # the rest of a committed update (sync + build), and it holds the checkout lock fd, so the
        # checkout stays locked for as long as it runs, even past this updater's death.
        # health: allow HX006 -- unbounded by design: the update's own supervised child
        return subprocess.run(argv, **kwargs, **_checkout_custody(root)).returncode  # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
    proc = subprocess.Popen(argv, creationflags=lock.CREATE_SUSPENDED, **kwargs)
    try:
        refusal = bind(proc)
        if refusal is not None:
            print(f"  ! The update's job would not take the takeover ({refusal}); it runs outside the job, "
                  "holding its own checkout lease.", flush=True)
        resume(proc)
        # The takeover is the rest of a committed update (sync + build): a bound would abandon it
        # mid-write. It dies with this updater (kill-on-close job, or the except path below).
        # health: allow HX006 -- unbounded by design: the update's own supervised child
        return proc.wait()
    except BaseException:
        proc.kill()
        raise


def _checkout_custody(root: Path) -> dict:
    """The takeover child keeps the checkout lock this (old) updater holds: its fd, read from the
    update_lock module ALREADY loaded in this process (nothing new is imported here). Releases
    without a checkout lock hold none, and the takeover acquires it itself."""
    held = getattr(sys.modules.get("hermes_cli.update_lock"), "checkout_lock_fds", None)
    try:
        fds = tuple(held(root)) if held is not None and os.name == "posix" else ()
    except Exception:  # health: allow BLE001 -- any older release's checkout_lock_fds; no fds = takeover locks itself
        fds = ()
    return {"pass_fds": fds} if fds else {}


def stop_for_relaunch(*, incomplete: bool = False) -> NoReturn:
    """Finish in a fresh child, then exit rather than resume an old fallback.

    The historical name is retained. A finally/atexit path can call another
    shim while unwinding; it must receive the same result, not start again.
    """
    if incomplete:
        # A newly retired completion hook has no complete captured worklist.
        # It must not start another update or invent a successful receipt. It names its stop on
        # the receipt the loaded update_receipt keeps open (an older one lacks the call: no-op).
        record = getattr(sys.modules.get("hermes_cli.update_receipt"), "record_stop_reason", None)
        if record is not None:
            record("old_version_handoff")
        print(
            "You're updating from an older version of Hermes Agent. "
            "To complete this update, run `hermes update` again.",
            file=sys.stderr,
        )
        raise SystemExit(1)
    global _result
    if _result is not None:
        raise SystemExit(_result)
    _result = 1
    try:
        request, resumes, receipt_slot = _historical_context()
        _result, completed = _run_child(request)
        if completed.get("resume_handled"):
            for token in resumes:
                token["resume_needed"] = False
        if completed.get("receipt_handled"):
            if hasattr(receipt_slot, "set"):
                receipt_slot.set(None)
            elif receipt_slot is not None:
                vars(sys.modules["hermes_cli.update_receipt"])["_current"] = None
    except (OSError, ValueError, TypeError) as exc:
        _result = 1
        print(f"Update takeover failed: {exc}", file=sys.stderr, flush=True)
    raise SystemExit(_result)


def relaunch_stopped_serves(token: dict) -> None:
    """A historical atexit token is separate work, not another whole update."""
    if not token.get("pending"):
        return
    code = 1
    try:
        code, completed = _run_child({"stopped_serves": token})
        if completed.get("serves_handled") is True:
            token["pending"] = False
    except (OSError, ValueError, TypeError) as exc:
        print(f"Stopped serve recovery failed: {exc}", file=sys.stderr, flush=True)
    if code or token.get("pending"):
        print("Stopped serve recovery did not complete; restart the affected "
              "serve/dashboard backends manually.", file=sys.stderr, flush=True)