nisse/nisse-dev/src/nisse_dev/loop/milestones.py

"""The workspace and milestones of a work item (D3, D4, D5).

The harness owns branches: a work item starts from a fresh default branch on
a branch of its own, a milestone is cut by start_milestone, and finishing it
commits only the work, pushes with the push credential, and opens the PR.
Publishing is idempotent under an intent (L8): an open PR for the branch is
adopted, and a PR the operator closed is an answer, never reopened.
"""

import time
from dataclasses import dataclass
from typing import Literal

from nisse_dev.git import GitError, GitRepo, PushCredential
from nisse_dev.gitea import GiteaClient, GiteaError, PullRequest
from nisse_dev.state import StateStore

OPEN_PR = "open_pr"
SET_ASIDE = "set-aside"


@dataclass(frozen=True)
class Published:
    kind: Literal["opened", "adopted", "nothing", "closed", "failed"]
    note: str
    pr: PullRequest | None = None
    left_out: tuple[str, ...] = ()


@dataclass(frozen=True)
class Fixed:
    kind: Literal["pushed", "nothing", "new-pr", "closed", "failed"]
    note: str
    pr: PullRequest | None = None
    branch: str = ""  # a new PR's branch, which later fixes go to


class Milestones:
    def __init__(
        self,
        store: StateStore,
        git: GitRepo,
        gitea: GiteaClient,
        *,
        repo: str,
        remote: str,
        credential: PushCredential | None,
        base: str = "main",
    ) -> None:
        self._store = store
        self._git = git
        self._gitea = gitea
        self._repo = repo
        self._remote = remote
        self._credential = credential
        self._base = base

    # -- a work item's workspace (D3) ------------------------------------------

    def prepare(self, item_branch: str) -> list[str]:
        """A fresh workspace for a work item: fetch, set aside (never
        discard) whatever was left, check out a clean default branch, and put
        the item on a branch of its own. Returns what was set aside."""
        git = self._git
        git.fetch(credential=self._credential)
        stamp = time.strftime("%Y%m%d-%H%M%S")
        notes: list[str] = []
        if git.has_changes():
            aside = f"{SET_ASIDE}/{stamp}/worktree"
            git.checkout_new_branch(aside)
            git.commit_all("Set aside: uncommitted changes left in the workspace")
            notes.append(f"uncommitted changes, as {aside}")
        for branch in git.local_branches():
            if branch == self._base or branch.startswith(f"{SET_ASIDE}/"):
                continue
            git.rename_branch(branch, f"{SET_ASIDE}/{stamp}/{branch}")
            notes.append(f"branch {branch}, as {SET_ASIDE}/{stamp}/{branch}")
        if self._base in git.local_branches() and self._ahead_of_origin():
            aside = f"{SET_ASIDE}/{stamp}/{self._base}"
            git.branch_at(aside, self._base)
            notes.append(f"unpushed commits on {self._base}, as {aside}")
        git.reset_to(self._base, f"origin/{self._base}")
        git.checkout_new_branch(item_branch)
        return notes

    def _ahead_of_origin(self) -> bool:
        current = self._git.current_branch()
        try:
            self._git.checkout_existing_branch(self._base)
            return self._git.commits_since(f"origin/{self._base}") > 0
        finally:
            if current and current != self._base:
                self._git.checkout_existing_branch(current)

    # -- milestones (D4) -------------------------------------------------------

    def start(self, branch: str) -> str:
        """Cut the milestone's branch where the item is; returns its start."""
        self._git.checkout_new_branch(branch)
        return self._git.head_sha()

    def publish(
        self,
        *,
        item_id: str,
        branch: str,
        title: str,
        summary: str,
        issue: int,
    ) -> Published:
        git = self._git
        try:
            if git.current_branch() != branch:
                git.checkout_new_branch(branch)
            git.commit_all(title)
            left_out = tuple(git.left_out())
            if git.commits_since(f"origin/{self._base}") == 0:
                return Published("nothing", "no changes to publish", left_out=left_out)
            existing = self._gitea.find_pull_request(self._repo, branch)
            if existing is not None and existing.state != "open":
                # D4: the operator closed it; that is their answer.
                return Published(
                    "closed",
                    f"PR #{existing.number} for this branch was closed",
                    existing,
                    left_out,
                )
            intent = self._store.begin_intent(
                OPEN_PR,
                {"branch": branch, "title": title, "issue": issue},
                cause=f"publishing {branch}",
                work_item=item_id,
            )
            git.push(self._remote, branch, credential=self._credential)
            existing = self._gitea.find_pull_request(self._repo, branch)
            if existing is not None and existing.state == "open":
                self._store.commit_intent(
                    intent, cause=f"adopted PR #{existing.number}"
                )
                return Published("adopted", "pushed to the open PR", existing, left_out)
            pr = self._gitea.open_pull_request(
                self._repo,
                head=branch,
                base=self._base,
                title=title,
                body=_body(summary, issue, left_out),
            )
            self._store.commit_intent(intent, cause=f"opened PR #{pr.number}")
            return Published("opened", f"opened PR #{pr.number}", pr, left_out)
        except (GitError, GiteaError) as exc:
            return Published("failed", f"publishing failed: {exc}")

    def push_fix(
        self, *, item_id: str, branch: str, pr: int, message: str, attempt: int
    ) -> Fixed:
        """Push a review fix to the PR's branch, then check the PR: one that
        merged before the fix landed gets the fix as a new PR, and one that
        was closed is reported (D5)."""
        git = self._git
        try:
            if git.current_branch() != branch:
                git.checkout_existing_branch(branch)
            sha = git.commit_all(message) or git.head_sha()
            if git.remote_sha(self._remote, branch, credential=self._credential) == sha:
                return Fixed("nothing", "no change was needed")
            git.push(self._remote, branch, credential=self._credential)
            state = self._gitea.pull_request(self._repo, pr)
            if state.merged:
                git.fetch(credential=self._credential)
                if git.is_ancestor(sha, f"origin/{self._base}"):
                    return Fixed("pushed", "pushed, and it was merged with the PR")
                follow = f"{branch}-fix-{attempt}"
                git.checkout_new_branch(follow)
                new = self.publish(
                    item_id=item_id,
                    branch=follow,
                    title=f"Follow-up to #{pr}: {message}",
                    summary=f"PR #{pr} merged before this review fix landed.",
                    issue=0,
                )
                return Fixed("new-pr", new.note, new.pr, follow)
            if state.state != "open":
                return Fixed("closed", f"PR #{pr} was closed", state)
            return Fixed("pushed", f"pushed {sha[:8]}", state)
        except (GitError, GiteaError) as exc:
            return Fixed("failed", f"pushing the fix failed: {exc}")


def _body(summary: str, issue: int, left_out: tuple[str, ...]) -> str:
    parts = [summary]
    if issue:
        parts += ["", f"Closes #{issue}"]
    if left_out:
        parts += ["", "## Left out", "", "Build artifacts, never part of the work:"]
        parts += [f"- `{path}`" for path in left_out]
    return "\n".join(parts)