Merge pull request 'Redeliver an interrupted review comment, and close the issue on completion' (#76) from redeliver-review-and-close-issue into main

Reviewed-on: dongar/nisse#76
Reviewed-by: Don Garrett <dgarrett@acm.org>

nisse-dev/src/nisse_dev/control.py

@@ -351,9 +351,15 @@ class Orchestrator:
             # 8): resume, and watch the PR again from its cursor.
             pr = self._data.get("pr") or {}
             self._watch_pr(pr)
+            if pr.get("reviewing"):
                 self._transport.send_message(
-                f"restarted with PR #{pr.get('number')} open; still watching it "
+                    f"restarted while answering review comment {pr['reviewing']} "
-                "for review comments"
+                    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}")
         )
+        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"):
+        if problem := self._close_plan_issue(board_state, note):
-            try:
+            problems.append(problem)
-                self._board.close(int(board_state["issue"]), note)
-            except GiteaError as exc:
-                problems.append(f"plan issue #{board_state['issue']}: {exc}")
         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",
+                "review", int(pr["number"]), self.handle_review_comment, after=after
-                int(pr["number"]),
-                self.handle_review_comment,
-                after=pr.get("last_comment"),
             )
 
-    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 <project>' to start another"
+                "project closed"
+                + (f"; could not close {problem}" if problem else "")
+                + ". Say 'use <project>' to start another"
             )
         else:
             pr = self._data.get("pr") or {}

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
+real orchestrator. Each names the scenario in specs/orchestrator_scenarios.qnt
-scenario in specs/orchestrator_scenarios.qnt it reproduces.
+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
+    # The comment was taken but never answered: after the restart the
-    # be handled again, or the operator told it was dropped.
+    # operator is told, and it is delivered and answered again.
-    delivered = agent.pollers[20].poll_once(20)
+    assert "answering it again" in agent.transport.outbound[-1].text
-    told = any(
+    [delivered] = agent.pollers[20].poll_once(20)
-        "please add a test" in m.text or "review" in m.text
+    assert delivered.body == "please add a test"
-        for m in agent.transport.outbound[1:]
+    assert "pushed" in agent.transport.outbound[-1].text
-    )
+    assert "reviewing" not in agent.store.get_state().data["pr"]
-    assert delivered or told
 
 
-@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

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

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
+// All fixed: 1-4 (replies bind to the revision up when they arrived), 5 (boot
-// and publishing adopt an open PR).
+// 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
+  //    restart in PR_OPEN resumed quietly, so it was never answered. Fixed:
-  //    and nobody is told.
+  //    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'
+  // 7. 'project complete' left the plan issue open. Fixed: it closes it.
-  //    closes nothing, unlike 'abandon'.
   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)
 }