"""Explicit-path Git maintenance with frozen plans and interruption receipts.

This is deliberately narrower than a Git frontend: no staging all files,
amending, force-pushing, hook bypasses, filter execution or automatic rollback.
"""

from __future__ import annotations

from contextlib import contextmanager
import hashlib
import os
from pathlib import Path
import re
import shutil
import stat
import tempfile
import uuid

from .common import (
    atomic_json,
    digest,
    identifier,
    now,
    read_json,
    resource_lock,
    state_root,
)
from .process import require_capture
from .workspace import load_project, selected_repositories

MAX_FILE_BYTES = 64 * 1024 * 1024
MAX_GIT_OUTPUT_BYTES = 32 * 1024 * 1024


def _git(
    repo: Path,
    *argv: str,
    allowed: tuple[int, ...] = (0,),
    timeout: int = 30,
    input_bytes: bytes | None = None,
    index_file: Path | None = None,
) -> bytes:
    # Namespace/common-dir overrides can redirect both push and its verification
    # to the same unintended ref. Unknown overrides are refused, not scrubbed.
    # GIT_PAGER is inert because every command explicitly passes --no-pager.
    safe_git_environment = {"GIT_OPTIONAL_LOCKS", "GIT_TERMINAL_PROMPT", "GIT_PAGER"}
    unsafe = {
        key
        for key in os.environ
        if key.startswith("GIT_") and key not in safe_git_environment
    }
    if unsafe:
        raise ValueError(
            "Git environment overrides are unsupported for frozen maintenance plans"
        )
    env = {
        **os.environ,
        "GIT_TERMINAL_PROMPT": "0",
        "GIT_OPTIONAL_LOCKS": "0",
        "GIT_LITERAL_PATHSPECS": "1",
    }
    if index_file is not None:
        env["GIT_INDEX_FILE"] = str(index_file)
    if input_bytes is not None and len(input_bytes) > MAX_FILE_BYTES:
        raise ValueError("Git input exceeds maintenance bounds")
    result = require_capture(
        ["git", "--no-pager", "-C", str(repo), *argv],
        env=env,
        timeout=timeout,
        max_stdout=MAX_GIT_OUTPUT_BYTES,
        max_stderr=1024 * 1024,
        input_bytes=input_bytes,
    )
    if result.returncode not in allowed:
        # Git stderr may contain credentials, hook output or source data.
        raise ValueError(
            f"Git {argv[0]} failed (exit {result.returncode}); inspect the repository locally"
        )
    return result.stdout


def _config(repo: Path) -> tuple[dict[str, list[str]], str]:
    raw = _git(repo, "config", "--null", "--list")
    values: dict[str, list[str]] = {}
    for item in raw.split(b"\0"):
        if not item:
            continue
        key, _, value = item.partition(b"\n")
        values.setdefault(os.fsdecode(key).lower(), []).append(os.fsdecode(value))
    for key, entries in values.items():
        if key in {
            "core.hookspath",
            "core.sshcommand",
            "core.gitproxy",
            "core.alternaterefscommand",
            "extensions.partialclone",
            "push.pushoption",
        } or re.fullmatch(r"remote\..*\.(?:receivepack|uploadpack|vcs)", key):
            raise ValueError(f"Unsupported external Git configuration: {key}")
        if key in {
            "core.fsmonitor",
            "commit.gpgsign",
            "push.gpgsign",
            "remote.origin.mirror",
        } and entries[-1].lower() not in {"false", "no", "off", "0"}:
            raise ValueError(f"Unsupported Git maintenance configuration: {key}")
        if (
            re.fullmatch(r"remote\..*\.promisor", key)
            or key in {"push.recursesubmodules", "submodule.recurse"}
        ) and entries[-1].lower() not in {"false", "no", "off", "0"}:
            raise ValueError(
                "Partial clone or recursive submodule operations require the normal Git workflow"
            )
    hooks_raw = os.fsdecode(_git(repo, "rev-parse", "--git-path", "hooks")).strip()
    hooks = Path(hooks_raw) if Path(hooks_raw).is_absolute() else repo / hooks_raw
    if hooks.is_symlink():
        raise ValueError("Symlinked Git hooks are unsupported")
    # Conservative discovery also covers post-index-change and future hook
    # names; Git's shipped *.sample files are never invoked as hooks.
    if hooks.exists() and any(
        not entry.name.endswith(".sample") and os.access(entry, os.X_OK)
        for entry in hooks.iterdir()
    ):
        raise ValueError(
            "Active Git hooks require the normal reviewed Git workflow; they are never bypassed"
        )
    return values, digest(raw.hex())


def _remote_urls(repo: Path) -> tuple[list[str], list[str]]:
    names = os.fsdecode(_git(repo, "remote")).splitlines()
    if "origin" not in names:
        return [], []
    fetch = os.fsdecode(_git(repo, "remote", "get-url", "--all", "origin")).splitlines()
    push = os.fsdecode(
        _git(repo, "remote", "get-url", "--push", "--all", "origin")
    ).splitlines()
    return fetch, push


def _safe_remote(url: str) -> bool:
    # Explicit standard Git transports only; never an ext:: or custom helper.
    return (
        bool(
            re.match(r"^(?:https://|ssh://|file:///|/)", url)
            or re.fullmatch(r"(?:[A-Za-z0-9_.-]+@)?[A-Za-z0-9_.-]+:[^\s]+", url)
        )
        and "::" not in url
        and "\n" not in url
    )


def _index(
    repo: Path, *, index_file: Path | None = None
) -> tuple[dict[str, dict], str]:
    raw = _git(repo, "ls-files", "--stage", "-z", index_file=index_file)
    entries = {}
    for entry in raw.split(b"\0"):
        if not entry:
            continue
        metadata, _, encoded_path = entry.partition(b"\t")
        mode, object_id, stage = metadata.decode("ascii").split()
        if stage != "0":
            raise ValueError(
                "Unmerged index entries require manual conflict resolution"
            )
        entries[os.fsdecode(encoded_path)] = {"mode": mode, "blob": object_id}
    return entries, hashlib.sha256(raw).hexdigest()


def _paths(
    repo: Path,
    paths: list[str],
    config: dict[str, list[str]],
    *,
    allow_missing: bool = False,
) -> list[dict]:
    if not paths or len(paths) > 256:
        raise ValueError(
            "Select 1–256 explicit file paths; directories and globs are not expanded"
        )
    entries, _ = _index(repo)
    result = []
    for name in sorted(set(paths)):
        relative = Path(name)
        if (
            not isinstance(name, str)
            or not name
            or relative.is_absolute()
            or name != relative.as_posix()
            or any(part in {".", "..", ".git"} for part in name.split("/"))
            or "\0" in name
            or "\n" in name
        ):
            raise ValueError("Maintenance paths must be explicit relative file paths")
        target = repo / relative
        if any(
            part.is_symlink()
            for part in (target, *target.parents)
            if part != repo.parent
        ):
            raise ValueError("Symlinked selected paths are unsupported")
        if not target.resolve().is_relative_to(repo.resolve()):
            raise ValueError("Selected path escapes its repository")
        if entries.get(name, {}).get("mode") == "160000":
            raise ValueError("Submodule paths require the normal reviewed Git workflow")
        if not target.exists():
            if (
                not allow_missing
                and name not in entries
                and name not in _tree_entries(repo, "HEAD")
            ):
                raise ValueError("Selected file is neither present nor tracked")
            result.append(
                {
                    "path": name,
                    "exists": False,
                    "sha256": None,
                    "mode": None,
                    "blob": None,
                }
            )
            continue
        data, metadata = _read_selected_bytes(target)
        attributes = _git(
            repo, "check-attr", "-z", "filter", "working-tree-encoding", "--", name
        ).split(b"\0")
        if any(
            value not in {b"", b"unspecified", b"unset"} for value in attributes[2::3]
        ):
            raise ValueError(
                "Selected filter/encoding attributes require the normal Git workflow"
            )
        blob = os.fsdecode(
            _git(repo, "hash-object", "--path", name, "--stdin", input_bytes=data)
        ).strip()
        mode = "100755" if metadata.st_mode & 0o111 else "100644"
        if (
            config.get("core.filemode", ["true"])[-1].lower() == "false"
            and name in entries
        ):
            mode = entries[name]["mode"]
        result.append(
            {
                "path": name,
                "exists": True,
                "sha256": hashlib.sha256(data).hexdigest(),
                "mode": mode,
                "blob": blob,
            }
        )
    return result


def snapshot(repo: Path, paths: list[str], *, allow_missing: bool = False) -> dict:
    config, config_hash = _config(repo)
    root = Path(
        os.fsdecode(_git(repo, "rev-parse", "--show-toplevel")).strip()
    ).resolve()
    if root != repo.resolve():
        raise ValueError(
            "Select the registered repository root, not an embedded directory"
        )
    if _git(repo, "rev-parse", "--shared-index-path").strip():
        raise ValueError("Split indexes require the normal reviewed Git workflow")
    head = os.fsdecode(_git(repo, "rev-parse", "--verify", "HEAD")).strip()
    branch = os.fsdecode(
        _git(repo, "symbolic-ref", "--quiet", "--short", "HEAD")
    ).strip()

    def git_path(name):
        return Path(os.fsdecode(_git(repo, "rev-parse", "--git-path", name)).strip())

    for marker in (
        "MERGE_HEAD",
        "CHERRY_PICK_HEAD",
        "REVERT_HEAD",
        "rebase-merge",
        "rebase-apply",
        "BISECT_LOG",
    ):
        location = git_path(marker)
        if (location if location.is_absolute() else repo / location).exists():
            raise ValueError(
                "An in-progress Git operation requires the normal reviewed Git workflow"
            )
    grafts = git_path("info/grafts")
    if (grafts if grafts.is_absolute() else repo / grafts).exists() or _git(
        repo, "for-each-ref", "--format=%(refname)", "refs/replace/"
    ).strip():
        raise ValueError(
            "Git replacement/graft history requires the normal reviewed Git workflow"
        )
    flags = _git(repo, "ls-files", "-v", "-z")
    if any(
        item and (chr(item[0]).islower() or item[:1] == b"S")
        for item in flags.split(b"\0")
    ):
        raise ValueError(
            "Assume-unchanged/skip-worktree entries are unsupported for frozen plans"
        )
    fetch, push = _remote_urls(repo)
    index_entries, index_hash = _index(repo)
    # A globally installed but unused LFS driver is harmless. Refuse active
    # filters/encodings across tracked files too: Git may refresh their index
    # metadata during commit, even when they are not selected.
    attr_paths = sorted(set(index_entries) | set(paths))
    attr_input = b"\0".join(os.fsencode(name) for name in attr_paths) + b"\0"
    attrs = _git(
        repo,
        "check-attr",
        "-z",
        "--stdin",
        "filter",
        "working-tree-encoding",
        input_bytes=attr_input,
    ).split(b"\0")
    if any(value not in {b"", b"unspecified", b"unset"} for value in attrs[2::3]):
        raise ValueError(
            "Active Git filter/encoding attributes require the normal reviewed Git workflow"
        )
    return {
        "head": head,
        "branch": branch,
        "origin_fetch_sha256": digest(fetch),
        "origin_push_sha256": digest(push),
        "origin_present": bool(fetch),
        "config_sha256": config_hash,
        "index_sha256": index_hash,
        "unselected_index_sha256": digest(
            {name: entry for name, entry in index_entries.items() if name not in paths}
        ),
        "files": _paths(repo, paths, config, allow_missing=allow_missing),
    }


def _receipt_path(args, plan_id: str) -> Path:
    identifier(plan_id)
    return (
        state_root(Path(args.workspace_root), getattr(args, "state_dir", None))
        / "maintenance"
        / plan_id
    )


def _seal(value: dict) -> dict:
    return {
        **value,
        "integrity_sha256": digest(
            {key: item for key, item in value.items() if key != "integrity_sha256"}
        ),
    }


def _read(args, plan_id: str) -> tuple[dict, dict, Path]:
    directory = _receipt_path(args, plan_id)
    plan, state = (
        read_json(directory / "plan.json"),
        read_json(directory / "state.json"),
    )
    for value in (plan, state):
        if (
            not isinstance(value, dict)
            or value.get("integrity_sha256") != _seal(value)["integrity_sha256"]
        ):
            raise ValueError("Maintenance receipt integrity mismatch")
        if value.get("plan_id") != plan_id:
            raise ValueError("Maintenance receipt identity mismatch")
    if (
        plan.get("workspace_root") != str(Path(args.workspace_root).resolve())
        or state.get("plan_sha256") != plan["integrity_sha256"]
    ):
        raise ValueError("Maintenance receipt belongs to another workspace or plan")
    project = load_project(Path(args.workspace_root), getattr(args, "project", None))
    matched = selected_repositories(project, [plan["repository"]])
    if len(matched) != 1 or str(matched[0].path) != plan["repository_path"]:
        raise ValueError("The registered repository identity changed")
    return plan, state, directory


def _write_state(directory: Path, state: dict, **updates) -> dict:
    result = _seal({**state, **updates, "updated_at": now()})
    atomic_json(directory / "state.json", result)
    return result


def _result(plan: dict, state: dict, *lines: str) -> dict:
    return {
        "plan": plan,
        "state": state,
        "summary": [
            f"Git maintenance {plan['plan_id']}: {state['status']}",
            *lines,
            "Only explicit selected paths are in scope; no hooks/filters bypass, force push, staging-all or automatic rollback.",
        ],
        "_exit_code": 0,
    }


@contextmanager
def _locked(args, repo: Path):
    root = state_root(Path(args.workspace_root), getattr(args, "state_dir", None))
    with resource_lock(root / "locks", "git-maintenance:" + str(repo.resolve())):
        yield


def plan(args) -> dict:
    project = load_project(Path(args.workspace_root), getattr(args, "project", None))
    selected = selected_repositories(project, [args.repo])
    if len(selected) != 1:
        raise ValueError("Select exactly one repository")
    repo = selected[0]
    message = args.message
    if (
        not isinstance(message, str)
        or not message.strip()
        or "\0" in message
        or len(message.encode()) > 16384
    ):
        raise ValueError("A bounded, nonempty commit message is required")
    frozen = snapshot(repo.path, args.path)
    entries, _ = _index(repo.path)
    # A partially staged selected file is ambiguous: users must first decide
    # whether to commit its index or working-tree version through normal Git.
    head_entries = _tree_entries(repo.path, frozen["head"])
    for item in frozen["files"]:
        current = entries.get(item["path"])
        base = head_entries.get(item["path"])
        desired = (
            {"mode": item["mode"], "blob": item["blob"]} if item["exists"] else None
        )
        if current != base and current != desired:
            raise ValueError(
                "A selected file has different staged and working-tree changes; reconcile it explicitly first"
            )
    if all(
        head_entries.get(item["path"])
        == ({"mode": item["mode"], "blob": item["blob"]} if item["exists"] else None)
        for item in frozen["files"]
    ):
        raise ValueError("Selected paths contain no commit changes")
    plan_id = "git-" + uuid.uuid4().hex
    payload = _seal(
        {
            "schema_version": 1,
            "plan_id": plan_id,
            "created_at": now(),
            "workspace_root": str(Path(args.workspace_root).resolve()),
            "repository": repo.name,
            "repository_path": str(repo.path),
            "message": message,
            "snapshot": frozen,
        }
    )
    state = _seal(
        {
            "schema_version": 1,
            "plan_id": plan_id,
            "plan_sha256": payload["integrity_sha256"],
            "status": "planned",
            "updated_at": now(),
        }
    )
    if getattr(args, "apply", False):
        with _locked(args, repo.path):
            if snapshot(repo.path, args.path) != frozen:
                raise ValueError(
                    "Repository changed while the maintenance plan was being prepared"
                )
            directory = _receipt_path(args, plan_id)
            if directory.exists():
                raise ValueError("Maintenance plan already exists")
            atomic_json(directory / "plan.json", payload)
            atomic_json(directory / "state.json", state)
        return _result(
            payload,
            state,
            "Frozen plan saved. Commit and push each require their own explicit --apply.",
        )
    return _result(
        payload,
        {**state, "status": "preview"},
        "Dry run: no plan file, index change, commit or network request was made.",
    )


def _tree_entries(repo: Path, commit: str) -> dict:
    entries = {}
    for row in _git(repo, "ls-tree", "-r", "-z", commit).split(b"\0"):
        if row:
            metadata, _, name = row.partition(b"\t")
            mode, kind, object_id = metadata.decode("ascii").split()
            entries[os.fsdecode(name)] = {"mode": mode, "blob": object_id}
    return entries


def _validate_commit(repo: Path, plan: dict, candidate: str) -> None:
    frozen = plan["snapshot"]
    parents = (
        os.fsdecode(_git(repo, "rev-list", "--parents", "-n", "1", candidate))
        .strip()
        .split()
    )
    if parents != [candidate, frozen["head"]]:
        raise ValueError("HEAD is not the single planned child commit")
    raw = _git(repo, "cat-file", "commit", candidate)
    message = raw.partition(b"\n\n")[2]
    if message.rstrip(b"\n") != plan["message"].encode().rstrip(b"\n"):
        raise ValueError("Commit message does not match the frozen plan")
    before, after = _tree_entries(repo, frozen["head"]), _tree_entries(repo, candidate)
    wanted = {item["path"] for item in frozen["files"]}
    changed = {
        name
        for name in before.keys() | after.keys()
        if before.get(name) != after.get(name)
    }
    if not changed or not changed <= wanted:
        raise ValueError("Commit contains changes outside the approved paths")
    for item in frozen["files"]:
        expected = (
            {"mode": item["mode"], "blob": item["blob"]} if item["exists"] else None
        )
        if after.get(item["path"]) != expected:
            raise ValueError("Committed file content does not match the frozen plan")


def _index_path(repo: Path) -> Path:
    raw = Path(os.fsdecode(_git(repo, "rev-parse", "--git-path", "index")).strip())
    path = raw if raw.is_absolute() else repo / raw
    if (
        path.is_symlink()
        or not path.is_file()
        or path.stat().st_size > MAX_GIT_OUTPUT_BYTES
    ):
        raise ValueError("The real Git index must be a bounded regular file")
    return path


@contextmanager
def _index_lock(repo: Path):
    path = _index_path(repo)
    lock = Path(str(path) + ".lock")
    try:
        descriptor = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    except FileExistsError:
        raise ValueError(
            "Git index is locked; inspect the owning operation, never delete an unverified lock automatically"
        ) from None
    identity = os.fstat(descriptor)
    try:
        yield path, lock, descriptor
    finally:
        os.close(descriptor)
        if lock.exists():
            current = lock.lstat()
            if (current.st_dev, current.st_ino) == (identity.st_dev, identity.st_ino):
                lock.unlink()


def _index_info(files: list[dict]) -> bytes:
    return b"".join(
        (
            f"{item['mode']} {item['blob']}\t{item['path']}\0"
            if item["exists"]
            else f"0 {'0' * 40}\t{item['path']}\0"
        ).encode()
        for item in files
    )


def _read_selected_bytes(path: Path):
    descriptor = os.open(
        path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
    )
    with os.fdopen(descriptor, "rb") as handle:
        metadata = os.fstat(handle.fileno())
        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_FILE_BYTES:
            raise ValueError("Selected content changed or exceeds its bound")
        data = handle.read(MAX_FILE_BYTES + 1)
    if len(data) > MAX_FILE_BYTES:
        raise ValueError("Selected content exceeds its bound")
    return data, metadata


def _capture_blob(repo: Path, item: dict) -> None:
    if not item["exists"]:
        return
    data, _ = _read_selected_bytes(repo / item["path"])
    if len(data) > MAX_FILE_BYTES or hashlib.sha256(data).hexdigest() != item["sha256"]:
        raise ValueError(
            "Selected content changed before it could be captured; prepare another plan"
        )
    # The same captured bytes are hashed and written. No second working-tree
    # read can substitute newer edits into the approved blob.
    blob = os.fsdecode(
        _git(
            repo,
            "hash-object",
            "-w",
            "--path",
            item["path"],
            "--stdin",
            input_bytes=data,
        )
    ).strip()
    if blob != item["blob"]:
        raise ValueError(
            "Git's captured blob differs from the frozen plan; no commit was published"
        )


def _selected_index(repo: Path, source: Path, target: Path, files: list[dict]) -> str:
    shutil.copyfile(source, target)
    _git(
        repo,
        "update-index",
        "-z",
        "--index-info",
        input_bytes=_index_info(files),
        index_file=target,
    )
    return _index(repo, index_file=target)[1]


def _publish_index(repo: Path, locked, prepared: Path, expected_before: str) -> None:
    index, lock, descriptor = locked
    if _index(repo)[1] != expected_before:
        raise ValueError(
            "The real index changed concurrently; its new contents were preserved"
        )
    data = prepared.read_bytes()
    if len(data) > MAX_GIT_OUTPUT_BYTES:
        raise ValueError("Prepared index exceeds its bound")
    os.fchmod(descriptor, stat.S_IMODE(index.stat().st_mode))
    os.ftruncate(descriptor, 0)
    os.lseek(descriptor, 0, os.SEEK_SET)
    view = memoryview(data)
    while view:
        written = os.write(descriptor, view)
        view = view[written:]
    os.fsync(descriptor)
    if _index(repo)[1] != expected_before:
        raise ValueError(
            "The real index changed concurrently; its new contents were preserved"
        )
    os.replace(lock, index)
    directory = os.open(index.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(directory)
    finally:
        os.close(directory)


def commit(args) -> dict:
    payload, state, directory = _read(args, args.plan_id)
    repo = Path(payload["repository_path"])
    with _locked(args, repo) if getattr(args, "apply", False) else _no_lock():
        payload, state, directory = _read(args, args.plan_id)
        if state["status"] == "committed":
            return _result(
                payload, state, "Commit already recorded; no duplicate commit created."
            )
        if state["status"] != "planned":
            raise ValueError(
                "Commit requires a planned receipt; reconcile interrupted/uncertain state first"
            )
        paths = [item["path"] for item in payload["snapshot"]["files"]]
        if snapshot(repo, paths) != payload["snapshot"]:
            raise ValueError(
                "HEAD, branch, origin, config, index or selected file content changed; prepare a new plan"
            )
        if not getattr(args, "apply", False):
            return _result(
                payload,
                {**state, "status": "preview"},
                "Dry run: the frozen selected-path commit is ready; no index or commit was changed.",
            )
        state = _write_state(directory, state, status="committing")
        try:
            with (
                _index_lock(repo) as locked,
                tempfile.TemporaryDirectory(
                    prefix=".commit-index-", dir=directory
                ) as scratch,
            ):
                if _index(repo)[1] != payload["snapshot"]["index_sha256"]:
                    raise ValueError("Real index changed before its lock was acquired")
                private = Path(scratch) / "commit.index"
                target = Path(scratch) / "selected.index"
                for item in payload["snapshot"]["files"]:
                    _capture_blob(repo, item)
                _git(repo, "read-tree", payload["snapshot"]["head"], index_file=private)
                _git(
                    repo,
                    "update-index",
                    "-z",
                    "--index-info",
                    input_bytes=_index_info(payload["snapshot"]["files"]),
                    index_file=private,
                )
                tree = os.fsdecode(_git(repo, "write-tree", index_file=private)).strip()
                target_hash = _selected_index(
                    repo, locked[0], target, payload["snapshot"]["files"]
                )
                current = snapshot(repo, paths, allow_missing=True)
                if any(
                    current[key] != payload["snapshot"][key]
                    for key in (
                        "head",
                        "branch",
                        "origin_fetch_sha256",
                        "origin_push_sha256",
                        "config_sha256",
                        "index_sha256",
                    )
                ):
                    raise ValueError(
                        "Repository identity or real index changed during commit preparation"
                    )
                candidate = os.fsdecode(
                    _git(
                        repo,
                        "commit-tree",
                        tree,
                        "-p",
                        payload["snapshot"]["head"],
                        "-m",
                        payload["message"],
                    )
                ).strip()
                _validate_commit(repo, payload, candidate)
                state = _write_state(
                    directory,
                    state,
                    status="committing",
                    phase="prepared",
                    commit=candidate,
                    prepared_tree=tree,
                    target_index_sha256=target_hash,
                )
                # Atomic compare-and-swap cannot overwrite another commit that
                # advanced this branch after our snapshot.
                _git(
                    repo,
                    "update-ref",
                    "--no-deref",
                    "-m",
                    "devkit selected-path commit",
                    f"refs/heads/{payload['snapshot']['branch']}",
                    candidate,
                    payload["snapshot"]["head"],
                )
                if os.fsdecode(_git(repo, "rev-parse", "HEAD")).strip() != candidate:
                    raise ValueError(
                        "HEAD changed while publishing the prepared commit; inspect before reconciliation"
                    )
                _publish_index(
                    repo, locked, target, payload["snapshot"]["index_sha256"]
                )
            committed_snapshot = snapshot(repo, paths, allow_missing=True)
            if (
                committed_snapshot["head"] != candidate
                or committed_snapshot["branch"] != payload["snapshot"]["branch"]
            ):
                raise ValueError(
                    "HEAD or branch changed before the exact commit result could be recorded"
                )
            if (
                committed_snapshot["unselected_index_sha256"]
                != payload["snapshot"]["unselected_index_sha256"]
            ):
                raise ValueError(
                    "Unrelated index entries changed during the commit; reconcile before proceeding"
                )
            state = _write_state(
                directory,
                state,
                status="committed",
                commit=candidate,
                committed_snapshot=committed_snapshot,
            )
        except BaseException:
            _write_state(directory, state, status="needs_reconcile", operation="commit")
            raise
        return _result(
            payload,
            state,
            "Selected paths committed. Unrelated staged/dirty files were left untouched; nothing was pushed.",
        )


@contextmanager
def _no_lock():
    yield


def _remote_head(repo: Path, url: str, branch: str) -> str | None:
    raw = _git(repo, "ls-remote", "--refs", url, f"refs/heads/{branch}", timeout=60)
    entries = raw.splitlines()
    if not entries:
        return None
    if len(entries) != 1:
        raise ValueError("Remote branch response is ambiguous")
    object_id, _, ref = entries[0].partition(b"\t")
    if ref.decode() != f"refs/heads/{branch}" or not re.fullmatch(
        rb"[0-9a-f]{40,64}", object_id
    ):
        raise ValueError("Remote branch response is malformed")
    return object_id.decode()


def push(args) -> dict:
    payload, state, directory = _read(args, args.plan_id)
    repo = Path(payload["repository_path"])
    with _locked(args, repo) if getattr(args, "apply", False) else _no_lock():
        payload, state, directory = _read(args, args.plan_id)
        if state["status"] == "pushed":
            return _result(
                payload,
                state,
                "Push already recorded; no duplicate network mutation attempted.",
            )
        if state["status"] != "committed":
            raise ValueError(
                "Push requires a verified recorded commit; reconcile uncertain state first"
            )
        paths = [item["path"] for item in payload["snapshot"]["files"]]
        current = snapshot(repo, paths, allow_missing=True)
        if current != state["committed_snapshot"] or current["head"] != state["commit"]:
            raise ValueError(
                "Repository changed after its recorded commit; push is blocked"
            )
        _validate_commit(repo, payload, state["commit"])
        _, urls = _remote_urls(repo)
        if len(urls) != 1 or not _safe_remote(urls[0]):
            raise ValueError(
                "Push requires exactly one origin push URL using a supported standard Git transport"
            )
        if not getattr(args, "apply", False):
            return _result(
                payload,
                {**state, "status": "preview"},
                "Dry run: normal push of the recorded commit/branch only; no network request made.",
            )
        state = _write_state(directory, state, status="pushing")
        try:
            _git(
                repo,
                "push",
                "--porcelain",
                "--no-follow-tags",
                "--recurse-submodules=no",
                "origin",
                f"{state['commit']}:refs/heads/{payload['snapshot']['branch']}",
                timeout=180,
            )
            remote_head = _remote_head(repo, urls[0], payload["snapshot"]["branch"])
            if remote_head != state["commit"]:
                raise ValueError(
                    "Remote head does not match the recorded commit; reconcile before retrying"
                )
            state = _write_state(
                directory, state, status="pushed", remote_commit=remote_head
            )
        except BaseException:
            _write_state(directory, state, status="needs_reconcile", operation="push")
            raise
        return _result(
            payload,
            state,
            "Normal push verified at the exact recorded branch. No force push or extra tags were requested.",
        )


def status(args) -> dict:
    payload, state, _ = _read(args, args.plan_id)
    return _result(
        payload,
        state,
        "Recorded local maintenance state; no remote request or repository mutation made.",
    )


def reconcile(args) -> dict:
    payload, state, directory = _read(args, args.plan_id)
    repo = Path(payload["repository_path"])
    if state["status"] not in {"committing", "pushing", "needs_reconcile"}:
        raise ValueError(
            "Only interrupted/uncertain maintenance requires reconciliation"
        )
    if not getattr(args, "apply", False):
        return _result(
            payload,
            {**state, "status": "preview"},
            "Dry run: reconciliation can record an exact existing commit/push, never create either. Push reconciliation reads the frozen remote only with --apply.",
        )
    with _locked(args, repo):
        payload, state, directory = _read(args, args.plan_id)
        if state["status"] in {"committed", "pushed", "not_committed"}:
            return _result(
                payload,
                state,
                "Another reconciliation already recorded this result; no replay or downgrade occurred.",
            )
        if state["status"] not in {"committing", "pushing", "needs_reconcile"}:
            raise ValueError("Receipt is no longer awaiting reconciliation")
        paths = [item["path"] for item in payload["snapshot"]["files"]]
        current = snapshot(repo, paths, allow_missing=True)
        if any(
            current[key] != payload["snapshot"][key]
            for key in (
                "branch",
                "origin_fetch_sha256",
                "origin_push_sha256",
                "config_sha256",
                "unselected_index_sha256",
            )
        ):
            raise ValueError(
                "Repository/remote identity changed; reconciliation cannot claim a result"
            )
        if state.get("operation") == "push" or state["status"] == "pushing":
            _, urls = _remote_urls(repo)
            if (
                len(urls) != 1
                or not _safe_remote(urls[0])
                or current["head"] != state.get("commit")
            ):
                raise ValueError("Frozen push identity cannot be established")
            remote = _remote_head(repo, urls[0], payload["snapshot"]["branch"])
            if remote != state["commit"]:
                raise ValueError(
                    "Push is not verifiably complete; inspect the remote before preparing another operation"
                )
            state = _write_state(
                directory, state, status="pushed", remote_commit=remote
            )
        else:
            if current["head"] == payload["snapshot"]["head"]:
                state = _write_state(directory, state, status="not_committed")
                return _result(
                    payload,
                    state,
                    "The planned commit is not current on this branch. Nothing was rolled back; inspect any prepared objects and create a new plan.",
                )
            if current["head"] != state.get("commit"):
                raise ValueError(
                    "HEAD is not the recorded prepared commit; reconciliation cannot adopt another commit"
                )
            _validate_commit(repo, payload, current["head"])
            if current["index_sha256"] != state.get("target_index_sha256"):
                if current["index_sha256"] != payload["snapshot"]["index_sha256"]:
                    raise ValueError(
                        "The real index changed independently; preserve it and resolve selected staging through normal Git"
                    )
                with (
                    _index_lock(repo) as locked,
                    tempfile.TemporaryDirectory(
                        prefix=".reconcile-index-", dir=directory
                    ) as scratch,
                ):
                    prepared = Path(scratch) / "selected.index"
                    target = _selected_index(
                        repo, locked[0], prepared, payload["snapshot"]["files"]
                    )
                    if target != state.get("target_index_sha256"):
                        raise ValueError(
                            "Reconstructed index differs from the recorded prepared index"
                        )
                    checked = snapshot(repo, paths, allow_missing=True)
                    if (
                        checked["head"] != state["commit"]
                        or checked["branch"] != payload["snapshot"]["branch"]
                    ):
                        raise ValueError(
                            "HEAD or branch changed during index reconciliation; no result was adopted"
                        )
                    _publish_index(
                        repo, locked, prepared, payload["snapshot"]["index_sha256"]
                    )
            current = snapshot(repo, paths, allow_missing=True)
            if (
                current["head"] != state["commit"]
                or current["branch"] != payload["snapshot"]["branch"]
            ):
                raise ValueError(
                    "HEAD or branch changed before the exact reconciled result could be recorded"
                )
            _validate_commit(repo, payload, state["commit"])
            state = _write_state(
                directory,
                state,
                status="committed",
                commit=state["commit"],
                committed_snapshot=current,
            )
        return _result(
            payload,
            state,
            "The already-existing result was verified and recorded; no commit or push was executed.",
        )


def register(subparsers) -> None:
    parser = subparsers.add_parser(
        "git", help="Frozen explicit-path Git maintenance; all writes require --apply"
    )
    commands = parser.add_subparsers(dest="git_command", required=True)
    create = commands.add_parser(
        "plan",
        help="Preview/save exact repository, paths, content, index and commit message",
    )
    create.add_argument("--repo", required=True)
    create.add_argument("--path", action="append", required=True)
    create.add_argument("--message", required=True)
    create.add_argument(
        "--apply",
        action="store_true",
        help="Save an immutable local plan; never commits or pushes",
    )
    create.set_defaults(handler=plan)
    for name, handler in (
        ("commit", commit),
        ("push", push),
        ("status", status),
        ("reconcile", reconcile),
    ):
        command = commands.add_parser(name)
        command.add_argument("plan_id")
        if name != "status":
            command.add_argument("--apply", action="store_true")
        command.set_defaults(handler=handler)