"""The single commit point of a source ``hermes update``: everything the tree move needs armed first.

Before git (or the ZIP swap) writes the first file of the new code, three things are durable:

* the interrupted-pull marker (``.git/hermes-update-pull``) naming the commit git moves to, for EVERY
  tree move (branch switch, upstream fork pull, the pull itself) so a launch after any exit that left
  a torn tree puts the old one back (``_early_recovery.restore_interrupted_pull``);
* the source-completion tail (``venv_sync.arm_completion``) and the host fleet-restart obligation,
  so a kill after the move but before the completion child starts still leaves the tail owed to the
  next launch (``venv_sync.prepare_launch``) and the restart owed to the next ``hermes update``.

Failures before the move are no-ops: ``disarm_commit_obligations`` puts both obligations back the way
this run found them once the tree is verified at its pre-update commit again.
"""

from __future__ import annotations

import os
import subprocess
import sys
from contextlib import suppress
from pathlib import Path
from typing import Optional

from hermes_cli._early_recovery import (
    RECOVERY_CLOSURE,
    RECOVERY_CLOSURE_INIT,
    RECOVERY_CLOSURE_MANIFEST,
    blob_id,
    interrupted_pull_marker,
    is_object_id,
    recovery_closure_dir,
    recovery_closure_verified,
    restore_interrupted_pull,
)
from hermes_cli.update_custody import run_git

# What this run found before it armed anything: {path: bytes or None}. None = nothing armed yet.
_armed_snapshot: Optional[dict[Path, Optional[bytes]]] = None
# (git_cmd, (HEAD, branch)) when this run reached its checkout phase, and the checkout it names.
_run_start: Optional[tuple[list, tuple[str, str]]] = None
_obligation_root: Optional[Path] = None


def _owns_live_checkout(root: Path) -> bool:
    from hermes_cli.update_cmd import _m

    return _m()._pytest_owns_live_checkout(root)


def _obligation_paths(root: Path) -> list[Path]:
    from hermes_cli.update_cmd_fleet import _fleet_restart_pending_marker_path
    from hermes_cli.update_host_obligation import host_obligation_path
    from hermes_cli.venv_sync import completion_pending_path

    paths = [completion_pending_path(root), _fleet_restart_pending_marker_path()]
    host = host_obligation_path()
    if host is not None:
        paths.append(host)
    return paths


def arm_commit_obligations(root: Path, expected_sha: str) -> None:
    """Owe the completion tail and the fleet restart for ``expected_sha`` BEFORE the tree moves.

    Idempotent within a run (the first call snapshots what to restore on a no-op failure). An
    unwritable install state raises: nothing has moved yet, and moving without the obligation is
    exactly the tail-never-runs state this exists to prevent.
    """
    global _armed_snapshot
    from hermes_cli.update_cmd_fleet import _write_fleet_restart_pending_marker
    from hermes_cli.venv_sync import arm_completion

    root = Path(root)
    if _owns_live_checkout(root):
        return
    if _armed_snapshot is None:
        snapshot: dict[Path, Optional[bytes]] = {}
        for path in _obligation_paths(root):
            try:
                snapshot[path] = path.read_bytes()
            except OSError:
                snapshot[path] = None
        _armed_snapshot = snapshot
    arm_completion(root)
    _write_fleet_restart_pending_marker(expected_sha=expected_sha or "")


def disarm_commit_obligations() -> None:
    """Restore both obligations to what this run found: the tree never left its pre-update commit.

    Refused (obligations stay armed) unless HEAD is the commit this run STARTED from: a failed later
    move (the upstream fork ff after a committed origin pull) is put back to a commit that is
    already new code, and that tree still owes its tail.
    """
    global _armed_snapshot
    if _run_start is not None:
        git_cmd, (start_head, _branch) = _run_start
        root = Path(_obligation_root) if _obligation_root is not None else None
        if root is None or not start_head or head_and_branch(git_cmd, root)[0] != start_head:
            return
    snapshot, _armed_snapshot = _armed_snapshot, None
    for path, data in (snapshot or {}).items():
        try:
            if data is None:
                path.unlink(missing_ok=True)
            else:
                tmp = path.with_name(path.name + ".restore")
                tmp.write_bytes(data)
                os.replace(tmp, path)
        except OSError:
            pass  # an owed tail/restart left armed is a retry, never a lost obligation


def commit_obligations_armed() -> bool:
    return _armed_snapshot is not None


def arm_tree_move(git_cmd, root: Path, *, pre: str | None, target: str, stash: str | None,
                  rollback: str | None = None) -> Path:
    """Write the interrupted-pull marker for one git tree move (pre -> target).

    ``rollback`` (``branch``/``detach``): a syntax rollback that moves HEAD and the index back to
    ``pre`` before any file; a kill before that step finds HEAD still on ``target`` and the restore
    redoes it first.
    """
    marker = interrupted_pull_marker(root)
    if pre:
        with suppress(OSError, subprocess.SubprocessError):  # no closure: the in-tree repair still runs
            publish_recovery_closure(git_cmd, root, pre)
    marker.write_text(f"pid={os.getpid()}\npre={pre or ''}\ntarget={target}\nstash={stash or ''}\n"
                      + (f"rollback={rollback}\n" if rollback else "")
                      + (f"git={git}\n" if (git := _absolute_git(git_cmd)) else ""), encoding="utf-8")
    return marker


def publish_recovery_closure(git_cmd, root: Path, pre: str) -> Path:
    """Copy the launch repair's modules, as committed at ``pre``, beside the marker (outside the tree).

    The bytes come from the commit the repair restores, never from the working tree: a rollback move
    starts from a broken tree, and a second move in one run starts from code this process never ran.
    A manifest names each file's blob id; the launcher runs the closure only when every file hashes
    to it (``recovery_closure_verified``) and rebuilds it from git's objects otherwise. Files and
    directory are fsynced before the atomic rename, so a power loss during the move that follows
    finds the whole closure or none. An older run's closure (another ``pre``) is dropped.
    """
    import shutil

    if not is_object_id(pre):
        raise OSError(f"not a commit id: {pre!r}")
    dest = recovery_closure_dir(Path(root), pre)
    if recovery_closure_verified(dest, pre):
        return dest
    git_cmd = [*([git_cmd] if isinstance(git_cmd, str) else git_cmd), "--no-replace-objects"]
    listed = run_git(git_cmd, ["ls-tree", "-z", "--full-tree", pre, "--", *RECOVERY_CLOSURE], cwd=str(root),
                     capture_output=True, stdin=subprocess.DEVNULL, timeout=120)
    ids = {}
    for entry in (listed.stdout or b"").split(b"\0") if listed.returncode == 0 else ():
        meta, _, rel = entry.decode("utf-8", "replace").partition("\t")
        if meta.split()[1:2] == ["blob"]:
            ids[rel] = meta.split()[2]
    blobs = {RECOVERY_CLOSURE_INIT: b""}
    for rel in RECOVERY_CLOSURE:
        shown = run_git(git_cmd, ["cat-file", "blob", ids.get(rel, "")], cwd=str(root), capture_output=True,
                        stdin=subprocess.DEVNULL, timeout=120) if rel in ids else None
        if shown is None or shown.returncode != 0 or blob_id(shown.stdout or b"", pre) != ids[rel]:
            raise OSError(f"{rel} is not in {pre[:10]}")
        blobs[rel] = shown.stdout
    manifest = "".join(f"{blob_id(data, pre)} {rel}\n" for rel, data in blobs.items()).encode("utf-8")
    staging = dest.with_name(f".{pre}.{os.getpid()}.staging")
    shutil.rmtree(staging, ignore_errors=True)
    try:
        for rel, data in {**blobs, RECOVERY_CLOSURE_MANIFEST: manifest}.items():
            (staging / rel).parent.mkdir(parents=True, exist_ok=True)
            with open(staging / rel, "wb") as handle:
                handle.write(data)
                handle.flush()
                os.fsync(handle.fileno())
        for directory in (staging / "hermes_cli", staging):
            _fsync_dir(directory)
        shutil.rmtree(dest, ignore_errors=True)
        os.replace(staging, dest)
        _fsync_dir(dest.parent)
    finally:
        shutil.rmtree(staging, ignore_errors=True)
    for stale in dest.parent.iterdir():
        if stale != dest:
            shutil.rmtree(stale, ignore_errors=True)
    return dest


def _fsync_dir(directory: Path) -> None:
    """POSIX: make a directory's entries durable. Windows opens no directory handle; NTFS journals
    the rename itself."""
    if os.name == "nt":
        return
    fd = os.open(directory, os.O_RDONLY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def _absolute_git(git_cmd) -> str:
    """The git this run moves the tree with, as an absolute path for the next launch's repair.

    A Windows install's git is often PM's store copy, on PATH only inside the updater
    (``expose_pm_git``); the repair cannot look it up through ``pm`` when the killed move tore a
    module ``pm`` imports (``hermes_constants``)."""
    import shutil

    head = (list(git_cmd) or ["git"])[0] if not isinstance(git_cmd, str) else git_cmd
    found = shutil.which(head)
    return os.path.abspath(found) if found else ""


def files_added_by(git_cmd, root: Path, pre: str, target: str | None) -> list[str]:
    """Paths ``target`` adds over ``pre``. A rollback's mixed reset un-tracks them and ``reset --hard``
    never touches untracked files, so ``drop_added_files`` removes them by name afterwards."""
    if not target or target == pre:
        return []
    cp = run_git(git_cmd, ["diff", "--name-only", "-z", "--no-renames", "--diff-filter=A", pre, target],
                        cwd=str(root), capture_output=True, text=True, encoding="utf-8", errors="replace",
                        stdin=subprocess.DEVNULL, timeout=120)
    return [p for p in cp.stdout.split("\0") if p] if cp.returncode == 0 else []


def drop_added_files(root: Path, added: list[str]) -> bool:
    """Every one is the move's own file: git refuses to move a checkout over an untracked file.

    False when one of them is still there (the caller keeps the marker; the restore retries)."""
    root = Path(root)
    gone = True
    for rel in added:
        try:
            (root / rel).unlink(missing_ok=True)
        except OSError:
            gone = False
    for parent in sorted({p for rel in added for p in Path(rel).parents if str(p) != "."},
                         key=lambda p: len(p.parts), reverse=True):
        try:
            (root / parent).rmdir()  # only when empty: anything else inside keeps it
        except OSError:
            pass
    return gone and not any(os.path.lexists(root / rel) for rel in added)


def tree_whole_at(git_cmd, root: Path, sha: str) -> bool:
    """HEAD is ``sha``, no ``index.lock`` is left and no tracked file differs from it."""
    def run(*args: str) -> subprocess.CompletedProcess:
        return run_git(git_cmd, [*args], cwd=str(root), capture_output=True, text=True, encoding="utf-8",
                              errors="replace", stdin=subprocess.DEVNULL, timeout=120)

    head = run("rev-parse", "-q", "--verify", "HEAD")
    if head.returncode != 0 or head.stdout.strip() != sha or (interrupted_pull_marker(Path(root)).parent / "index.lock").exists():
        return False
    status = run("status", "--porcelain", "-z", "--untracked-files=no")
    return status.returncode == 0 and not status.stdout.strip("\0")


def settle_failed_tree_move(root: Path) -> bool:
    """Git exited without finishing a tree move: put back what it wrote, keep the marker if we can't.

    True when the tree is verified whole again (at the pre-move commit, or at the target when git got
    there anyway); False leaves the marker for the next launch's ``restore_interrupted_pull``.
    """
    restore_interrupted_pull(Path(root), after_failure=True)
    return not interrupted_pull_marker(Path(root)).is_file()


def requires_other_python(pyproject: bytes | str | None) -> bool:
    """True when a target's ``requires-python`` excludes the running interpreter.

    Its startup modules may then use a newer Python's syntax that this interpreter's ``compile()``
    cannot judge (PM provisions the required Python with the new code). Unknown -> False: keep
    checking, so a real syntax error is never waved through on a guess.
    """
    if not pyproject:
        return False
    import platform
    import re
    import tomllib

    try:
        text = pyproject.decode("utf-8") if isinstance(pyproject, bytes) else pyproject
        spec = tomllib.loads(text)["project"]["requires-python"]
        try:
            from packaging.specifiers import SpecifierSet
        except ImportError:  # the lower bound is the part a Python bump moves
            floor = re.search(r">=\s*(\d+)\.(\d+)", spec)
            return bool(floor) and sys.version_info[:2] < (int(floor[1]), int(floor[2]))
        return not SpecifierSet(spec).contains(platform.python_version(), prereleases=True)
    except (KeyError, TypeError, ValueError):  # TOMLDecodeError / InvalidSpecifier are ValueErrors
        return False


def target_syntax_error(git_cmd, root: Path, target_ref: str, relpaths) -> tuple[str, str] | None:
    """``(path, error)`` for the first startup-critical file that does not compile at ``target_ref``.

    Read from the object store, never written to the tree: this runs BEFORE HEAD moves, so a broken
    release is refused with the install untouched (the post-pull rollback stays as the backstop).
    Skipped for a target that requires a Python this interpreter is not (``requires_other_python``).
    """
    pyproject = run_git(git_cmd, ["show", f"{target_ref}:pyproject.toml"], cwd=str(root),
                               capture_output=True, stdin=subprocess.DEVNULL, timeout=120)
    if pyproject.returncode == 0 and requires_other_python(pyproject.stdout):
        return None
    for rel in relpaths:
        shown = run_git(git_cmd, ["show", f"{target_ref}:{rel}"], cwd=str(root), capture_output=True,
                               stdin=subprocess.DEVNULL, timeout=120)
        if shown.returncode != 0:
            continue  # absent at the target (or unreadable): the post-pull guard has the last word
        try:
            compile(shown.stdout, rel, "exec", dont_inherit=True)
        except (SyntaxError, ValueError) as exc:
            return rel, f"{type(exc).__name__}: {exc}"
    return None


def head_and_branch(git_cmd, root: Path) -> tuple[str, str]:
    def out(*args: str) -> str:
        cp = run_git(git_cmd, [*args], cwd=str(root), capture_output=True, text=True, encoding="utf-8", errors="replace",
                            stdin=subprocess.DEVNULL, timeout=60)
        return cp.stdout.strip() if cp.returncode == 0 else ""

    return out("rev-parse", "HEAD"), out("rev-parse", "--abbrev-ref", "HEAD")


def checkout_untouched(git_cmd, root: Path, start: tuple[str, str] | None) -> bool:
    """True when the checkout is exactly where this run found it: same HEAD, same branch, no
    autostash taken, no tree move armed. Only then may a failed git update fall back to ZIP."""
    from hermes_cli.update_cmd_stash import _unrestored_autostash_notice

    if start is None or not start[0]:
        return False
    if _unrestored_autostash_notice() is not None or interrupted_pull_marker(root).is_file():
        return False
    return head_and_branch(git_cmd, root) == start


def record_run_start(git_cmd, root: Path) -> None:
    global _run_start, _obligation_root
    _run_start = (list(git_cmd), head_and_branch(git_cmd, root))
    _obligation_root = Path(root)


def run_checkout_untouched(root: Path) -> bool:
    """``checkout_untouched`` for this run; True before the checkout phase (nothing can have moved)."""
    if _run_start is None:
        return True
    return checkout_untouched(_run_start[0], Path(root), _run_start[1])


def preflight_refusal(git_cmd, root: Path, target_ref: str, critical_files) -> str | None:
    """Why this update must not start, checked BEFORE the first tree move; ``None`` to proceed.

    * a venv owned by another OS user (the completion child used to refuse only after the swap);
    * a target whose startup-critical modules do not compile (rollback used to be the only guard).
    """
    root = Path(root)
    if not _owns_live_checkout(root):
        try:
            from hermes_cli.venv_sync import refuse_foreign_owned_venv

            refuse_foreign_owned_venv(root)
        except ImportError:
            pass
        except Exception as exc:  # pm's refusal carries its own remediation text
            return f"✗ {exc}"
    broken = target_syntax_error(git_cmd, root, target_ref, critical_files)
    if broken is not None:
        path, error = broken
        return (f"✗ The update target has a syntax error in a critical file:\n  {path}\n    "
                + "\n    ".join(error.splitlines()[:6]))
    return None


def reset_for_tests() -> None:
    global _armed_snapshot, _run_start, _obligation_root
    _armed_snapshot = None
    _run_start = None
    _obligation_root = None