diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index 3215403..a0dfdef 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/nisse-dev/src/nisse_dev/control.py @@ -351,10 +351,16 @@ class Orchestrator: # 8): resume, and watch the PR again from its cursor. pr = self._data.get("pr") or {} self._watch_pr(pr) - self._transport.send_message( - f"restarted with PR #{pr.get('number')} open; still watching it " - "for review comments" - ) + if pr.get("reviewing"): + self._transport.send_message( + f"restarted while answering review comment {pr['reviewing']} " + f"on PR #{pr.get('number')}; answering it again" + ) + else: + self._transport.send_message( + f"restarted with PR #{pr.get('number')} open; still watching " + "it for review comments" + ) elif state.state == IDLE or ( state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF) and gate ): @@ -532,7 +538,7 @@ class Orchestrator: return self._handle_gate_reply(msg, text, about) elif current == PR_OPEN: - self._handle_pr_open(text) + self._handle_pr_open(text, msg.sender) elif current == AWAIT_INPUT: self._handle_answer(text) else: @@ -869,13 +875,23 @@ class Orchestrator: pr = self._data.get("pr") or {} milestone = int(pr.get("milestone", self._data.get("milestone", 1))) + # Taken at most once, but answered at least once: until the executor + # returns, the comment is marked, and a restart delivers it again. + pr["reviewing"] = comment.id + self._save(f"answering review comment {comment.id}") # The comment enters the transcript, so the executor sees review # feedback as conversation rather than as an out-of-band instruction. self._store.append_event( UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}") ) - with self._long_work(): - outcome = self._executor.run(milestone) + try: + with self._long_work(): + outcome = self._executor.run(milestone) + except Exception: + # Reported to the operator by _settled: answered, if not well. + self._done_reviewing(pr) + raise + self._done_reviewing(pr) if outcome.status == "stopped": self._halt_for_stop( what=f"the response to review comment {comment.id}", @@ -913,6 +929,13 @@ class Orchestrator: if pr.get("number"): self._publisher.comment(int(pr["number"]), note) + def _done_reviewing(self, pr: dict[str, Any]) -> None: + """The executor returned: the comment is being answered one way or + another. Deliberately not in a `finally`: if the process dies + mid-response, the mark must survive so the comment is redelivered.""" + if pr.pop("reviewing", None) is not None: + self._save("review comment answered") + def _publish_milestone(self, milestone: int) -> None: if self._publisher is None: # No PR to wait on, but PR_OPEN's commands still lead on. @@ -988,11 +1011,8 @@ class Orchestrator: for intent in self._store.pending_intents(): self._store.abort_intent(intent.id, cause=note) problems: list[str] = [] - if self._board is not None and board_state.get("issue"): - try: - self._board.close(int(board_state["issue"]), note) - except GiteaError as exc: - problems.append(f"plan issue #{board_state['issue']}: {exc}") + if problem := self._close_plan_issue(board_state, note): + problems.append(problem) if self._publisher is not None and pr.get("number"): try: self._publisher.close_pull_request(int(pr["number"]), note) @@ -1017,6 +1037,17 @@ class Orchestrator: except GiteaError: return None + def _close_plan_issue(self, board_state: dict[str, Any], note: str) -> str | None: + """Close a finished task's plan issue with a note; what went wrong, + if anything, for the operator's message.""" + if self._board is None or not board_state.get("issue"): + return None + try: + self._board.close(int(board_state["issue"]), note) + except GiteaError as exc: + return f"plan issue #{board_state['issue']}: {exc}" + return None + def _return_to_pr_open(self, cause: str) -> None: self._transition(PR_OPEN, cause=cause) pr = self._data.get("pr") or {} @@ -1030,14 +1061,18 @@ class Orchestrator: """Watch the open PR from its recorded cursor, catching up on what arrived while it was not watched.""" if pr.get("number"): + after = pr.get("last_comment") + reviewing = pr.get("reviewing") + if reviewing: + # Its answer never finished: resume just before it. + after = min( + int(after if after is not None else reviewing), reviewing - 1 + ) self._watch( - "review", - int(pr["number"]), - self.handle_review_comment, - after=pr.get("last_comment"), + "review", int(pr["number"]), self.handle_review_comment, after=after ) - def _handle_pr_open(self, text: str) -> None: + def _handle_pr_open(self, text: str, sender: str = "the operator") -> None: command = command_form(text) if command == "next milestone": self._unwatch("review") @@ -1047,13 +1082,18 @@ class Orchestrator: self._transition(MILESTONE_PLANNING, cause="operator: next milestone") self._run_planning_turn(f"milestone-{self._data['milestone']}") elif command == "project complete": + board_state = self._data.get("board") or {} self.stop_watching() self._data = {**_fresh_data()} self._project = None self._store.oplog.task_id = None self._transition(IDLE, cause="operator: project complete") + # IDLE first, as for abandon; the PR is the operator's to merge. + problem = self._close_plan_issue(board_state, f"completed by {sender}") self._transport.send_message( - "project closed; say 'use ' to start another" + "project closed" + + (f"; could not close {problem}" if problem else "") + + ". Say 'use ' to start another" ) else: pr = self._data.get("pr") or {} diff --git a/nisse-dev/tests/test_quint_counterexamples.py b/nisse-dev/tests/test_quint_counterexamples.py index 8555bc8..bdca1c7 100644 --- a/nisse-dev/tests/test_quint_counterexamples.py +++ b/nisse-dev/tests/test_quint_counterexamples.py @@ -1,6 +1,6 @@ """The counterexamples found by specs/orchestrator.qnt, replayed against the -real orchestrator. Each is marked xfail until its fix lands, and names the -scenario in specs/orchestrator_scenarios.qnt it reproduces. +real orchestrator. Each names the scenario in specs/orchestrator_scenarios.qnt +it reproduces; all seven are fixed, and the tests stay as regressions. The model is a hand-written reading of control.py; reproducing each trace here is what shows the flaw is in the design, not in the model. @@ -194,9 +194,6 @@ class Agent: self.transport.receive("don", text) -XFAIL = pytest.mark.xfail(strict=True, reason="counterexample; fix pending") - - def test_double_approve_approves_an_unseen_plan(tmp_path: Path) -> None: """doubleApproveApprovesAnUnseenPlanTest""" worker = QueueingWorker() @@ -275,7 +272,6 @@ def test_crash_after_opening_the_pr_publishes_again(tmp_path: Path) -> None: assert agent.state == PR_OPEN -@XFAIL def test_crash_during_a_review_loses_the_comment(tmp_path: Path) -> None: """crashDuringAReviewLosesTheCommentTest""" gitea = FakeGitea() @@ -296,17 +292,15 @@ def test_crash_during_a_review_loses_the_comment(tmp_path: Path) -> None: agent = Agent(tmp_path, gitea) # restart assert agent.state == PR_OPEN - # The comment was taken but never answered: after the restart it must - # be handled again, or the operator told it was dropped. - delivered = agent.pollers[20].poll_once(20) - told = any( - "please add a test" in m.text or "review" in m.text - for m in agent.transport.outbound[1:] - ) - assert delivered or told + # The comment was taken but never answered: after the restart the + # operator is told, and it is delivered and answered again. + assert "answering it again" in agent.transport.outbound[-1].text + [delivered] = agent.pollers[20].poll_once(20) + assert delivered.body == "please add a test" + assert "pushed" in agent.transport.outbound[-1].text + assert "reviewing" not in agent.store.get_state().data["pr"] -@XFAIL def test_project_complete_leaves_the_plan_issue_open(tmp_path: Path) -> None: """projectCompleteLeavesThePlanIssueOpenTest""" closed: list[int] = [] @@ -347,3 +341,60 @@ def test_boot_asks_as_before_when_gitea_cannot_say( agent = Agent(tmp_path, gitea) assert agent.state == AWAIT_INPUT assert "restarted mid-EXECUTING" in agent.transport.outbound[-1].text + + +class Crash(BaseException): + """The process dying: no handler runs.""" + + +def crash_mid_review(agent: Agent, gitea: FakeGitea) -> None: + def dies(milestone: int) -> ExecutorOutcome: + raise Crash() + + agent.executor.run = dies # type: ignore[method-assign] + with pytest.raises(Crash): + agent.pollers[20].poll_once(20) + + +def test_a_second_crash_while_answering_again_still_redelivers(tmp_path: Path) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea) + agent.say("use demo", "approve", "approve", "start milestone 1") + gitea.add(20, "don", "please add a test") + crash_mid_review(agent, gitea) + agent = Agent(tmp_path, gitea) + crash_mid_review(agent, gitea) # dies again while answering it again + agent = Agent(tmp_path, gitea) + [delivered] = agent.pollers[20].poll_once(20) + assert delivered.body == "please add a test" + + +def test_an_executor_error_answers_the_comment_so_it_is_not_redelivered( + tmp_path: Path, +) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea) + agent.say("use demo", "approve", "approve", "start milestone 1") + gitea.add(20, "don", "please add a test") + + def fails(milestone: int) -> ExecutorOutcome: + raise RuntimeError("bug in the executor") + + agent.executor.run = fails # type: ignore[method-assign] + agent.pollers[20].poll_once(20) + assert "could not handle a review comment" in agent.transport.outbound[-1].text + agent = Agent(tmp_path, gitea) # restart: nothing to answer again + assert "still watching" in agent.transport.outbound[-1].text + assert agent.pollers[20].poll_once(20) == [] + + +def test_project_complete_still_completes_when_gitea_is_down(tmp_path: Path) -> None: + agent = Agent(tmp_path, FakeGitea()) + + def down(issue: int, note: str) -> None: + raise GiteaError("gitea is down") + + agent.orch._board.close = down # type: ignore[attr-defined] + agent.say("use demo", "approve", "approve", "start milestone 1", "project complete") + assert agent.state == IDLE + assert "could not close plan issue #10" in agent.transport.outbound[-1].text diff --git a/specs/orchestrator.qnt b/specs/orchestrator.qnt index e66a82d..fcf9385 100644 --- a/specs/orchestrator.qnt +++ b/specs/orchestrator.qnt @@ -60,6 +60,7 @@ module orchestrator { ms: int, pending: Pending, prOn: bool, prKey: int, prLast: int, // _data["pr"] + reviewing: int, // pr["reviewing"]: the comment being answered, 0 if none boardOn: bool, boardPosted: bool, boardLast: int, // _data["board"] intents: Set[int], // pending open_pr intents, by PR key // -- Gitea -- @@ -127,7 +128,7 @@ module orchestrator { startPlanning( { ...s, tid: t, task: true, gate: NO_GATE, revs: Map("arch" -> 0, "ms1" -> 0, "ms2" -> 0), lastRev: 0, ms: 1, pending: NO_PENDING, - prOn: false, prKey: 0, prLast: 0, + prOn: false, prKey: 0, prLast: 0, reviewing: 0, boardOn: giteaUp, boardPosted: false, boardLast: 0, openIssues: if (giteaUp) s.openIssues.union(Set(t)) else s.openIssues, planWatch: OFF, prWatch: OFF }, @@ -182,15 +183,17 @@ module orchestrator { // _abandon: IDLE first, then close the issue and the PR, abort intents. pure def abandon(s: Sys): Sys = { ...s, st: "IDLE", task: false, gate: NO_GATE, pending: NO_PENDING, - prOn: false, boardOn: false, boardPosted: false, intents: Set(), + prOn: false, reviewing: 0, boardOn: false, boardPosted: false, intents: Set(), openIssues: s.openIssues.exclude(Set(s.tid)), giteaPr: if (s.prOn) s.giteaPr.put(s.prKey, "closed") else s.giteaPr, planWatch: OFF, prWatch: OFF } - // _handle_pr_open: 'project complete' + // _handle_pr_open: 'project complete' closes the plan issue; the PR is + // the operator's to merge. pure def complete(s: Sys): Sys = { ...s, st: "IDLE", task: false, gate: NO_GATE, pending: NO_PENDING, - prOn: false, boardOn: false, boardPosted: false, planWatch: OFF, prWatch: OFF } + prOn: false, reviewing: 0, boardOn: false, boardPosted: false, + openIssues: s.openIssues.exclude(Set(s.tid)), planWatch: OFF, prWatch: OFF } // _resume_after_restart pure def resume(s: Sys, p: Pending, j: Job): Sys = @@ -245,11 +248,12 @@ module orchestrator { if (not(SIGNOFF.contains(s.st))) s // not taken else applyGate({ ...s, boardLast: max(s.boardLast, j.cid) }, j) - // _handle_review_comment_now, first half: the executor starts on it. + // _handle_review_comment_now, first half: the executor starts on it, + // with the comment marked (durably) until it returns. pure def handleReviewComment(s: Sys, j: Job): Sys = if (s.st != "PR_OPEN") s // not taken else { ...s, prLast: max(s.prLast, j.cid), taken: s.taken.union(Set(j.cid)), - work: "review", workCid: j.cid } + reviewing: j.cid, work: "review", workCid: j.cid } pure def handleJob(s: Sys, j: Job, giteaUp: bool): Sys = if (j.src == "matrix") handleMatrix(s, j, giteaUp) @@ -271,7 +275,10 @@ module orchestrator { { ...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 } + // A comment whose answer never finished is delivered again. + { ...s0, prWatch: if (s.prOn) + { on: true, floor: if (s.reviewing > 0) s.reviewing - 1 else s.prLast } + else OFF } else if (s.st == "IDLE" or (SIGNOFF.contains(s.st) and s.gate.on)) if (s.gate.on and s.boardPosted) { ...s0, planWatch: { on: true, floor: s.boardLast } } else s0 else @@ -284,7 +291,7 @@ module orchestrator { action init = s' = { st: "IDLE", tid: 0, task: false, gate: NO_GATE, revs: Map("arch" -> 0, "ms1" -> 0, "ms2" -> 0), lastRev: 0, ms: 1, pending: NO_PENDING, - prOn: false, prKey: 0, prLast: 0, + prOn: false, prKey: 0, prLast: 0, reviewing: 0, boardOn: false, boardPosted: false, boardLast: 0, intents: Set(), planThread: List(), prThread: List(), nextId: 1, @@ -418,7 +425,7 @@ module orchestrator { action finishReview(outcome: str): bool = all { s.work == "review", outcome != "stopped" or s.cancel, s' = - val t = endWork({ ...s, answered: s.answered.union(Set(s.workCid)) }) + val t = endWork({ ...s, answered: s.answered.union(Set(s.workCid)), reviewing: 0 }) if (outcome == "stopped" or outcome == "escalate") awaitInput(t, { kind: "review_stopped", g: "", interrupted: "" }) else t, @@ -464,9 +471,11 @@ module orchestrator { // 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. + // A review comment the agent took up is answered, being worked on, or + // marked to be delivered again: never silently dropped. val review_comments_answered = - s.taken.forall(id => s.answered.contains(id) or (s.work == "review" and s.workCid == id)) + s.taken.forall(id => s.answered.contains(id) or (s.work == "review" and s.workCid == id) + or s.reviewing == id) // With no task, no plan issue is left open. val no_open_issue_without_a_task = not(s.task) implies s.openIssues.size() == 0 diff --git a/specs/orchestrator_scenarios.qnt b/specs/orchestrator_scenarios.qnt index 8348cd1..88307a1 100644 --- a/specs/orchestrator_scenarios.qnt +++ b/specs/orchestrator_scenarios.qnt @@ -1,8 +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), 5 (boot -// and publishing adopt an open PR). +// All fixed: 1-4 (replies bind to the revision up when they arrived), 5 (boot +// and publishing adopt an open PR), 6 (redelivered), 7 (issue closed). // // Run: quint test specs/orchestrator_scenarios.qnt (runs every *Test) module orchestrator_scenarios { @@ -107,18 +107,19 @@ module orchestrator_scenarios { // 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 - // restart in PR_OPEN resumes quietly, so the comment is never answered - // and nobody is told. + // restart in PR_OPEN resumed quietly, so it was never answered. Fixed: + // the comment is marked while answered, and delivered again. run crashDuringAReviewLosesTheCommentTest = toPrOpen .then(commentPr).then(pollPr).then(runJob(true)) // the executor starts on it .then(crash(true)) - .expect(s.st == "PR_OPEN" and not(review_comments_answered)) + .expect(s.st == "PR_OPEN" and review_comments_answered) + .then(pollPr).then(runJob(true)).then(finishReview("completed")) + .expect(s.answered.size() == 1 and s.reviewing == 0) - // 7. A completed task leaves its plan issue open: 'project complete' - // closes nothing, unlike 'abandon'. + // 7. 'project complete' left the plan issue open. Fixed: it closes it. run projectCompleteLeavesThePlanIssueOpenTest = toPrOpen .then(send("complete")).then(runJob(true)) - .expect(s.st == "IDLE" and not(no_open_issue_without_a_task)) + .expect(s.st == "IDLE" and no_open_issue_without_a_task) }