#!/usr/bin/env python3
"""Assignment-bound Git gate routed exclusively through the PR14 SCM broker."""

from __future__ import annotations

import json
import os
import re
import stat
import subprocess
import threading
import urllib.parse
from pathlib import Path
from typing import Any

import scm_broker_client
from execution_pool_project import ATLAS_REPO, validate_branch
from execution_pool_protocol import ProtocolError, atomic_json, verify_envelope


WORKSPACE_ROOT = Path(os.environ.get("HERMES_WORKER_ROOT", "/workspace"))
SCM_ROOT = Path(os.environ.get("HERMES_SCM_STATE_ROOT", "/scm-state"))
ORDINAL = int(os.environ.get("HERMES_WORKER_ORDINAL", "-1"))
BROKER_ORIGIN = scm_broker_client.BROKER_ORIGIN.rstrip("/")
MAX_STATUS_BYTES = 4 * 1024 * 1024


def _git_environment() -> dict[str, str]:
    """Run Git without credentials, prompts, ambient config, or hook execution."""
    return {
        "HOME": "/nonexistent",
        "PATH": "/usr/bin:/bin",
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_TERMINAL_PROMPT": "0",
    }


def _run(*arguments: str, cwd: Path | None = None, timeout: int = 300) -> str:
    command = [
        "/usr/bin/git",
        "-c",
        "core.hooksPath=/dev/null",
        "-c",
        "core.fsmonitor=false",
        *arguments,
    ]
    completed = subprocess.run(
        command,
        cwd=cwd,
        env=_git_environment(),
        stdin=subprocess.DEVNULL,
        text=True,
        capture_output=True,
        timeout=timeout,
        check=False,
    )
    if completed.returncode:
        message = (completed.stderr or completed.stdout or "SCM operation failed")[-2000:]
        raise RuntimeError(message.strip())
    if len(completed.stdout.encode()) > MAX_STATUS_BYTES:
        raise ProtocolError("SCM command output exceeds the safe limit")
    return completed.stdout.strip()


def _binding(envelope: dict[str, Any]) -> tuple[dict[str, Any], str, str, str]:
    if envelope["kind"] != "assignment" or int(envelope["worker_ordinal"]) != ORDINAL:
        raise ProtocolError("assignment does not belong to this worker ordinal")
    payload = envelope.get("payload")
    if not isinstance(payload, dict):
        raise ProtocolError("assignment payload must be an object")
    repo = str(payload.get("repo_url") or "")
    match = ATLAS_REPO.fullmatch(repo)
    if not match:
        raise ProtocolError("assignment repository is outside Atlas")
    try:
        branch = validate_branch(payload.get("branch"), feature=True)
        base = validate_branch(payload.get("base_branch"), feature=False)
    except ValueError as error:
        raise ProtocolError(str(error)) from error
    return payload, match.group("repo"), branch, base


def _broker_repo(repo: str) -> str:
    return f"{BROKER_ORIGIN}/git/atlas/{repo}.git"


def workspace_path(envelope: dict[str, Any]) -> Path:
    """Derive a private ordinal path; no caller-provided path is accepted."""
    parts = tuple(str(envelope[name]) for name in ("board", "task_id", "run_id"))
    identifier = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\Z")
    if any(not identifier.fullmatch(part) for part in parts):
        raise ProtocolError("assignment path binding is invalid")
    if WORKSPACE_ROOT.is_symlink():
        raise ProtocolError("workspace root must not be a symlink")
    workspace_root = WORKSPACE_ROOT.resolve()
    root = WORKSPACE_ROOT / "runs"
    if root.is_symlink():
        raise ProtocolError("workspace run root must not be a symlink")
    root.mkdir(mode=0o700, parents=True, exist_ok=True)
    root = root.resolve()
    root.relative_to(workspace_root)
    candidate = root.joinpath(*parts)
    current = root
    for part in parts[:-1]:
        current /= part
        if current.is_symlink():
            raise ProtocolError("workspace parent must not be a symlink")
        current.mkdir(mode=0o700, exist_ok=True)
    if candidate.is_symlink():
        raise ProtocolError("workspace must not be a symlink")
    candidate.resolve(strict=False).relative_to(root)
    return candidate


def _regular_text(path: Path, maximum: int) -> str:
    descriptor = os.open(
        path, os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0)
    )
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode) or info.st_size > maximum:
            raise ProtocolError("private SCM state is invalid")
        raw = os.read(descriptor, maximum + 1)
    finally:
        os.close(descriptor)
    try:
        return raw.decode().strip()
    except UnicodeError as error:
        raise ProtocolError("private SCM state is malformed") from error


def _state_path(envelope: dict[str, Any]) -> Path:
    if SCM_ROOT.is_symlink():
        raise ProtocolError("private SCM root must not be a symlink")
    root = SCM_ROOT.resolve()
    root.mkdir(mode=0o700, parents=True, exist_ok=True)
    name = "-".join(
        str(envelope[name]) for name in ("board", "task_id", "run_id")
    )
    path = root / f"{name}.json"
    if path.is_symlink():
        raise ProtocolError("private SCM state must not be a symlink")
    path.resolve(strict=False).relative_to(root)
    return path


def submission_refs(branch: str, attempt: int, head: str) -> tuple[str, ...]:
    """Candidate push refs for one attempt, in order of preference.

    The broker accepts branch *creation* only, so a retry that adds commits can
    never update the ref a previous attempt already published, and no attempt may
    ever move a protected ref. Each fallback is therefore a fresh name in the same
    reviewed namespace -- bound first to the exact attempt, then to the exact
    content -- so every push stays a creation and stays idempotent under replay.
    """
    candidates = (
        branch,
        f"{branch}-attempt-{max(1, int(attempt))}",
        f"{branch}-{head[:12]}",
    )
    allowed: list[str] = []
    for candidate in dict.fromkeys(candidates):
        try:
            allowed.append(validate_branch(candidate, feature=True))
        except ValueError:
            continue
    if not allowed:
        raise ProtocolError("no reviewed branch name is available for this attempt")
    return tuple(allowed)


def _remote_heads(destination: Path, refs: tuple[str, ...]) -> dict[str, str]:
    """Read the broker's current head for each candidate ref, read-only."""
    output = _run(
        "ls-remote", "--heads", "hermes-broker",
        *(f"refs/heads/{ref}" for ref in refs),
        cwd=destination, timeout=300,
    )
    heads: dict[str, str] = {}
    for line in output.splitlines():
        fields = line.split()
        if len(fields) == 2 and fields[1].startswith("refs/heads/"):
            heads[fields[1].removeprefix("refs/heads/")] = fields[0]
    return heads


def _workspace_identity(destination: Path, repo: str, branch: str) -> str:
    """Validate the checkout using bounded Git plumbing with all hooks disabled."""
    if destination.is_symlink() or not (destination / ".git").is_dir():
        raise ProtocolError("workspace Git metadata is unavailable")
    ref_path = destination / ".git/refs/heads"
    for part in branch.split("/"):
        ref_path /= part
        if ref_path.is_symlink():
            raise ProtocolError("workspace branch ref must not be a symlink")
    origin = _run("remote", "get-url", "origin", cwd=destination)
    if origin != f"https://scm.bstein.dev/atlas/{repo}.git":
        raise ProtocolError("workspace origin does not match assignment")
    broker = _run("remote", "get-url", "hermes-broker", cwd=destination)
    if broker != _broker_repo(repo):
        raise ProtocolError("workspace broker remote does not match assignment")
    current = _run("symbolic-ref", "--short", "HEAD", cwd=destination)
    if current != branch:
        raise ProtocolError("workspace branch does not match assignment")
    head = _run("rev-parse", "--verify", "HEAD", cwd=destination)
    if not re.fullmatch(r"[0-9a-f]{40,64}", head):
        raise ProtocolError("workspace HEAD is invalid")
    return head


class Boundary:
    """The only process allowed to turn model output into an SCM/result handoff."""

    def __init__(self, key: bytes):
        self.key = key
        self.lock = threading.RLock()

    def verify(self, raw: Any) -> dict[str, Any]:
        return verify_envelope(self.key, raw, expected_kind="assignment")

    def checkout(self, envelope: dict[str, Any]) -> dict[str, Any]:
        _payload, repo, branch, base = _binding(envelope)
        destination = workspace_path(envelope)
        state_path = _state_path(envelope)
        with self.lock:
            if (destination / ".git").exists():
                _workspace_identity(destination, repo, branch)
                state = json.loads(_regular_text(state_path, 16 * 1024))
                baseline = state.get("baseline_sha") if isinstance(state, dict) else None
                if not isinstance(baseline, str) or not re.fullmatch(
                    r"[0-9a-f]{40,64}", baseline
                ):
                    raise ProtocolError("private SCM baseline is unavailable")
                return {"workspace": str(destination), "baseline_sha": baseline}
            if destination.exists() and any(destination.iterdir()):
                raise ProtocolError("workspace is non-empty and unmanaged")
            destination.parent.mkdir(parents=True, exist_ok=True)
            broker = _broker_repo(repo)
            try:
                _run(
                    "clone", "--single-branch", "--branch", branch, "--no-tags",
                    broker, str(destination), timeout=900,
                )
            except RuntimeError as error:
                if destination.exists() and any(destination.iterdir()):
                    raise ProtocolError(
                        "failed branch checkout left unmanaged workspace state"
                    ) from error
                if destination.exists():
                    destination.rmdir()
                _run(
                    "clone", "--single-branch", "--branch", base, "--no-tags",
                    broker, str(destination), timeout=900,
                )
                _run("checkout", "-b", branch, cwd=destination)
            _run(
                "remote", "set-url", "origin",
                f"https://scm.bstein.dev/atlas/{repo}.git", cwd=destination,
            )
            _run("remote", "add", "hermes-broker", broker, cwd=destination)
            _run("config", "user.name", "Hermes Execution Worker", cwd=destination)
            _run("config", "user.email", "hermes@bstein.dev", cwd=destination)
            baseline = _workspace_identity(destination, repo, branch)
            atomic_json(
                state_path,
                {"baseline_sha": baseline, "repo": repo, "branch": branch},
            )
        return {"workspace": str(destination), "baseline_sha": baseline}

    @staticmethod
    def _draft(repo: str, branch: str, base: str, head: str, title: str, body: str) -> str:
        query = urllib.parse.urlencode(
            {"state": "open", "head": f"atlas:{branch}", "limit": 10}
        )
        existing = json.loads(
            scm_broker_client.read(f"/api/v1/repos/atlas/{repo}/pulls?{query}")
        )
        if isinstance(existing, list) and existing:
            return str(existing[0].get("html_url") or "")
        created = json.loads(
            scm_broker_client.create_draft(
                repo,
                base=base,
                head=branch,
                head_sha=head,
                title=title,
                body=body,
            )
        )
        return str(created.get("html_url") or "")

    def submit(self, envelope: dict[str, Any], request: dict[str, Any]) -> dict[str, Any]:
        """Enforce clean/committed state, broker push, and reviewed draft creation."""
        _payload, repo, branch, base = _binding(envelope)
        destination = workspace_path(envelope)
        title = str(request.get("title") or f"Hermes task {envelope['task_id']}").strip()
        body = str(request.get("body") or "Automated Hermes draft.")
        if not title or len(title.encode()) > 512 or len(body.encode()) > 32 * 1024:
            raise ProtocolError("pull-request metadata exceeds the safe limit")
        with self.lock:
            head = _workspace_identity(destination, repo, branch)
            state = json.loads(_regular_text(_state_path(envelope), 16 * 1024))
            baseline = state.get("baseline_sha") if isinstance(state, dict) else ""
            if not isinstance(baseline, str) or not re.fullmatch(
                r"[0-9a-f]{40,64}", baseline
            ):
                raise ProtocolError("private SCM baseline is unavailable")
            status = _run(
                "status", "--porcelain=v1", "--untracked-files=all", cwd=destination
            )
            if status:
                raise ProtocolError("workspace has uncommitted or untracked files")
            candidates = submission_refs(branch, int(envelope["attempt"]), head)
            heads = _remote_heads(destination, candidates)
            # A candidate already at this exact head was published by an earlier
            # attempt or an interrupted one: adopt it instead of pushing again, so
            # replay is a no-op and prior work is never re-derived or dropped.
            published = next(
                (ref for ref in candidates if heads.get(ref) == head), None
            )
            ahead = int(_run("rev-list", "--count", f"{baseline}..{head}", cwd=destination))
            if ahead <= 0 and published is None:
                return {"workspace": str(destination), "branch": branch, "pull_request": ""}
            target = published or next(
                (ref for ref in candidates if ref not in heads), None
            )
            if target is None:
                raise ProtocolError("every reviewed branch name for this attempt is taken")
            if published is None:
                _run(
                    "push", "hermes-broker", f"HEAD:refs/heads/{target}",
                    cwd=destination, timeout=900,
                )
            pull = self._draft(repo, target, base, head, title, body)
        return {"workspace": str(destination), "branch": target, "pull_request": pull}