Publish finished milestones as PRs (B2/A3) — step 6a, part 2
CI / checks (push) Successful in 41s
CI / checks (pull_request) Successful in 38s

Completes the B1 loop past EXECUTING, and answers the question raised
reviewing #6: PR acceptance is a milestone's final gate, so there is
no separate FINISHED state — completion returns to milestone planning
or to IDLE.

- publish.py: MilestonePublisher commits the work on a task-namespaced
  branch, pushes, and opens a PR carrying the approved plan. Intent is
  recorded before the first externally visible act, so a crash
  anywhere in the sequence leaves a durable record and boot
  reconciliation asks Gitea whether the PR exists rather than opening
  a second one
- Nothing to commit is reported, not raised: the executor may
  legitimately conclude there was nothing to change
- control.py: PR_OPEN state with EXECUTING -> PR_OPEN and
  PR_OPEN -> {MILESTONE_PLANNING, IDLE}; operator commands
  'next milestone' and 'project complete'
- A configured publisher always contributes its own reconciler at
  boot, so an interrupted publish is settled against external truth
  without the caller having to remember to register it
- A publish failure gates to the operator rather than retrying into a
  possible duplicate

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Opus 5 committed 2026-09-07 20:43:20 +00:00
1 parent 4025ac7875
commit 3117565928
4 files changed
+546 -3

No files matched your search

+103
View File
@@ -0,0 +1,103 @@
"""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, 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:
# 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")
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 _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)