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

"""Publishing a finished milestone as a Gitea PR (spec B2, A2/A3).

The ordering here is the whole point. Intent is recorded *before* the first
externally visible act, so a crash anywhere in the sequence leaves a durable
record of what was in flight. Boot reconciliation then asks Gitea whether the
PR exists rather than repeating the push-and-open — the spec's own example of
an idempotent transition reconciled against external truth.
"""

from dataclasses import dataclass

from nisse_dev.git import GitRepo, authenticated_url
from nisse_dev.gitea import (
    GiteaClient,
    GiteaError,
    PullRequest,
    pull_request_reconciler,
)
from nisse_dev.reconcile import Reconciler
from nisse_dev.state import StateStore

OPEN_PR_INTENT = "open_pr"


@dataclass(frozen=True)
class PublishResult:
    pull_request: PullRequest | None
    note: str


class MilestonePublisher:
    def __init__(
        self,
        store: StateStore,
        client: GiteaClient,
        repo: str,
        git_repo: GitRepo,
        *,
        base_url: str,
        account: str,
        token: str,
        base_branch: str = "main",
    ) -> None:
        self._store = store
        self._client = client
        self._repo = repo
        self._git = git_repo
        self._base_branch = base_branch
        self._remote = authenticated_url(base_url, repo, account, token)
        self._token = token

    def reconcilers(self) -> dict[str, Reconciler]:
        return {OPEN_PR_INTENT: pull_request_reconciler(self._client, self._repo)}

    def branch_for(self, task_id: str, milestone: int) -> str:
        # Namespaced by task so re-running a project against the same repo
        # cannot collide with an earlier run's branch.
        return f"{task_id}/milestone-{milestone}"

    def publish(
        self,
        *,
        task_id: str,
        milestone: int,
        project: str,
        plan: str,
        service_url: str | None = None,
    ) -> PublishResult:
        branch = self.branch_for(task_id, milestone)
        if self._git.current_branch() != branch:
            self._git.checkout_new_branch(branch)
        sha = self._git.commit_all(f"Milestone {milestone}: {project}")
        if sha is None:
            if self._git.commits_since(f"origin/{self._base_branch}") == 0:
                # Not an error: the executor may have concluded there was
                # nothing to change. Report it and let the operator decide.
                return PublishResult(None, "no changes to publish")
            # The executor committed its work itself; publish that.
            sha = self._git.head_sha()

        intent_id = self._store.begin_intent(
            OPEN_PR_INTENT,
            {"branch": branch, "milestone": milestone, "sha": sha},
            cause=f"publishing milestone {milestone}",
        )
        self._git.push(self._remote, branch, token=self._token)
        pr = self._client.open_pull_request(
            self._repo,
            head=branch,
            base=self._base_branch,
            title=f"Milestone {milestone}: {project}",
            body=_pr_body(milestone, plan, service_url),
        )
        self._store.commit_intent(intent_id, cause=f"opened PR #{pr.number}")
        return PublishResult(pr, f"opened PR #{pr.number}: {pr.url}")

    def comment(self, index: int, body: str) -> None:
        """Reply on the PR. Courtesy only — the pushed commit is the real
        answer — so a failure here must not derail the review loop."""
        try:
            self._client.post_comment(self._repo, index, body)
        except GiteaError:
            pass

    def close_pull_request(self, index: int, note: str) -> None:
        """Close a milestone PR with a note saying why. The branch is kept,
        so the work can be reopened in Gitea."""
        self._client.post_comment(self._repo, index, note)
        self._client.close_pull_request(self._repo, index)

    def push_followup(
        self, *, task_id: str, milestone: int, message: str
    ) -> str | None:
        """Push a fix commit onto an open milestone's branch.

        No intent is recorded: the PR already exists, so a crash mid-push
        costs a repeated push of the same commit, which is harmless. That is
        the difference from publish(), where the side effect is a new PR.
        """
        branch = self.branch_for(task_id, milestone)
        if self._git.current_branch() != branch:
            self._git.checkout_new_branch(branch)
        sha = self._git.commit_all(message)
        if sha is None:
            # The executor may have committed the fix itself.
            sha = self._git.head_sha()
            pushed = self._git.remote_sha(self._remote, branch, token=self._token)
            if sha == pushed:
                return None
        self._git.push(self._remote, branch, token=self._token)
        return sha


def _pr_body(milestone: int, plan: str, service_url: str | None) -> str:
    parts = [
        f"Milestone {milestone}, implemented by nisse-dev against the approved plan.",
        "",
        "## Approved plan",
        "",
        plan,
    ]
    if service_url:
        # B2: a runnable test service's address belongs in the PR description
        # so a reviewer can exercise it. Nothing reports one yet; the browser
        # tool (A6) is what will.
        parts += ["", "## Running test service", "", service_url]
    return "\n".join(parts)