Keep Gitea threads watched across restarts and interruptions

- Record a comment cursor per thread (board.last_comment, the field that
  already existed unused; pr.last_comment), advanced durably when a
  comment is taken up. CommentPoller.resume_after() skips comments at or
  below it, so re-attaching needs no request and cannot replay.
- Boot re-attaches the plan watch at an open gate and the review watch in
  PR_OPEN from their cursors, catching up on comments posted while the
  agent was down (#59). Data saved without a cursor skips what is there,
  so an old approval cannot approve the current revision.
- A restart in PR_OPEN resumes with a notice instead of the restart
  question (#58). A stored restart question for PR_OPEN returns to it on
  either reply instead of re-planning the milestone.
- A stopped or failed review response gets its own pending kind; the
  reply returns to PR_OPEN and resumes the watch from the cursor, so
  comments that arrived meanwhile are delivered. It used to return to
  milestone sign-off, where starting the milestone would open a second PR.
- Comments ignored because of the state leave the cursor alone, so they
  are delivered once the agent is back where they apply.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE

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"

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:

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)