nisse/nisse-dev/src/nisse_dev/git.py

"""Git operations in the work item's workspace, on its build VM (D4, L13).

Every command runs on the build VM over SSH. Remotes are plain URLs with no
credential in them, so nothing secret is ever written to the clone's
configuration (#89). Operations that reach the remote (clone, fetch, push,
ls-remote) are handed the push credential for that one operation, and the
model's commands never receive it:

- an HTTPS token (the default: HTTPS reaches the forge from everywhere) goes
  to git through a credential helper that names environment variables, never
  values; the values are that one command's secrets, so they are never on a
  command line and are redacted from its output;
- an SSH key is fed over stdin to a short-lived ``ssh-agent`` on the VM, gone
  when the command ends.

Pull requests are opened from the agent's own host with its Gitea API token
(publish.py), which never reaches the VM.
"""

import shlex
from dataclasses import dataclass, field
from pathlib import Path

from nisse_core.execution.ssh import HostUnavailable, SSHHost

TIMEOUT_S = 120

# ssh for git on the VM: no prompts, and trust Gitea's host key on first use.
_GIT_SSH = "ssh -o BatchMode=yes -o StrictHostKeyChecking=accept-new"


class GitError(RuntimeError):
    pass


# Build artifacts and caches that are never the work (D4, #80). Kept out of
# every commit the harness makes, tracked or not, and excluded in the clone
# so they don't even show as changes. Conservative on purpose: a pattern here
# must never match a project's real source.
ARTIFACT_PATTERNS = (
    "__pycache__/",
    "*.py[cod]",
    ".pytest_cache/",
    ".mypy_cache/",
    ".ruff_cache/",
    ".tox/",
    ".nox/",
    "*.egg-info/",
    ".coverage",
    "htmlcov/",
    ".venv/",
    "node_modules/",
    ".DS_Store",
)
_EXCLUDES_MARK = "# nisse-dev: build artifacts are never the work (D4)"


def _pathspec(pattern: str) -> str:
    """A pathspec leaving ``pattern`` out wherever it occurs in the tree."""
    glob = pattern.rstrip("/")
    tail = "/**" if pattern.endswith("/") else ""
    return f":(exclude,glob)**/{glob}{tail}"


@dataclass(frozen=True)
class HTTPSToken:
    """A push-only account's token, for HTTPS remotes (D4)."""

    username: str
    # Never in a repr: a failing test or a log line would print it.
    token: str = field(repr=False)


@dataclass(frozen=True)
class SSHKey:
    """A deploy key's private half, for SSH remotes (D4)."""

    private_key: str = field(repr=False)


PushCredential = HTTPSToken | SSHKey

# git asks the helper for credentials; it answers from the environment, so
# the command line names variables, never values.
_HELPER = (
    '!f() { test "$1" = get || return 0; '
    'echo "username=$NISSE_PUSH_USER"; echo "password=$NISSE_PUSH_TOKEN"; }; f'
)


def _with_key(command: str, known_hosts: str | None) -> str:
    """Run ``command`` with the key on stdin loaded into a throwaway agent."""
    git_ssh = _GIT_SSH
    if known_hosts is not None:
        git_ssh += f" -o UserKnownHostsFile={shlex.quote(known_hosts)}"
    return (
        'eval "$(ssh-agent -s)" > /dev/null && '
        "ssh-add -q - 2> /dev/null && "
        f"GIT_SSH_COMMAND={shlex.quote(git_ssh)} {command}; "
        "rc=$?; ssh-agent -k > /dev/null 2>&1; exit $rc"
    )


@dataclass(frozen=True)
class GitRepo:
    host: SSHHost
    root: Path
    # Where git's ssh on the VM records the forge's host key; the default is
    # the VM user's own known_hosts.
    known_hosts: str | None = None

    def _git(
        self,
        *args: str,
        credential: PushCredential | None = None,
        cwd: str | None = None,
        raw: bool = False,
    ) -> str:
        quoted = " ".join(shlex.quote(a) for a in args)
        command, stdin, secrets = f"git {quoted}", None, {}
        if isinstance(credential, SSHKey):
            command = _with_key(command, self.known_hosts)
            stdin = credential.private_key
        elif isinstance(credential, HTTPSToken):
            # The empty helper first clears any configured on the VM.
            command = (
                "GIT_TERMINAL_PROMPT=0 git -c credential.helper= "
                f"-c credential.helper={shlex.quote(_HELPER)} {quoted}"
            )
            secrets = {
                "NISSE_PUSH_USER": credential.username,
                "NISSE_PUSH_TOKEN": credential.token,
            }
        try:
            result = self.host.execute(
                command,
                cwd=cwd or str(self.root),
                stdin=stdin,
                secrets=secrets,
                timeout_s=TIMEOUT_S,
            )
        except HostUnavailable as exc:
            raise GitError(f"git {args[0]} failed: {exc}") from exc
        if result.timed_out:
            raise GitError(f"git {args[0]} timed out after {TIMEOUT_S}s")
        if result.exit_code != 0:
            raise GitError(
                f"git {args[0]} failed: "
                f"{result.stderr.strip() or result.stdout.strip()}"
            )
        # Raw for output whose leading whitespace means something (porcelain).
        return result.stdout if raw else result.stdout.strip()

    # -- inspection -------------------------------------------------------

    def current_branch(self) -> str:
        # --show-current rather than rev-parse: it answers correctly on a
        # freshly initialised repo, where HEAD points at an unborn branch.
        return self._git("branch", "--show-current")

    def head_sha(self) -> str:
        return self._git("rev-parse", "HEAD")

    def has_changes(self) -> bool:
        return bool(self._git("status", "--porcelain"))

    def commits_since(self, ref: str) -> int:
        """How many commits HEAD has that ref lacks. The executor may commit
        its own work, so a clean tree does not mean nothing to publish."""
        return int(self._git("rev-list", "--count", f"{ref}..HEAD"))

    def remote_sha(
        self, remote_url: str, branch: str, *, credential: PushCredential | None = None
    ) -> str | None:
        """What the remote's branch points at, or None if it has no such
        branch. Pushes go by URL, so there is no tracking ref to read."""
        out = self._git(
            "ls-remote", remote_url, f"refs/heads/{branch}", credential=credential
        )
        return out.split()[0] if out else None

    def exists(self) -> bool:
        """Whether the workspace is already a git repository."""
        try:
            result = self.host.execute(
                f"test -d {shlex.quote(str(self.root / '.git'))}", timeout_s=TIMEOUT_S
            )
        except HostUnavailable as exc:
            raise GitError(f"could not check the workspace: {exc}") from exc
        return result.exit_code == 0

    # -- mutation ---------------------------------------------------------

    def init(self, *, default_branch: str = "main") -> None:
        self.host.execute(f"mkdir -p -- {shlex.quote(str(self.root))}")
        self._git("init", "-q", "-b", default_branch)

    def clone(
        self, remote_url: str, *, credential: PushCredential | None = None
    ) -> None:
        """Clone into the workspace root. The remote URL carries no
        credential, so none is stored in the clone's configuration."""
        parent = str(self.root.parent)
        self.host.execute(f"mkdir -p -- {shlex.quote(parent)}")
        self._git(
            "clone", "-q", remote_url, str(self.root), credential=credential, cwd=parent
        )

    def configure_identity(self, name: str, email: str) -> None:
        self._git("config", "user.name", name)
        self._git("config", "user.email", email)

    def checkout_new_branch(self, name: str, start: str | None = None) -> None:
        """A new branch, at ``start`` if given. Uncommitted changes come
        along; git refuses if one would conflict with ``start``."""
        self._git("checkout", "-q", "-b", name, *([start] if start else []))

    def checkout_existing_branch(self, name: str) -> None:
        self._git("checkout", "-q", name)

    def branch_exists(self, name: str) -> bool:
        try:
            self._git("rev-parse", "--verify", "--quiet", f"refs/heads/{name}")
        except GitError:
            return False
        return True

    def install_excludes(self) -> None:
        """List the build artifacts in the clone's own exclude file, so they
        never appear as changes. Local to the clone: nothing in the project
        changes. Idempotent."""
        path = shlex.quote(str(self.root / ".git" / "info" / "exclude"))
        block = "\n".join([_EXCLUDES_MARK, *ARTIFACT_PATTERNS])
        result = self.host.execute(
            f'mkdir -p -- "$(dirname -- {path})" && touch -- {path} && '
            f"grep -qxF {shlex.quote(_EXCLUDES_MARK)} {path} || "
            f"cat >> {path}",
            stdin=block,
            timeout_s=TIMEOUT_S,
        )
        if not result.ok:
            raise GitError(f"could not set excludes: {result.stderr.strip()}")

    def commit_all(self, message: str) -> str | None:
        """Commit the work: every change except build artifacts, which are
        left out even when the project tracks them (#80). Returns the new
        commit sha, or None when there was nothing to commit — an empty
        milestone is a fact to report, not an error."""
        if not self.has_changes():
            return None
        self._git("add", "-A", "--", ".", *(_pathspec(p) for p in ARTIFACT_PATTERNS))
        staged = self._git("diff", "--cached", "--name-only")
        if not staged:
            return None
        self._git("commit", "-q", "-m", message)
        return self.head_sha()

    def changed_paths(self, since: str) -> list[str]:
        """Every path the work changes relative to ``since``, committed or
        not, new files included; build artifacts aren't work (#80)."""
        tracked = self._git("diff", "--name-only", since).splitlines()
        new = self._git("ls-files", "--others", "--exclude-standard").splitlines()
        return sorted({p for p in [*tracked, *new] if p})

    def left_out(self) -> list[str]:
        """Changes not in the last commit: build artifacts the harness left
        out, to be reported rather than silently dropped (D4)."""
        out = self._git("status", "--porcelain", "--untracked-files=all", raw=True)
        return sorted(line[3:] for line in out.splitlines() if line)

    # -- a fresh workspace (D3) ----------------------------------------------

    def fetch(self, *, credential: PushCredential | None = None) -> None:
        """Bring origin's branches up to date, pruning deleted ones. The
        remote's URL in the clone carries no credential; this one fetch is
        handed it."""
        self._git("fetch", "-q", "--prune", "origin", credential=credential)

    def local_branches(self) -> list[str]:
        out = self._git("for-each-ref", "--format=%(refname:short)", "refs/heads")
        return [line for line in out.splitlines() if line]

    def rename_branch(self, old: str, new: str) -> None:
        self._git("branch", "-m", old, new)

    def branch_at(self, name: str, ref: str) -> None:
        self._git("branch", name, ref)

    def reset_to(self, branch: str, start: str) -> None:
        """Check out ``branch`` at ``start``, discarding what the work tree
        had (anything worth keeping is set aside first), then remove every
        untracked and ignored file: build artifacts never survive (D3)."""
        self._git("checkout", "-q", "-f", "-B", branch, start)
        self._git("clean", "-q", "-f", "-d", "-x")

    def is_ancestor(self, commit: str, ref: str) -> bool:
        try:
            self._git("merge-base", "--is-ancestor", commit, ref)
        except GitError:
            return False
        return True

    def push(
        self, remote_url: str, branch: str, *, credential: PushCredential | None = None
    ) -> None:
        # By URL, per invocation, rather than a configured remote with a
        # credential in it.
        self._git(
            "push", "-q", remote_url, f"HEAD:refs/heads/{branch}", credential=credential
        )

    def push_commit(
        self,
        remote_url: str,
        sha: str,
        branch: str,
        *,
        credential: PushCredential | None = None,
    ) -> None:
        """Point the remote's ``branch`` at ``sha``, whatever it was."""
        self._git(
            "push",
            "-q",
            "-f",
            remote_url,
            f"{sha}:refs/heads/{branch}",
            credential=credential,
        )

    def stash_all(self, message: str) -> str | None:
        """Take every uncommitted change, untracked files included and build
        artifacts not, off the work tree as a stash commit; returns its sha,
        or None when there was nothing. The commit is kept by pushing it."""
        if not self.has_changes():
            return None
        self._git("stash", "push", "-q", "-u", "-m", message)
        sha = self._git("rev-parse", "stash@{0}")
        self._git("stash", "drop", "-q")
        return sha

    def apply_stash(self, sha: str) -> None:
        """Put a stash commit's changes back on the work tree, uncommitted."""
        self._git("stash", "apply", "-q", sha)

    def has_commit(self, sha: str) -> bool:
        try:
            self._git("cat-file", "-e", f"{sha}^{{commit}}")
        except GitError:
            return False
        return True

    def discard_branch(self, branch: str, *, back_to: str) -> None:
        """Throw away a branch and everything on it.

        Used when a delegation aborts: L9 gives a delegation the same
        atomicity that reconciliation gives an interrupted transition, and
        half-finished work must not survive to be built upon.
        """
        self._git("checkout", "-q", "--force", back_to)
        self._git("branch", "-q", "-D", branch)

    def delete_remote_branch(
        self, remote_url: str, branch: str, *, credential: PushCredential | None = None
    ) -> None:
        self._git(
            "push", "-q", remote_url, f":refs/heads/{branch}", credential=credential
        )