diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index 18eeaf1..00c5b12 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/nisse-dev/src/nisse_dev/control.py @@ -277,12 +277,33 @@ class Orchestrator: "restarted while waiting on you: " + pending.get("prompt", "my last question still stands") ) + 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. + 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" + ) elif state.state == IDLE or ( state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF) and gate ): # A gate is a safe place to have been interrupted: nothing was in - # flight, the plan and revision are durable. Announce and resume. + # flight, the plan and revision are durable. Announce and resume, + # watching the plan issue again from its cursor. if gate: + board_state = self._data.get("board") + if board_state and board_state.get("comment_id"): + self._watch( + "plan", + board_state["issue"], + self.handle_plan_comment, + # No cursor means data from before cursors were + # kept: skip what is there, as a fresh watch does, + # so an old approval cannot approve this revision. + after=board_state.get("last_comment") or None, + ) self._transport.send_message( f"restarted while awaiting sign-off on {gate['name']} " f"rev-{gate['rev']}; the gate is still open" @@ -671,8 +692,12 @@ class Orchestrator: return board_state["comment_id"] = post.comment_id # Approvals can arrive on the issue as well as over transport (A4), - # so watch it for as long as the gate stands. - self._watch("plan", board_state["issue"], self.handle_plan_comment) + # so watch it for as long as the gate stands. A new revision skips + # what came before it: an early "approve" must not approve it. + board_state["last_comment"] = self._watch( + "plan", board_state["issue"], self.handle_plan_comment + ) + self._save(f"{gate_name} rev-{rev} posted to Gitea") def handle_plan_comment(self, comment: Any) -> None: """A comment on the project issue may satisfy an open plan gate (A4). @@ -689,14 +714,17 @@ class Orchestrator: ) def _handle_plan_comment_now(self, comment: Any) -> None: + board_state = self._data.get("board") if comment.author not in self._gitea_operators: + self._take_comment(board_state, comment) # never actionable return state = self._store.get_state() if state is None or state.state not in ( AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF, ): - return + return # not taken: a watch attached later still sees it + self._take_comment(board_state, comment) self._store.append_event( UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}") ) @@ -720,6 +748,7 @@ class Orchestrator: def _handle_review_comment_now(self, comment: Any) -> None: if comment.author not in self._gitea_operators: + self._take_comment(self._data.get("pr"), comment) # never actionable self._store.oplog.emit( "gitea.comment_ignored", cause=f"author {comment.author!r} not an operator", @@ -728,12 +757,15 @@ class Orchestrator: return state = self._store.get_state() if state is None or state.state != PR_OPEN: + # Not taken: back in PR_OPEN, the watch resumes from the cursor + # and delivers it again. self._store.oplog.emit( "gitea.comment_ignored", cause=f"not in PR_OPEN (state={state.state if state else None})", comment_id=comment.id, ) return + self._take_comment(self._data.get("pr"), comment) if self._executor is None or self._publisher is None: return @@ -750,11 +782,8 @@ class Orchestrator: self._halt_for_stop( what=f"the response to review comment {comment.id}", note=outcome.note, - next_step=( - f"The milestone {milestone} plan is still approved; say " - f"'start milestone {milestone}' to run it again." - ), - pending={"kind": "milestone_stopped", "milestone": milestone}, + next_step="Your reply returns to the open PR.", + pending={"kind": "review_stopped", "milestone": milestone}, ) return if outcome.status != "completed": @@ -762,9 +791,10 @@ class Orchestrator: cause=f"review response {outcome.status}", prompt=( f"could not address the review comment ({outcome.status}): " - f"{outcome.note}" + f"{outcome.note}. Reply to return to the open PR; comment " + "again there to retry." ), - pending={"kind": "milestone_stopped", "milestone": milestone}, + pending={"kind": "review_stopped", "milestone": milestone}, ) return try: @@ -824,19 +854,41 @@ class Orchestrator: pending={"kind": "milestone_stopped", "milestone": milestone}, ) return + self._unwatch("plan") # the gate is behind us; the PR is the thread now self._data["pr"] = { "number": result.pull_request.number, "url": result.pull_request.url, "milestone": milestone, + "last_comment": self._watch( + "review", result.pull_request.number, self.handle_review_comment + ), } - self._unwatch("plan") # the gate is behind us; the PR is the thread now - self._watch("review", result.pull_request.number, self.handle_review_comment) self._transition(PR_OPEN, cause=f"milestone {milestone} published") self._transport.send_message( f"{result.note} — review it, then say 'next milestone' or " "'project complete'" ) + def _return_to_pr_open(self, cause: str) -> None: + self._transition(PR_OPEN, cause=cause) + pr = self._data.get("pr") or {} + self._watch_pr(pr) + self._transport.send_message( + f"PR #{pr.get('number')} is open and watched for review comments; " + "say 'next milestone' or 'project complete'" + ) + + def _watch_pr(self, pr: dict[str, Any]) -> None: + """Watch the open PR from its recorded cursor, catching up on what + arrived while it was not watched.""" + if pr.get("number"): + self._watch( + "review", + int(pr["number"]), + self.handle_review_comment, + after=pr.get("last_comment"), + ) + def _handle_pr_open(self, text: str) -> None: if text == "next milestone": self._unwatch("review") @@ -861,17 +913,42 @@ class Orchestrator: "'project complete'" ) - def _watch(self, kind: str, index: int, handler: Any) -> None: + def _watch( + self, kind: str, index: int, handler: Any, *, after: int | None = None + ) -> int: """Start watching one Gitea thread, replacing any previous watch of - the same kind. Priming first, so attaching does not replay comments - already addressed.""" + the same kind, and return the comment cursor to record. + + A fresh watch (after=None) skips everything already there, so it + does not replay comments already addressed. Watching again from a + recorded cursor delivers what came in while nothing was watching.""" if self._poller_factory is None: - return + return after or 0 self._unwatch(kind) poller = self._poller_factory(handler) - poller.prime(index) + if after is None: + cursor = poller.prime(index) + else: + poller.resume_after(after) + cursor = after poller.start(index) self._pollers[kind] = poller + return cursor + + def _save(self, cause: str) -> None: + """Make changes to the task's data durable without a state change.""" + state = self._store.get_state() + self._transition(state.state if state else IDLE, cause=cause) + + def _take_comment(self, record: dict[str, Any] | None, comment: Any) -> None: + """Advance a thread's cursor past a comment being acted on, durably, + before acting: a comment is handled at most once, even across a + crash mid-way.""" + if record is not None: + record["last_comment"] = max( + int(record.get("last_comment") or 0), comment.id + ) + self._save(f"gitea comment {comment.id} taken") def _unwatch(self, kind: str) -> None: poller = self._pollers.pop(kind, None) @@ -910,6 +987,8 @@ class Orchestrator: back = PLANNING_ARCH if gate_name == "arch" else MILESTONE_PLANNING self._transition(back, cause="operator answered; resuming planning") self._run_planning_turn(gate_name, answer=text) + case "review_stopped": + self._return_to_pr_open("operator answered; back to the open PR") case "boot": self._resume_after_restart(pending, text) case "milestone_stopped": @@ -939,6 +1018,12 @@ class Orchestrator: milestone = int(self._data.get("milestone", 1)) gate = self._data.get("gate") or {} interrupted = pending.get("interrupted") + if interrupted == PR_OPEN: + # Recorded before a restart in PR_OPEN resumed quietly. Its PR is + # still open, so either reply goes back to it; re-planning the + # milestone would open a second PR. + self._return_to_pr_open(f"operator: {choice} after restart") + return if choice == "continue" and interrupted == EXECUTING and gate.get("approved"): self._transition( AWAIT_MILESTONE_SIGNOFF, cause="operator: continue after restart" diff --git a/nisse-dev/src/nisse_dev/poller.py b/nisse-dev/src/nisse_dev/poller.py index d486707..c039d50 100644 --- a/nisse-dev/src/nisse_dev/poller.py +++ b/nisse-dev/src/nisse_dev/poller.py @@ -42,29 +42,43 @@ class CommentPoller: self._initial_backoff_s = initial_backoff_s self._max_backoff_s = max_backoff_s self._seen: set[int] = set() + # Comments at or below this ID were handled before this poller + # existed (a restart, or a watch attached again); Gitea's comment IDs + # only grow, so one number says it all. + self._after = 0 self._stop = threading.Event() self._thread: threading.Thread | None = None - def prime(self, index: int) -> None: - """Mark existing comments as seen without delivering them. + def prime(self, index: int) -> int: + """Mark existing comments as seen without delivering them, and + return the highest ID marked (0 if none). A poller that starts by replaying the whole thread would re-act to feedback already addressed — including its own earlier replies. """ + highest = 0 try: for comment in self._client.list_comments(self._repo, index): self._seen.add(comment.id) + highest = max(highest, comment.id) except GiteaError: # Nothing seen means the first successful poll delivers the # backlog once. Better than refusing to start. pass + return max(highest, self._after) + + def resume_after(self, comment_id: int) -> None: + """Deliver only comments newer than comment_id: the ones posted + while nothing was watching are caught up, the handled ones are not + replayed. Needs no request, so an outage cannot cause a replay.""" + self._after = comment_id def poll_once(self, index: int) -> list[Comment]: """One pass; returns what was newly delivered. Separated from the loop so behaviour is testable without threads or sleeps.""" delivered: list[Comment] = [] for comment in self._client.list_comments(self._repo, index): - if comment.id in self._seen: + if comment.id in self._seen or comment.id <= self._after: continue self._seen.add(comment.id) if comment.author in self._ignore: diff --git a/nisse-dev/tests/test_pr_open_handling.py b/nisse-dev/tests/test_pr_open_handling.py new file mode 100644 index 0000000..0ecc91d --- /dev/null +++ b/nisse-dev/tests/test_pr_open_handling.py @@ -0,0 +1,250 @@ +"""Gitea threads across restarts and interruptions (#58, #59). + +Uses the real CommentPoller against a fake Gitea, with its thread replaced +by explicit polls, so the cursor logic under test is the real one. +""" + +import io +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from nisse_dev.control import ( + AWAIT_ARCH_SIGNOFF, + AWAIT_INPUT, + AWAIT_MILESTONE_SIGNOFF, + MILESTONE_PLANNING, + PR_OPEN, + Orchestrator, + StubAgent, +) +from nisse_dev.executor import ExecutorOutcome +from nisse_dev.gitea import Comment, PullRequest +from nisse_dev.oplog import OpLog +from nisse_dev.poller import CommentPoller +from nisse_dev.publish import PublishResult +from nisse_dev.state import StateStore +from nisse_dev.transport import InMemoryTransport + +AGENT = "nisse-bot" +PLAN_ISSUE = 10 +PR = 20 + + +class FakeGitea: + """Comment threads by issue number; IDs grow across all threads.""" + + def __init__(self) -> None: + self.threads: dict[int, list[Comment]] = {} + self._next_id = 100 + + def add(self, index: int, author: str, body: str) -> int: + self._next_id += 1 + self.threads.setdefault(index, []).append( + Comment(id=self._next_id, author=author, body=body, created_at="t") + ) + return self._next_id + + def list_comments(self, repo: str, index: int, since: str | None = None) -> list: + return list(self.threads.get(index, [])) + + +@dataclass +class Posted: + comment_id: int + + +class Board: + def __init__(self, gitea: FakeGitea) -> None: + self._gitea = gitea + + def open_project_issue(self, project: str, task_id: str) -> int: + return PLAN_ISSUE + + def post_plan(self, issue: int, gate: str, rev: int, plan: str) -> Posted: + return Posted(self._gitea.add(issue, AGENT, f"{gate} rev-{rev}: {plan}")) + + def echo(self, issue: int, text: str) -> None: + self._gitea.add(issue, AGENT, text) + + +class Executor: + def __init__(self) -> None: + self.runs = 0 + self.outcome = ExecutorOutcome("completed", "done") + + def run(self, milestone: int) -> ExecutorOutcome: + self.runs += 1 + return self.outcome + + +class Publisher: + def __init__(self) -> None: + self.followups = 0 + + def reconcilers(self) -> dict[str, Any]: + return {} + + def publish(self, **kwargs: Any) -> PublishResult: + return PublishResult( + PullRequest(number=PR, state="open", head="h", url="u"), f"opened PR #{PR}" + ) + + def push_followup(self, **kwargs: Any) -> str | None: + self.followups += 1 + return "abcdef1234" + + def comment(self, index: int, body: str) -> None: + pass + + +class Agent: + """One orchestrator process: a restart builds a new one on the same + store and the same Gitea.""" + + def __init__(self, tmp_path: Path, gitea: FakeGitea, executor: Executor) -> None: + self.store = StateStore( + tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()) + ) + self.transport = InMemoryTransport() + self.pollers: dict[int, CommentPoller] = {} + self.publisher = Publisher() + + def factory(handler: Any) -> CommentPoller: + poller = CommentPoller(gitea, "o/r", handler, ignore_authors=[AGENT]) + agent = self + + def start(index: int) -> None: # no thread: tests poll explicitly + agent.pollers[index] = poller + + poller.start = start # type: ignore[method-assign] + poller.stop = lambda: None # type: ignore[method-assign] + return poller + + self.orch = Orchestrator( + self.store, + self.transport, + StubAgent(), + operators={"don"}, + gitea_operators={"don"}, + board=Board(gitea), + executor=executor, + publisher=self.publisher, + poller_factory=factory, + ) + self.orch.start() + + @property + def state(self) -> str: + state = self.store.get_state() + assert state is not None + return state.state + + def poll(self, index: int) -> list[Comment]: + return self.pollers[index].poll_once(index) + + +def to_pr_open(agent: Agent) -> None: + for reply in ("use demo", "approve", "approve", "start milestone 1"): + agent.transport.receive("don", reply) + assert agent.state == PR_OPEN + + +def test_an_approval_posted_while_down_is_applied_after_a_restart( + tmp_path: Path, +) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea, Executor()) + agent.transport.receive("don", "use demo") + assert agent.state == AWAIT_ARCH_SIGNOFF + + gitea.add(PLAN_ISSUE, "don", "approve") # while the agent is down + agent = Agent(tmp_path, gitea, Executor()) + [delivered] = agent.poll(PLAN_ISSUE) + assert delivered.body == "approve" + assert agent.state == AWAIT_MILESTONE_SIGNOFF + + +def test_a_handled_approval_is_not_replayed_after_a_restart(tmp_path: Path) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea, Executor()) + agent.transport.receive("don", "use demo") + gitea.add(PLAN_ISSUE, "don", "approve") + agent.poll(PLAN_ISSUE) # arch approved; the milestone plan is posted + assert agent.state == AWAIT_MILESTONE_SIGNOFF + + agent = Agent(tmp_path, gitea, Executor()) + assert agent.poll(PLAN_ISSUE) == [] + assert agent.state == AWAIT_MILESTONE_SIGNOFF + + +def test_a_new_revision_ignores_an_approval_posted_before_it(tmp_path: Path) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea, Executor()) + agent.transport.receive("don", "use demo") + agent.transport.receive("don", "needs tests") # rejects: rev-2 is posted + assert agent.state == AWAIT_ARCH_SIGNOFF + stale = gitea.add(PLAN_ISSUE, "don", "approve") + agent.transport.receive("don", "and docs") # rev-3 posted after it + assert all(c.id != stale for c in agent.poll(PLAN_ISSUE)) + assert agent.state == AWAIT_ARCH_SIGNOFF + + +def test_data_from_before_cursors_skips_old_comments_on_restart( + tmp_path: Path, +) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea, Executor()) + agent.transport.receive("don", "use demo") + gitea.add(PLAN_ISSUE, "don", "approve") # an old approval, say of rev-1 + state = agent.store.get_state() + state.data["board"].pop("last_comment") # as saved before this change + agent.store.transition( + state.state, cause="test", project=state.project, data=state.data + ) + agent = Agent(tmp_path, gitea, Executor()) + assert agent.poll(PLAN_ISSUE) == [] + assert agent.state == AWAIT_ARCH_SIGNOFF + + +def test_a_restart_in_pr_open_resumes_and_catches_up_on_review( + tmp_path: Path, +) -> None: + gitea = FakeGitea() + agent = Agent(tmp_path, gitea, Executor()) + to_pr_open(agent) + + gitea.add(PR, "don", "please add a test") # while the agent is down + agent = Agent(tmp_path, gitea, Executor()) + assert agent.state == PR_OPEN # no restart question + assert "restarted with PR #20 open" in agent.transport.outbound[-1].text + [delivered] = agent.poll(PR) + assert delivered.body == "please add a test" + assert agent.publisher.followups == 1 + assert agent.state == PR_OPEN + + +def test_a_failed_review_response_returns_to_the_open_pr(tmp_path: Path) -> None: + gitea = FakeGitea() + executor = Executor() + agent = Agent(tmp_path, gitea, executor) + to_pr_open(agent) + + executor.outcome = ExecutorOutcome("escalate", "stuck") + gitea.add(PR, "don", "please add a test") + agent.poll(PR) + assert agent.state == AWAIT_INPUT + + # A comment while it waits is not lost: back in PR_OPEN it is delivered. + executor.outcome = ExecutorOutcome("completed", "done") + gitea.add(PR, "don", "try again") + agent.poll(PR) + assert agent.state == AWAIT_INPUT + agent.transport.receive("don", "ok") + assert agent.state == PR_OPEN + [delivered] = agent.poll(PR) + assert delivered.body == "try again" + assert agent.publisher.followups == 1 + + agent.transport.receive("don", "next milestone") + assert agent.state in (MILESTONE_PLANNING, AWAIT_MILESTONE_SIGNOFF)