diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index e28ed92..3215403 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/nisse-dev/src/nisse_dev/control.py @@ -130,6 +130,8 @@ class MilestonePublisher(Protocol): def close_pull_request(self, index: int, note: str) -> None: ... + def published_pull_request(self, task_id: str, milestone: int) -> Any: ... + class PlanBoard(Protocol): """The Gitea side of plan gating; the concrete one is in @@ -323,6 +325,27 @@ class Orchestrator: "restarted while waiting on you: " + pending.get("prompt", "my last question still stands") ) + elif state.state == EXECUTING and (published := self._published_pr()): + # A3 step 4: Gitea shows this milestone's PR open, so it was + # published and only PR_OPEN went unrecorded. Adopt the PR rather + # than ask, and never publish it again. Nothing watched it yet, + # so every comment on it is new. + milestone = int(self._data.get("milestone", 1)) + self._data["pr"] = { + "number": published.number, + "url": published.url, + "milestone": milestone, + "last_comment": self._watch( + "review", published.number, self.handle_review_comment, after=0 + ), + } + self._unwatch("plan") + self._transition(PR_OPEN, cause=f"boot: milestone {milestone} published") + self._transport.send_message( + f"restarted after milestone {milestone} was published: PR " + f"#{published.number} is open ({published.url}); watching it for " + "review comments" + ) elif state.state == PR_OPEN: # Waiting on review is a committed state like a gate (A3 step # 8): resume, and watch the PR again from its cursor. @@ -980,6 +1003,20 @@ class Orchestrator: f"task on {project} abandoned{closed}. Say 'use ' to start another" ) + def _published_pr(self) -> Any: + """The open PR for the current milestone, if Gitea shows one; None + when there is none or Gitea cannot say (then the restart question + is asked as before).""" + task_id = self._data.get("task_id") + if self._publisher is None or not task_id: + return None + try: + return self._publisher.published_pull_request( + str(task_id), int(self._data.get("milestone", 1)) + ) + except GiteaError: + return None + def _return_to_pr_open(self, cause: str) -> None: self._transition(PR_OPEN, cause=cause) pr = self._data.get("pr") or {} diff --git a/nisse-dev/src/nisse_dev/publish.py b/nisse-dev/src/nisse_dev/publish.py index 4ba7c6f..96c764d 100644 --- a/nisse-dev/src/nisse_dev/publish.py +++ b/nisse-dev/src/nisse_dev/publish.py @@ -84,6 +84,19 @@ class MilestonePublisher: cause=f"publishing milestone {milestone}", ) self._git.push(self._remote, branch, token=self._token) + # Idempotent: if a PR for this branch is already open (a crash after + # opening it, or a response lost on the way back), the push above + # has updated it; adopt it rather than fail on a second. + existing = self.published_pull_request(task_id, milestone) + if existing is not None: + self._store.commit_intent( + intent_id, cause=f"adopted open PR #{existing.number}" + ) + return PublishResult( + existing, + f"PR #{existing.number} was already open for this milestone; " + f"pushed to it: {existing.url}", + ) pr = self._client.open_pull_request( self._repo, head=branch, @@ -94,6 +107,17 @@ class MilestonePublisher: self._store.commit_intent(intent_id, cause=f"opened PR #{pr.number}") return PublishResult(pr, f"opened PR #{pr.number}: {pr.url}") + def published_pull_request( + self, task_id: str, milestone: int + ) -> PullRequest | None: + """The open PR for a milestone's branch, if there is one. A closed one + does not count: the operator closed it, so a new one may be opened. + Raises GiteaError when Gitea cannot say.""" + pr = self._client.find_pull_request( + self._repo, self.branch_for(task_id, milestone) + ) + return pr if pr is not None and pr.state == "open" else None + 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.""" diff --git a/nisse-dev/tests/test_publish.py b/nisse-dev/tests/test_publish.py index b01dc0c..2d2bf22 100644 --- a/nisse-dev/tests/test_publish.py +++ b/nisse-dev/tests/test_publish.py @@ -16,6 +16,7 @@ class FakeGitea: def __init__(self, *, fail: bool = False) -> None: self.opened: list[dict[str, Any]] = [] self.existing: list[str] = [] + self.closed: set[str] = set() self._fail = fail def open_pull_request( @@ -33,7 +34,8 @@ class FakeGitea: def find_pull_request(self, repo: str, head: str) -> PullRequest | None: if head in self.existing: - return PullRequest(number=1, state="open", head=head, url="u") + state = "closed" if head in self.closed else "open" + return PullRequest(number=1, state=state, head=head, url="u") return None @@ -212,3 +214,32 @@ def test_a_followup_the_executor_committed_itself_is_pushed( # Nothing new since that push: no second push, and it says so. assert publisher.push_followup(task_id="t", milestone=1, message="m") is None assert len(git.pushed) == 2 + + +def test_publishing_again_adopts_the_open_pr_instead_of_opening_another( + store: StateStore, tmp_path: Path +) -> None: + # E.g. Gitea opened the PR but the response was lost, and the milestone + # was published again: the PR is adopted and updated, never duplicated. + git = FakeGit(tmp_path / "work") + client = FakeGitea() + publisher = build(store, git, client) + publisher.publish(task_id="t", milestone=1, project="demo", plan="p") + (tmp_path / "work" / "more.txt").write_text("more") + result = publisher.publish(task_id="t", milestone=1, project="demo", plan="p") + assert len(client.opened) == 1 + assert result.pull_request is not None and result.pull_request.number == 1 + assert "already open" in result.note + assert git.pushed == ["t/milestone-1", "t/milestone-1"] # updated it + assert store.pending_intents() == [] + + +def test_a_closed_pr_is_not_adopted(store: StateStore, tmp_path: Path) -> None: + git = FakeGit(tmp_path / "work") + client = FakeGitea() + client.existing.append("t/milestone-1") + client.closed.add("t/milestone-1") + publisher = build(store, git, client) + assert publisher.published_pull_request("t", 1) is None + publisher.publish(task_id="t", milestone=1, project="demo", plan="p") + assert len(client.opened) == 1 # the operator closed it; a new one opens diff --git a/nisse-dev/tests/test_quint_counterexamples.py b/nisse-dev/tests/test_quint_counterexamples.py index ff7c820..8555bc8 100644 --- a/nisse-dev/tests/test_quint_counterexamples.py +++ b/nisse-dev/tests/test_quint_counterexamples.py @@ -134,6 +134,12 @@ class Publisher: PullRequest(number=20, state="open", head=branch, url="u"), "opened PR #20" ) + def published_pull_request(self, task_id: str, milestone: int) -> Any: + branch = f"{task_id}/milestone-{milestone}" + if branch in self._gitea.prs: + return PullRequest(number=20, state="open", head=branch, url="u") + return None + def push_followup(self, **kwargs: Any) -> str | None: return "abc12345" @@ -249,7 +255,6 @@ def test_an_afterthought_rejects_the_next_plan(tmp_path: Path) -> None: assert agent.gate().get("rev") == 2 # arch was rev-1; not re-planned -@XFAIL def test_crash_after_opening_the_pr_publishes_again(tmp_path: Path) -> None: """crashAfterOpeningThePrPublishesAgainTest""" gitea = FakeGitea() @@ -261,9 +266,11 @@ def test_crash_after_opening_the_pr_publishes_again(tmp_path: Path) -> None: assert agent.state == EXECUTING # the PR exists; PR_OPEN was never recorded agent = Agent(tmp_path, gitea) # restart: the intent is verified - agent.say("continue", "start milestone 1") - # There is a PR for this milestone: the agent must end up watching it, - # not trying to open another. + # There is a PR for this milestone: the agent adopts it on boot and ends + # up watching it, never trying to open another. + assert agent.state == PR_OPEN + assert "PR #20 is open" in agent.transport.outbound[-1].text + agent.say("continue", "start milestone 1") # no longer means anything assert agent.publisher.attempts == 0 assert agent.state == PR_OPEN @@ -321,3 +328,22 @@ def test_the_harness_reaches_pr_open_without_faults(tmp_path: Path) -> None: assert agent.state == PR_OPEN assert agent.publisher.attempts == 1 _ = (AWAIT_INPUT, MILESTONE_PLANNING) + + +def test_boot_asks_as_before_when_gitea_cannot_say( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea) + agent.say("use demo", "approve", "approve") + agent.publisher.crash_after_opening = True + with pytest.raises(SystemExit): + agent.say("start milestone 1") + + def unreachable(self: Publisher, task_id: str, milestone: int) -> Any: + raise GiteaError("gitea is down") + + monkeypatch.setattr(Publisher, "published_pull_request", unreachable) + agent = Agent(tmp_path, gitea) + assert agent.state == AWAIT_INPUT + assert "restarted mid-EXECUTING" in agent.transport.outbound[-1].text diff --git a/specs/orchestrator.qnt b/specs/orchestrator.qnt index ce0e818..e66a82d 100644 --- a/specs/orchestrator.qnt +++ b/specs/orchestrator.qnt @@ -67,6 +67,7 @@ module orchestrator { prThread: List[PrComment], nextId: int, giteaPr: int -> str, // PR key -> "open" / "closed" + opened: int -> int, // PR key -> how many PRs Gitea ever opened for it openIssues: Set[int], // plan issues still open, by task // -- volatile: lost in a crash -- queue: List[Job], @@ -83,7 +84,6 @@ module orchestrator { answered: Set[int], // ... and answered (pushed, asked, or reported) badApproval: bool, staleFeedback: bool, - dupPublish: bool, } // ---- helpers ------------------------------------------------------------ @@ -97,6 +97,11 @@ module orchestrator { pure def prIds(s: Sys, key: int): int = s.prThread.foldl(0, (acc, c) => if (c.key == key and c.id > acc) c.id else acc) pure def max(a: int, b: int): int = if (a > b) a else b + // Gitea opens a PR for a branch. + pure def openPr(s: Sys, key: int): Sys = + { ...s, giteaPr: s.giteaPr.put(key, "open"), + opened: s.opened.put(key, (if (s.opened.keys().contains(key)) s.opened.get(key) else 0) + 1) } + pure def queued(s: Sys, src: str): bool = s.queue.foldl(false, (acc, j) => acc or j.src == src) // _await_input @@ -260,6 +265,11 @@ module orchestrator { awaitInput(s0, { kind: "boot", g: "", interrupted: s.st }) else if (s.st == "AWAIT_INPUT" and s.pending.kind != "") s0 + else if (s.st == "EXECUTING" and giteaUp and prState(s, prKeyOf(s)) == "open") + // A3 step 4: the milestone's PR is open, so it was published and only + // PR_OPEN went unrecorded. Adopt it; nothing watched it, so from 0. + { ...s0, st: "PR_OPEN", prOn: true, prKey: prKeyOf(s), prLast: 0, + prWatch: { on: true, floor: 0 }, planWatch: OFF } else if (s.st == "PR_OPEN") { ...s0, prWatch: if (s.prOn) { on: true, floor: s.prLast } else OFF } else if (s.st == "IDLE" or (SIGNOFF.contains(s.st) and s.gate.on)) @@ -278,11 +288,11 @@ module orchestrator { boardOn: false, boardPosted: false, boardLast: 0, intents: Set(), planThread: List(), prThread: List(), nextId: 1, - giteaPr: Map(), openIssues: Set(), + giteaPr: Map(), opened: Map(), openIssues: Set(), queue: List(), work: "none", workGate: "", workCid: 0, pubStage: 0, cancel: false, planWatch: OFF, prWatch: OFF, shown: NO_PLAN, taken: Set(), answered: Set(), - badApproval: false, staleFeedback: false, dupPublish: false, + badApproval: false, staleFeedback: false, } // Each action has a deterministic form taking its choices as parameters, @@ -372,8 +382,9 @@ module orchestrator { // Publishing, one step at a time, so a crash can land between them // (MilestonePublisher.publish and control._publish_milestone): - // 0 commit, begin_intent; 1 push, open the PR; 2 commit_intent; - // 3 record the PR, watch it, PR_OPEN. + // 0 commit, begin_intent; 1 push, then adopt an open PR for the branch or + // open one ("lost": Gitea opened it but the response never came back); + // 2 commit_intent; 3 record the PR, watch it, PR_OPEN. action publishStep(outcome: str): bool = { val key = prKeyOf(s) all { @@ -389,10 +400,11 @@ module orchestrator { else if (outcome == "fail" or outcome == "nochanges") awaitInput(endWork(s), { kind: "milestone_stopped", g: "", interrupted: "" }) else if (prState(s, key) == "open") - // Gitea refuses a second PR for the branch: a GiteaError. - awaitInput(endWork({ ...s, dupPublish: true }), + { ...s, pubStage: 2 } // adopted, never a second + else if (outcome == "lost") + awaitInput(endWork(openPr(s, key)), { kind: "milestone_stopped", g: "", interrupted: "" }) - else { ...s, giteaPr: s.giteaPr.put(key, "open"), pubStage: 2 } + else { ...openPr(s, key), pubStage: 2 } else if (s.pubStage == 2) { ...s, intents: s.intents.exclude(Set(key)), pubStage: 3 } else @@ -423,7 +435,7 @@ module orchestrator { nondet up = Set(true, false).oneOf() nondet planOutcome = Set("submit", "ask", "error", "stopped", "exception").oneOf() nondet workOutcome = Set("completed", "stopped", "escalate", "exception").oneOf() - nondet pubOutcome = Set("ok", "nochanges", "fail", "exception").oneOf() + nondet pubOutcome = Set("ok", "nochanges", "fail", "lost", "exception").oneOf() any { send(kind), commentPlan(ckind), commentPr, pollPlan, pollPr, runJob(up), finishPlanning(planOutcome, up), finishExec(workOutcome), publishStep(pubOutcome), @@ -449,8 +461,8 @@ module orchestrator { // Rejection feedback only ever lands on the plan it was written about. val feedback_only_on_what_was_seen = not(s.staleFeedback) - // Never a second attempt to open a PR for a milestone that has one. - val one_pr_per_milestone = not(s.dupPublish) + // Never more than one PR opened for a milestone. + val one_pr_per_milestone = s.opened.keys().forall(k => s.opened.get(k) <= 1) // A review comment the agent took up is answered, or is being worked on. val review_comments_answered = diff --git a/specs/orchestrator_scenarios.qnt b/specs/orchestrator_scenarios.qnt index 9e0e64b..8348cd1 100644 --- a/specs/orchestrator_scenarios.qnt +++ b/specs/orchestrator_scenarios.qnt @@ -1,7 +1,8 @@ // Counterexamples found in orchestrator.qnt, written out as short traces. // Each ends by expecting the property it breaks to be false; once the design // is fixed, the expectation is flipped and the trace kept as a regression. -// Fixed: 1-4 (replies bind to the revision up when they arrived). +// Fixed: 1-4 (replies bind to the revision up when they arrived), 5 (boot +// and publishing adopt an open PR). // // Run: quint test specs/orchestrator_scenarios.qnt (runs every *Test) module orchestrator_scenarios { @@ -77,22 +78,32 @@ module orchestrator_scenarios { .expect(s.st == "AWAIT_MILESTONE_SIGNOFF" and feedback_only_on_what_was_seen) // 5. A crash after the PR is opened but before PR_OPEN is recorded. - // Boot finds the intent's PR and marks it done, but the state still - // says EXECUTING, so it asks the restart question. 'continue' returns - // to the approved plan, and starting it publishes again: Gitea refuses - // a second PR for the branch, and there is no path back to PR_OPEN. + // Boot found the intent's PR and marked it done, but the state still + // said EXECUTING, so it asked the restart question; 'continue' and + // 'start' then published again, which Gitea refused, with no path back + // to PR_OPEN. Fixed: boot sees the open PR and adopts it. run crashAfterOpeningThePrPublishesAgainTest = toMilestoneApproved .then(send("start")).then(runJob(true)) .then(finishExec("completed")) .then(publishStep("ok")) // begin_intent .then(publishStep("ok")) // push, open the PR - .then(crash(true)) // intent verified and committed; state EXECUTING - .then(send("continue")).then(runJob(true)) + .then(crash(true)) // adopted on boot + .expect(s.st == "PR_OPEN" and s.prOn and one_pr_per_milestone) + + // 5b. Gitea opens the PR but the response is lost; the operator starts + // the milestone again. Publishing adopts the open PR. + run lostResponseThenRetryAdoptsThePrTest = + toMilestoneApproved + .then(send("start")).then(runJob(true)) + .then(finishExec("completed")) + .then(publishStep("ok")).then(publishStep("lost")) + .then(send("continue")).then(runJob(true)) // back to the approved plan .then(send("start")).then(runJob(true)) .then(finishExec("completed")) .then(publishStep("ok")).then(publishStep("ok")) - .expect(s.st == "AWAIT_INPUT" and not(one_pr_per_milestone)) + .then(publishStep("ok")).then(publishStep("ok")) + .expect(s.st == "PR_OPEN" and one_pr_per_milestone) // 6. A crash while the executor answers a review comment. The comment's // cursor was saved before the work started (at most once), and a