"""Source-install launchers shared by setup, installers, and Windows repair.

Launchers execute store Python in isolated mode. They set the install's
default home and load hermes_bootstrap before the entry point. Bootstrap
reads the selected dependency generation at each start.

Windows uses distlib executables or a command-file fallback. POSIX uses
an executable shell wrapper. The standalone writer requires PM's store
interpreter before it publishes either command.
"""

from __future__ import annotations

import json
import os
import shlex
import sys
from pathlib import Path

if __name__ == "__main__":
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from pm.environments import store_root


def _inline_string_literal(value: str) -> str:
    """Keep inline Python source intact through Windows PowerShell's native argv quoting."""
    if os.name != "nt":
        return repr(value)
    escaped = value.encode("unicode_escape").decode("ascii")
    return "'" + escaped.replace("'", "\\x27").replace('"', "\\x22") + "'"


def runtime_command(repo_root: Path, args=(), *, module: str = "hermes_cli.main",
                    code: str | None = None, python: str | Path | None = None,
                    home: str | Path | None = None) -> list[str]:
    """An installation-bound command, safe to persist across dependency GC.

    Store Python owns the ABI; bootstrap selects and leases dependencies at
    child start. Nix and developer interpreters retain their external owner.
    No selected generation or ambient PYTHONPATH is captured in the command.
    """
    root = Path(repo_root).resolve()
    python = python or resolve_store_python(root) or Path(sys.executable)
    entry = f"exec({_inline_string_literal(code)})" if code is not None else (
        f"runpy.run_module({_inline_string_literal(module)}, run_name='__main__', alter_sys=True)")
    default_home = (_inline_string_literal(str(home)) if home is not None else
                    "str(__import__('hermes_constants').get_default_hermes_root())")
    bootstrap = (
        "import os, sys, runpy; "
        "os.environ.pop('PYTHONHOME', None); os.environ.pop('PYTHONPATH', None); "
        "os.environ.pop('VIRTUAL_ENV', None); "
        f"sys.path.insert(0, {_inline_string_literal(str(root))}); "
        f"os.environ['HERMES_HOME'] = os.environ.get('HERMES_HOME') or {default_home}; "
        "import hermes_bootstrap; "
        + entry
    )
    return [str(python), "-I", "-c", bootstrap, *args]


def print_runtime_command(repo_root: Path, argv: list[str]) -> None:
    """Machine boundary for consumers holding the exact published launcher."""
    import argparse

    parser = argparse.ArgumentParser(description="Resolve this installation's launch command.")
    parser.add_argument("--module", default="hermes_cli.main")
    parser.add_argument("args", nargs=argparse.REMAINDER)
    options = parser.parse_args(argv)
    args = options.args[1:] if options.args[:1] == ["--"] else options.args
    print(json.dumps(runtime_command(repo_root, args, module=options.module)))


def installation_command(repo_root: Path, args=(), *, module: str = "hermes_cli.main",
                         python: str | Path | None = None, home: str | Path | None = None) -> list[str]:
    """Persist a source launcher, never the versioned tool it currently uses.

    External/Nix installs retain their externally owned interpreter contract.
    Source installation/update publication refreshes the local launcher when
    the managed Python pin changes.
    """
    root = Path(repo_root)
    if resolve_store_python(root) is None:
        return runtime_command(root, args, module=module, python=python, home=home)
    prefix = [] if module == "hermes_cli.main" else ["--run-module", module]
    return [str(root / ".hermes" / "bin" / "hermes"), *prefix, *args]

#: Launcher command names — keep in lockstep with scripts/install.ps1
#: Publish-UserCommand and hermes_cli/_install_repair.py.
WINDOWS_BIN_LAUNCHERS = ("hermes", "hermes-acp")

#: command name -> (entry module, callable) — mirrors pyproject.toml
#: [project.scripts].
ENTRY_POINTS = {
    "hermes": ("hermes_cli.main", "main"),
    "hermes-acp": ("acp_adapter.entry", "main"),
}


def _is_windows() -> bool:
    return os.name == "nt"


def resolve_store_python(repo_root: Path) -> Path | None:
    """Read PM's committed Python tool, without adopting unrecorded bytes."""
    runtime = store_root(repo_root)
    rel = "python.exe" if _is_windows() else "bin/python3"

    facts = runtime / "facts.json"
    if facts.is_file():
        try:
            packages = json.loads(facts.read_text(encoding="utf-8-sig")).get(
                "packages", {}
            )
            entry = (packages.get("python") or {}).get("entry")
        except (OSError, ValueError):
            entry = None
        if entry:
            candidate = runtime / entry / rel
            if candidate.is_file():
                return candidate

    return None


def _load_script_maker():
    """distlib's ScriptMaker — standalone first, then pip's vendored copy."""
    try:
        from distlib.scripts import ScriptMaker

        return ScriptMaker
    except ImportError:
        pass
    try:
        from pip._vendor.distlib.scripts import ScriptMaker

        return ScriptMaker
    except ImportError:
        return None


def exe_is_venv_bound(exe: Path, venv_dir: Path | None) -> bool:
    """True when an existing launcher exe embeds the venv interpreter —
    i.e. it is a copied venv console-script trampoline from the pre-pm
    installer, which boots through ``venv\\Scripts\\python.exe`` and must be
    replaced. distlib launchers append the interpreter shebang after the zip
    payload; both encodings are scanned to be safe."""
    if venv_dir is None:
        return False
    needles = set()
    for interpreter in (
        venv_dir / "Scripts" / "hermes.exe",
        venv_dir / "Scripts" / "hermes-acp.exe",
        venv_dir / "Scripts" / "python.exe",
        venv_dir / "bin" / "python3",
        venv_dir / "bin" / "hermes",
        venv_dir / "bin" / "hermes-acp",
    ):
        for enc in ("utf-8", "utf-16-le"):
            try:
                needles.add(str(interpreter).encode(enc))
            except Exception:
                continue
    try:
        data = Path(exe).read_bytes()
    except OSError:
        return False
    return any(needle in data for needle in needles)


def _write_atomic(target: Path, write) -> Path | None:
    """Stage under a pid-suffixed name then os.replace, so a concurrent
    process start never sees a torn launcher."""
    staging = target.with_name(f"{target.name}.stage.{os.getpid()}")
    try:
        write(staging)
        os.replace(staging, target)
        return target
    except OSError:
        try:
            staging.unlink()
        except OSError:
            pass
        return None


def mint_launcher(
    name: str,
    repo_root: Path,
    out_dir: Path,
    python_exe: Path,
    site_packages: Path | None,
) -> Path | None:
    """Write a native launcher with the shared bootstrap script, or return None."""
    module, func = ENTRY_POINTS[name]
    out_dir = Path(out_dir)
    script = _launcher_script(name, Path(repo_root), site_packages)

    if not _is_windows():
        return _mint_shell_launcher(name, out_dir, python_exe, script)

    script_maker_cls = _load_script_maker()
    if script_maker_cls is not None:
        class _PathedScriptMaker(script_maker_cls):  # type: ignore[misc,valid-type]
            def _get_script_text(self, entry):
                return script

        import tempfile
        from zipfile import BadZipFile, ZipFile

        def make(directory: Path) -> list[str]:
            maker = _PathedScriptMaker(None, str(directory), add_launchers=True)
            maker.executable = str(python_exe)
            maker.variants = {""}
            maker.clobber = True
            return maker.make(f"{name} = {module}:{func}", {"interpreter_args": ["-I"]})

        # distlib's exe ZIP records the current time, so byte equality cannot
        # detect an unchanged launcher. Compare its loader + shebang and script;
        # leave an active executable alone when only the ZIP timestamp changed.
        with tempfile.TemporaryDirectory(prefix=f".{name}-", dir=out_dir) as staging:
            try:
                candidate = next((Path(p) for p in make(Path(staging))
                                  if Path(p).suffix.lower() == ".exe"), None)
            except Exception:
                candidate = None
            if candidate is not None:
                target = out_dir / candidate.name
                try:
                    with ZipFile(target) as old, ZipFile(candidate) as new:
                        if (old.namelist() == new.namelist() == ["__main__.py"]
                                and target.read_bytes()[:old.infolist()[0].header_offset]
                                == candidate.read_bytes()[:new.infolist()[0].header_offset]
                                and old.read("__main__.py") == new.read("__main__.py")):
                            return target
                except (OSError, BadZipFile, KeyError):
                    pass
                # For changed launchers, distlib's .deleteme replacement can
                # move an executable Windows still has mapped in memory.
                try:
                    written = make(out_dir)
                except Exception:
                    written = []
                for path in written:
                    if Path(path).suffix.lower() == ".exe":
                        return Path(path)
        # distlib ran but produced no exe (unexpected) — fall through to cmd.

    # A prepared app environment need not include distlib, even when the
    # installer used it to publish a native exe. Keep that executable if its
    # embedded interpreter and script still match; replacing it with a cmd
    # would require unlinking the currently running exe on Windows.
    existing = out_dir / f"{name}.exe"
    from zipfile import BadZipFile, ZipFile
    try:
        with ZipFile(existing) as archive:
            if archive.namelist() == ["__main__.py"]:
                prefix = existing.read_bytes()[:archive.infolist()[0].header_offset]
                shebangs = (f"#!{python_exe} -I\n".encode("utf-8"),
                            f'#!"{python_exe}" -I\n'.encode("utf-8"))
                if (any(prefix.endswith(shebang) for shebang in shebangs)
                        and archive.read("__main__.py") == script.encode("utf-8")):
                    return existing
    except (OSError, BadZipFile, KeyError):
        pass

    # The script is data to Python, not interpolated shell source.
    import base64
    encoded = base64.b64encode(script.encode("utf-8")).decode("ascii")
    code = f"import base64; exec(base64.b64decode('{encoded}'))"
    body = (
        "@echo off\r\n"
        f'"{python_exe}" -I -c "{code}" %*\r\n'
    )
    return _write_atomic(out_dir / f"{name}.cmd", lambda p: p.write_text(body, encoding="utf-8"))


def _launcher_script(name: str, repo_root: Path, dependencies: Path | None) -> str:
    module, func = ENTRY_POINTS[name]
    # Profile boot repairs shared launchers: their default must stay at the
    # install's dependency root, not whichever profile triggered publication.
    return (
        "import os, re, sys\n"
        "os.environ.pop('PYTHONHOME', None)\n"
        "os.environ.pop('PYTHONPATH', None)\n"
        f"sys.path.insert(0, {str(repo_root.resolve())!r})\n"
        "if sys.argv[1:2] == ['--print-runtime-command']: sys.dont_write_bytecode = True\n"
        "from hermes_constants import get_default_hermes_root\n"
        "os.environ['HERMES_HOME'] = os.environ.get('HERMES_HOME') or str(get_default_hermes_root())\n"
        "if sys.argv[1:2] == ['--print-runtime-command']:\n"
        "    from pathlib import Path\n"
        "    from hermes_cli._launchers import print_runtime_command\n"
        f"    print_runtime_command(Path({str(repo_root.resolve())!r}), sys.argv[2:])\n"
        "    sys.exit(0)\n"
        "import hermes_bootstrap\n"
        "if sys.argv[1:2] == ['--run-module']:\n"
        "    import runpy\n"
        "    if len(sys.argv) < 3: sys.exit('hermes: --run-module needs a module')\n"
        "    module = sys.argv.pop(2)\n"
        "    del sys.argv[1]\n"
        "    runpy.run_module(module, run_name='__main__', alter_sys=True)\n"
        "    sys.exit(0)\n"
        f"from {module} import {func}\n"
        "sys.argv[0] = re.sub(r'(-script\\.pyw|\\.exe)?$', '', sys.argv[0])\n"
        f"sys.exit({func}())\n"
    )


def _write_shell(target: Path, command: list[str]) -> Path | None:
    body = f'#!/bin/sh\nexec {shlex.join(command)} "$@"\n'
    try:
        if not target.is_symlink() and target.read_bytes() == body.encode("utf-8"):
            if os.access(target, os.X_OK):
                return target
    except OSError:
        pass

    def write(staging: Path) -> None:
        staging.write_text(body, encoding="utf-8", newline="\n")
        staging.chmod(0o755)

    return _write_atomic(target, write)


def _mint_shell_launcher(name: str, out_dir: Path, python_exe: Path, script: str) -> Path | None:
    return _write_shell(out_dir / name, [str(python_exe), "-I", "-c", script])


def _owns_launcher(target: Path, root: Path) -> bool:
    """Recognize our old source/venv launchers, never a mere mention in a comment."""
    if target.is_symlink():
        return target.resolve().is_relative_to(root)
    try:
        tokens = shlex.split(target.read_text(encoding="utf-8-sig"), comments=True)
    except (OSError, UnicodeError, ValueError):
        return False
    paths = {str(root / p) for p in (
        "hermes", "run_agent.py", "venv/bin/python", "venv/bin/python3",
        "venv/bin/hermes",
        ".hermes/bin/hermes", ".hermes/bin/hermes-acp",
    )}
    # Current store launchers pass this Python bootstrap as one shell argument.
    bootstrap = f"sys.path.insert(0, {str(root)!r})"
    if paths.intersection(tokens) or any(bootstrap in token for token in tokens):
        return True
    # The historical updater wrote ACP as a sibling-hermes forwarder. Adopt
    # it only when that sibling demonstrably belongs to this installation.
    if target.name == "hermes-acp":
        sibling = target.with_name("hermes")
        return (tokens == ["exec", str(sibling), "acp", "$@"]
                and _owns_launcher(sibling, root))
    return False


def _publish_conveniences(root: Path, out_dir: Path, names, *, create: bool = True) -> dict[Path, bool]:
    """User-bin commands forward to durable local launchers, not a Python pin."""
    if create:
        out_dir.mkdir(parents=True, exist_ok=True)
    elif not out_dir.is_dir():
        return {}
    published = {}
    for name in names:
        target = out_dir / name
        if (not create or target.exists() or target.is_symlink()) and not _owns_launcher(target, root):
            continue
        before = target.lstat().st_mtime_ns if target.exists() or target.is_symlink() else None
        command = ([str(root / ".hermes/bin/hermes"), "--run-module", "run_agent"]
                   if name == "hermes-agent" else [str(root / ".hermes/bin" / name)])
        if _write_shell(target, command) is None:
            raise OSError(f"could not publish launcher {target}")
        published[target] = before != target.lstat().st_mtime_ns
    return published


def stage_launcher(name: str, repo_root: Path, out_dir: Path) -> Path | None:
    """Publish one launcher bound to store Python, or refuse missing tools."""
    repo_root = Path(repo_root)
    store_python = resolve_store_python(repo_root)
    if store_python is not None:
        path = mint_launcher(name, repo_root, out_dir, store_python, None)
        if path is not None and path.suffix == ".cmd":
            # cmd.exe prefers .exe. An older launcher must not shadow the
            # newly published command when distlib is unavailable.
            try:
                (Path(out_dir) / f"{name}.exe").unlink(missing_ok=True)
            except OSError:
                return None
        return path
    return None


def ensure_install_launchers(repo_root: Path, out_dir: Path) -> list[str]:
    """Publish exact-install commands; conveniences follow them across Python repins."""
    root = Path(repo_root).resolve()
    local = root / ".hermes" / "bin"
    local.mkdir(parents=True, exist_ok=True)
    written = [str(path) for name in WINDOWS_BIN_LAUNCHERS
               if (path := stage_launcher(name, root, local)) is not None]
    if Path(out_dir).resolve() == local:
        return written
    if len(written) != len(WINDOWS_BIN_LAUNCHERS):
        return []
    if not _is_windows():
        return [str(path) for path in _publish_conveniences(root, Path(out_dir), WINDOWS_BIN_LAUNCHERS)]
    Path(out_dir).mkdir(parents=True, exist_ok=True)
    return [str(path) for name in WINDOWS_BIN_LAUNCHERS
            if (path := stage_launcher(name, root, Path(out_dir))) is not None]


def expose_cli(project_root: Path | None = None, *, create: bool = True) -> dict:
    """Repair PATH conveniences without taking over another installation's files.

    macOS CLI-first launches link the bundle's signed shims directly; Electron
    need not have run. Source installs converge on the store launcher owner.
    Shell rc/PATH registration remains installer-owned.

    Before dependency sync succeeds, create=False maintains only commands we
    already own, without loading application config or enabling new exposure.
    """
    # Resolved before the platform branch: the Windows path needs it too.
    from pm.paths import install_root

    root = Path(project_root or install_root()).resolve()
    if _is_windows():
        # The installer stages the user-facing commands into $HERMES_HOME\bin
        # and registers that directory in the User PATH. An update skipped both
        # (this used to answer "windows-installer-owned"), so a machine updated
        # from a release predating that convention kept the old
        # venv\Scripts entry and never converged on it. Mirror the installer.
        return _expose_windows_user_bin(root, create=create)
    if create:
        try:
            from hermes_cli.config import load_config
        except ImportError:
            # Installers expose after sync; never invent a config reader here.
            return {"ok": True, "skipped": "config-unavailable"}
        cli_cfg = (load_config() or {}).get("cli", {})
        if isinstance(cli_cfg, dict) and not cli_cfg.get("expose_on_path", True):
            return {"ok": True, "skipped": "config-disabled"}
    from hermes_cli.steward import read_install_stamp

    if _is_bundled_payload(root):
        if create and sys.platform == "darwin":
            return _symlink_sealed_launchers(root.parent / "bin")
        return {"ok": True, "skipped": "bundle-owns-launchers"}
    if read_install_stamp(root).get("updateMechanism") == "external":
        return {"ok": True, "skipped": "externally-owned"}
    if resolve_store_python(root) is None:
        return {"ok": True, "skipped": "no-store-python"}
    try:
        local = root / ".hermes" / "bin"
        if len(ensure_install_launchers(root, local)) != len(WINDOWS_BIN_LAUNCHERS):
            return {"ok": False, "error": "source launcher publication failed"}
        dirs = [Path.home() / ".local" / "bin"]
        # Repair existing FHS/custom-home exposure, but never create new global
        # entries or reclaim a convenience that was repointed to another root.
        from hermes_constants import get_default_hermes_root
        for directory in (get_default_hermes_root() / "bin", Path("/usr/local/bin")):
            if directory not in dirs and (not create or _owns_launcher(directory / "hermes", root)):
                dirs.append(directory)
        written = []
        for directory in dirs:
            published = _publish_conveniences(root, directory, (*WINDOWS_BIN_LAUNCHERS, "hermes-agent"), create=create)
            written.extend(path.name for path, changed in published.items() if changed)
        return {"ok": True, "written": written}
    except OSError as exc:
        return {"ok": False, "error": str(exc)}


def _merge_user_path(existing: str, entry: str) -> str | None:
    """``entry`` first, or None when the PATH already names it.

    Windows paths are case-insensitive and may carry a trailing separator, so
    compare that way rather than by exact string.
    """
    wanted = entry.rstrip("\\/")
    parts = [part for part in (existing or "").split(";") if part]
    if any(part.rstrip("\\/").casefold() == wanted.casefold() for part in parts):
        return None
    return ";".join([entry, *parts])


def _register_windows_user_path(entry: Path) -> str:
    """Put ``entry`` on the User PATH. Returns 'present' or 'added'.

    Preserves the stored value type: install.ps1 writes expandable entries
    (``%LOCALAPPDATA%\\...``), and rewriting the value as a plain string would
    freeze those.
    """
    import winreg  # type: ignore

    with winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Environment", 0,
                        winreg.KEY_QUERY_VALUE | winreg.KEY_SET_VALUE) as key:
        try:
            current, kind = winreg.QueryValueEx(key, "Path")
        except FileNotFoundError:
            current, kind = "", winreg.REG_EXPAND_SZ
        merged = _merge_user_path(str(current), str(entry))
        if merged is None:
            return "present"
        if kind not in (winreg.REG_SZ, winreg.REG_EXPAND_SZ):
            kind = winreg.REG_EXPAND_SZ
        winreg.SetValueEx(key, "Path", 0, kind, merged)
    _broadcast_environment_change()
    return "added"


def _broadcast_environment_change() -> None:
    """Tell running processes the environment changed (best effort)."""
    import ctypes

    try:
        ctypes.windll.user32.SendMessageTimeoutW(0xFFFF, 0x1A, 0, "Environment", 0x0002, 5000, None)  # type: ignore[attr-defined]
    except Exception:  # noqa: BLE001 - never fail an update over the broadcast
        pass


def _expose_windows_user_bin(root: Path, *, create: bool) -> dict:
    """Windows twin of expose_cli's POSIX half.

    Stage the user-facing commands into ``$HERMES_HOME\\bin`` and register that
    directory in the User PATH -- the same two things scripts/install.ps1 does --
    so an update converges a machine installed under the older venv\\Scripts
    convention instead of leaving it there forever.
    """
    from hermes_constants import get_default_hermes_root

    directory = get_default_hermes_root() / "bin"
    try:
        if not create:
            # Bootstrap: repair only commands we already own, enable nothing new.
            published = _publish_conveniences(root, directory, WINDOWS_BIN_LAUNCHERS, create=False)
            return {"ok": True, "written": [path.name for path, changed in published.items() if changed]}
        directory.mkdir(parents=True, exist_ok=True)
        written = ensure_install_launchers(root, directory)
        if len(written) != len(WINDOWS_BIN_LAUNCHERS):
            return {"ok": False, "error": "source launcher publication failed"}
        return {"ok": True, "path": _register_windows_user_path(directory),
                "written": [Path(path).name for path in written]}
    except OSError as exc:
        return {"ok": False, "error": str(exc)}


def _is_bundled_payload(root: Path) -> bool:
    """Is ``root`` a desktop bundle's agent payload? The stamp is the
    authority (payload marker / desktop-app distribution), never a
    sibling-directory sniff; a .git tree is a dev checkout regardless."""
    if (root / ".git").exists():
        return False
    from hermes_cli.steward import STEWARD_DESKTOP, read_install_stamp

    stamp = read_install_stamp(root)
    if not stamp:
        return False
    return bool(stamp.get("payload")) or stamp.get("distribution") == STEWARD_DESKTOP


def _symlink_sealed_launchers(payload_bin) -> dict:
    """Link ~/.local/bin/{hermes,hermes-agent,hermes-acp} at a sealed
    bundle's own prebuilt shims (macOS only).

    Symlinks, not copies: the shims are signed as part of the app bundle,
    and a copy would both orphan the signature's context and go stale on
    every app update — a symlink into the .app follows the bundle's
    content wherever Squirrel.Mac swaps it.

    Ownership guard mirrors expose_cli's wrapper logic: an existing
    entry is replaced only when it is ours — a symlink into THIS app
    bundle's payload — or missing/broken. A user's own `hermes` (pipx,
    another checkout's wrapper) is never touched.
    """
    link_dir = Path.home() / ".local" / "bin"
    payload_root = payload_bin.parent
    written: list[str] = []
    try:
        link_dir.mkdir(parents=True, exist_ok=True)
        for name in ("hermes", "hermes-agent", "hermes-acp"):
            source = payload_bin / name
            if not source.is_file():
                continue
            target = link_dir / name
            if target.is_symlink():
                current = os.readlink(target)
                if current == str(source):
                    continue  # already ours and current
                # Ours if it points into this payload (stale app path from
                # a previous version counts — resolve() of a dangling link
                # still yields the old path text) — or dangling entirely.
                points_into_payload = str(Path(current)).startswith(str(payload_root) + os.sep)
                if not points_into_payload and target.exists():
                    continue  # a live foreign link — user's arrangement
            elif target.exists():
                continue  # a real file we did not write — never clobber
            target.unlink(missing_ok=True)
            target.symlink_to(source)
            written.append(name)
    except OSError as exc:
        return {"ok": False, "error": str(exc)}
    return {"ok": True, "written": written, "mode": "sealed-symlinks"}


if __name__ == "__main__":
    import argparse

    if sys.argv[1:2] == ["--print-runtime-command"]:
        print_runtime_command(Path(__file__).resolve().parents[1], sys.argv[2:])
        raise SystemExit(0)

    parser = argparse.ArgumentParser(description="Publish source-install launchers.")
    parser.add_argument("out_dir", type=Path)
    args = parser.parse_args()
    repo_root = Path(__file__).resolve().parents[1]
    if resolve_store_python(repo_root) is None:
        parser.exit(1, "hermes: store interpreter is missing; finish pm install before publishing launchers\n")
    args.out_dir.mkdir(parents=True, exist_ok=True)
    written = ensure_install_launchers(repo_root, args.out_dir)
    if len(written) != len(ENTRY_POINTS):
        parser.exit(1, "hermes: launcher publication failed\n")
    print("\n".join(written))