Add PR comment intake and the review loop (B2) — step 6b

- poller.py: CommentPoller polls PR comments rather than taking
  webhooks, keeping the VM outbound-only (B3). On an outage it backs
  off exponentially, caps, and retries indefinitely while logging:
  polling is pure intake, so an outage delays a reaction but never
  fails a task or forces a transition
- prime() marks the existing thread as seen without delivering it, so
  starting a watch does not re-address feedback already handled
- Review loop: a comment in PR_OPEN enters the canonical transcript,
  the executor responds, and the fix is pushed to the same branch with
  a reply on the PR. No intent is recorded for a follow-up push — the
  PR already exists, so a repeated push is harmless, unlike opening a
  second PR
- Comment authors are checked against a separate Gitea operator
  allowlist: a PR comment is an instruction channel into an agent that
  writes code, so it is operator-scoped, and a Gitea login is not a
  Matrix MXID (A4 per-channel allowlists)
- Orchestrator now holds a lock: transport and the poller deliver on
  separate threads, and state transitions must not interleave
- Backoff timings are injectable, so the outage path is tested in
  milliseconds instead of being untestable

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.env.example

@@ -31,3 +31,6 @@ NISSE_GITEA_URL=https://gitea.example
 NISSE_GITEA_TOKEN=your-scoped-service-account-token
 # Sandbox repo for live tests; never the real project repo.
 NISSE_GITEA_TEST_REPO=owner/sandbox-repo
+# Gitea logins allowed to drive the agent via PR comments (A4 per-channel
+# allowlist; a Matrix MXID and a Gitea login are different names).
+NISSE_GITEA_OPERATORS=your-gitea-login

nisse-dev/src/nisse_dev/control.py

@@ -8,6 +8,7 @@ arrives over transport. Every position change is checkpointed through the state
 store before the loop continues, so a restart lands mid-gate, not mid-air.
 """
 
+import threading
 import uuid
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
@@ -86,6 +87,12 @@ class MilestonePublisher(Protocol):
         self, *, task_id: str, milestone: int, project: str, plan: str
     ) -> Any: ...
 
+    def push_followup(
+        self, *, task_id: str, milestone: int, message: str
+    ) -> str | None: ...
+
+    def comment(self, index: int, body: str) -> None: ...
+
 
 class PlanningAgent(Protocol):
     def plan(
@@ -139,6 +146,7 @@ class Orchestrator:
         roles: RoleMap | None = None,
         executor: "MilestoneExecutor | None" = None,
         publisher: "MilestonePublisher | None" = None,
+        gitea_operators: Iterable[str] = (),
     ) -> None:
         self._store = store
         self._transport = transport
@@ -147,6 +155,13 @@ class Orchestrator:
         self._executor = executor
         self._publisher = publisher
         self._operators = set(operators)
+        # Per-channel allowlists (A4): a Matrix MXID and a Gitea login
+        # are different names for the same person, and neither implies
+        # the other.
+        self._gitea_operators = set(gitea_operators)
+        # Transport and the comment poller deliver on separate threads;
+        # state transitions must not interleave.
+        self._lock = threading.RLock()
         self._data = _fresh_data()
         self._project: str | None = None
         transport.on_message(self._handle)
@@ -197,6 +212,10 @@ class Orchestrator:
             )
 
     def _handle(self, msg: InboundMessage) -> None:
+        with self._lock:
+            self._handle_locked(msg)
+
+    def _handle_locked(self, msg: InboundMessage) -> None:
         if msg.sender not in self._operators:
             return
         # Every operator message enters the canonical transcript, so planning
@@ -376,6 +395,72 @@ class Orchestrator:
             pending={"kind": "milestone_stopped", "milestone": milestone},
         )
 
+    def handle_review_comment(self, comment: Any) -> None:
+        """React to a PR review comment (B2). Called from the poller thread.
+
+        Comment authors are checked against the Gitea operator allowlist:
+        a PR comment is an instruction channel into an agent that writes
+        code, so it is operator-scoped like any other (A4, and the same
+        reasoning C4 applies to tickets).
+        """
+        with self._lock:
+            self._handle_review_comment_locked(comment)
+
+    def _handle_review_comment_locked(self, comment: Any) -> None:
+        if comment.author not in self._gitea_operators:
+            self._store.oplog.emit(
+                "gitea.comment_ignored",
+                cause=f"author {comment.author!r} not an operator",
+                comment_id=comment.id,
+            )
+            return
+        state = self._store.get_state()
+        if state is None or state.state != PR_OPEN:
+            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
+        if self._executor is None or self._publisher is None:
+            return
+
+        pr = self._data.get("pr") or {}
+        milestone = int(pr.get("milestone", self._data.get("milestone", 1)))
+        # 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}")
+        )
+        outcome = self._executor.run(milestone)
+        if outcome.status != "completed":
+            self._await_input(
+                cause=f"review response {outcome.status}",
+                prompt=(
+                    f"could not address the review comment ({outcome.status}): "
+                    f"{outcome.note}"
+                ),
+                pending={"kind": "milestone_stopped", "milestone": milestone},
+            )
+            return
+        try:
+            sha = self._publisher.push_followup(
+                task_id=str(self._data.get("task_id")),
+                milestone=milestone,
+                message=f"Address review comment {comment.id}",
+            )
+        except (GiteaError, GitError) as exc:
+            self._transport.send_message(f"could not push the fix: {exc}")
+            return
+        note = (
+            f"pushed {sha[:8]} addressing the review comment"
+            if sha
+            else "no code change was needed for that comment"
+        )
+        self._transport.send_message(note)
+        if pr.get("number"):
+            self._publisher.comment(int(pr["number"]), note)
+
     def _publish_milestone(self, milestone: int) -> None:
         if self._publisher is None:
             return

nisse-dev/src/nisse_dev/poller.py

@@ -0,0 +1,104 @@
+"""Gitea comment intake (spec B2).
+
+Polling, not webhooks: the agent VM stays outbound-only (B3), so nothing
+needs to reach in. Intake is structured JSON straight from the API and never
+routed through a chat bridge.
+
+Outage posture, per B2: back off exponentially, cap at a few minutes, retry
+indefinitely, and log. Polling is pure intake — an outage delays a reaction
+but must never fail a task or force a state transition.
+"""
+
+import threading
+from collections.abc import Callable, Iterable
+
+from nisse_dev.gitea import Comment, GiteaClient, GiteaError
+from nisse_dev.oplog import OpLog
+
+DEFAULT_INTERVAL_S = 20.0
+INITIAL_BACKOFF_S = 5.0
+MAX_BACKOFF_S = 300.0
+
+
+class CommentPoller:
+    def __init__(
+        self,
+        client: GiteaClient,
+        repo: str,
+        on_comment: Callable[[Comment], None],
+        *,
+        oplog: OpLog | None = None,
+        ignore_authors: Iterable[str] = (),
+        interval_s: float = DEFAULT_INTERVAL_S,
+        initial_backoff_s: float = INITIAL_BACKOFF_S,
+        max_backoff_s: float = MAX_BACKOFF_S,
+    ) -> None:
+        self._client = client
+        self._repo = repo
+        self._on_comment = on_comment
+        self._oplog = oplog
+        self._ignore = set(ignore_authors)
+        self._interval_s = interval_s
+        self._initial_backoff_s = initial_backoff_s
+        self._max_backoff_s = max_backoff_s
+        self._seen: set[int] = set()
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+
+    def prime(self, index: int) -> None:
+        """Mark existing comments as seen without delivering them.
+
+        A poller that starts by replaying the whole thread would re-act to
+        feedback already addressed — including its own earlier replies.
+        """
+        try:
+            for comment in self._client.list_comments(self._repo, index):
+                self._seen.add(comment.id)
+        except GiteaError:
+            # Nothing seen means the first successful poll delivers the
+            # backlog once. Better than refusing to start.
+            pass
+
+    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:
+                continue
+            self._seen.add(comment.id)
+            if comment.author in self._ignore:
+                continue  # the agent's own replies are not new input
+            delivered.append(comment)
+            self._on_comment(comment)
+        return delivered
+
+    def start(self, index: int) -> None:
+        self._stop.clear()
+        self._thread = threading.Thread(target=self._loop, args=(index,), daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join(timeout=self._interval_s + 5)
+            self._thread = None
+
+    def _loop(self, index: int) -> None:
+        backoff = self._initial_backoff_s
+        while not self._stop.is_set():
+            try:
+                self.poll_once(index)
+            except GiteaError as exc:
+                if self._oplog is not None:
+                    self._oplog.emit(
+                        "gitea.poll_failed",
+                        cause="gitea unreachable",
+                        error=str(exc)[:300],
+                        retry_in_s=backoff,
+                    )
+                self._stop.wait(backoff)
+                backoff = min(backoff * 2, self._max_backoff_s)
+                continue
+            backoff = self._initial_backoff_s
+            self._stop.wait(self._interval_s)

nisse-dev/src/nisse_dev/publish.py

@@ -10,7 +10,12 @@ an idempotent transition reconciled against external truth.
 from dataclasses import dataclass
 
 from nisse_dev.git import GitRepo, authenticated_url
-from nisse_dev.gitea import GiteaClient, PullRequest, pull_request_reconciler
+from nisse_dev.gitea import (
+    GiteaClient,
+    GiteaError,
+    PullRequest,
+    pull_request_reconciler,
+)
 from nisse_dev.reconcile import Reconciler
 from nisse_dev.state import StateStore
 
@@ -86,6 +91,32 @@ 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 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."""
+        try:
+            self._client.post_comment(self._repo, index, body)
+        except GiteaError:
+            pass
+
+    def push_followup(
+        self, *, task_id: str, milestone: int, message: str
+    ) -> str | None:
+        """Push a fix commit onto an open milestone's branch.
+
+        No intent is recorded: the PR already exists, so a crash mid-push
+        costs a repeated push of the same commit, which is harmless. That is
+        the difference from publish(), where the side effect is a new PR.
+        """
+        branch = self.branch_for(task_id, milestone)
+        if self._git.current_branch() != branch:
+            self._git.checkout_new_branch(branch)
+        sha = self._git.commit_all(message)
+        if sha is None:
+            return None
+        self._git.push(self._remote, branch, token=self._token)
+        return sha
+
 
 def _pr_body(milestone: int, plan: str, service_url: str | None) -> str:
     parts = [

nisse-dev/tests/test_poller.py

@@ -0,0 +1,132 @@
+import io
+import json
+import time
+
+import pytest
+
+from nisse_dev.gitea import Comment, GiteaError
+from nisse_dev.oplog import OpLog
+from nisse_dev.poller import MAX_BACKOFF_S, CommentPoller
+
+
+class FakeGitea:
+    def __init__(self, pages: list[list[Comment]] | None = None) -> None:
+        self.pages = pages or []
+        self.calls = 0
+        self.fail_times = 0
+
+    def list_comments(self, repo: str, index: int, since: str | None = None):
+        self.calls += 1
+        if self.fail_times > 0:
+            self.fail_times -= 1
+            raise GiteaError("gitea unreachable")
+        if not self.pages:
+            return []
+        return self.pages.pop(0)
+
+
+def comment(cid: int, author: str = "don", body: str = "please fix") -> Comment:
+    return Comment(id=cid, author=author, body=body, created_at="t")
+
+
+def build(client, **kw) -> tuple[CommentPoller, list[Comment]]:
+    seen: list[Comment] = []
+    poller = CommentPoller(client, "o/r", seen.append, **kw)
+    return poller, seen
+
+
+def test_new_comments_are_delivered_once() -> None:
+    client = FakeGitea([[comment(1)], [comment(1), comment(2)]])
+    poller, seen = build(client)
+    poller.poll_once(1)
+    poller.poll_once(1)
+    assert [c.id for c in seen] == [1, 2]  # id 1 not redelivered
+
+
+def test_prime_suppresses_the_existing_backlog() -> None:
+    # Starting a watch must not replay feedback already addressed.
+    client = FakeGitea([[comment(1), comment(2)], [comment(1), comment(2), comment(3)]])
+    poller, seen = build(client)
+    poller.prime(1)
+    poller.poll_once(1)
+    assert [c.id for c in seen] == [3]
+
+
+def test_own_comments_are_ignored() -> None:
+    client = FakeGitea([[comment(1, author="nisse-dev-testing"), comment(2, "don")]])
+    poller, seen = build(client, ignore_authors={"nisse-dev-testing"})
+    poller.poll_once(1)
+    assert [c.id for c in seen] == [2]
+
+
+def test_ignored_comments_are_not_redelivered_later() -> None:
+    client = FakeGitea(
+        [
+            [comment(1, author="bot")],
+            [comment(1, author="bot"), comment(2, "don")],
+        ]
+    )
+    poller, seen = build(client, ignore_authors={"bot"})
+    poller.poll_once(1)
+    poller.poll_once(1)
+    assert [c.id for c in seen] == [2]
+
+
+def test_outage_does_not_raise_out_of_the_loop() -> None:
+    stream = io.StringIO()
+    client = FakeGitea([[comment(1)]])
+    client.fail_times = 2
+    poller, seen = build(
+        client,
+        oplog=OpLog("nisse-dev", stream),
+        interval_s=0.01,
+        initial_backoff_s=0.01,
+    )
+    poller.start(1)
+    deadline = time.monotonic() + 5
+    while not seen and time.monotonic() < deadline:
+        time.sleep(0.05)
+    poller.stop()
+
+    # It recovered and delivered rather than dying on the outage (B2: an
+    # outage delays a reaction, never fails a task).
+    assert [c.id for c in seen] == [1]
+    assert "gitea.poll_failed" in stream.getvalue()
+
+
+def test_prime_survives_an_outage() -> None:
+    client = FakeGitea()
+    client.fail_times = 1
+    poller, _ = build(client)
+    poller.prime(1)  # must not raise
+
+
+def test_backoff_grows_then_caps() -> None:
+    client = FakeGitea()
+    client.fail_times = 50
+    stream = io.StringIO()
+    poller, _ = build(
+        client,
+        oplog=OpLog("nisse-dev", stream),
+        interval_s=0.001,
+        initial_backoff_s=0.001,
+        max_backoff_s=0.008,
+    )
+    poller.start(1)
+    time.sleep(0.3)
+    poller.stop()
+
+    delays = [json.loads(line)["retry_in_s"] for line in stream.getvalue().splitlines()]
+    assert delays[:4] == [0.001, 0.002, 0.004, 0.008]
+    assert max(delays) == 0.008  # capped, and retried indefinitely (B2)
+    assert MAX_BACKOFF_S <= 300.0  # the shipped cap is "a few minutes"
+
+
+@pytest.mark.parametrize("author", ["don", "someone-else"])
+def test_poller_does_not_filter_by_author_itself(author: str) -> None:
+    # Authorisation belongs to the orchestrator's allowlist, not here: the
+    # poller is pure intake.
+    client = FakeGitea([[comment(1, author=author)]])
+    poller, seen = build(client)
+    poller.poll_once(1)
+    assert len(seen) == 1

nisse-dev/tests/test_review_loop.py

@@ -0,0 +1,184 @@
+"""Reacting to PR review comments while in PR_OPEN (spec B2)."""
+
+import io
+from pathlib import Path
+from typing import Any
+
+from nisse_dev.canonical import CanonicalEvent, StoredEvent, ToolCall
+from nisse_dev.control import AWAIT_INPUT, PR_OPEN, Orchestrator
+from nisse_dev.executor import ExecutorOutcome
+from nisse_dev.gitea import Comment, GiteaError, PullRequest
+from nisse_dev.llm import RoleMap
+from nisse_dev.oplog import OpLog
+from nisse_dev.planner import LLMPlanningAgent
+from nisse_dev.publish import PublishResult
+from nisse_dev.state import StateStore
+from nisse_dev.transport import InMemoryTransport
+
+ROLES = RoleMap(executor="x", planner="y", delegate="y", consultant="y")
+
+
+class ScriptedClient:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def complete(self, events, tools, model, system=None) -> list[CanonicalEvent]:
+        self.calls += 1
+        return [
+            ToolCall(
+                call_id=f"c{self.calls}",
+                tool="submit_plan",
+                arguments={"type": "x", "content": f"plan {self.calls}"},
+            )
+        ]
+
+
+class RecordingExecutor:
+    def __init__(self, outcomes: list[str] | None = None) -> None:
+        self.runs: list[int] = []
+        self.seen_events: list[list[StoredEvent]] = []
+        # One entry per run, so a test can let the milestone succeed and
+        # only the review response fail.
+        self._outcomes = outcomes or []
+        self.store: StateStore | None = None
+
+    def run(self, milestone: int) -> ExecutorOutcome:
+        self.runs.append(milestone)
+        if self.store is not None:
+            self.seen_events.append(self.store.read_events())
+        status = self._outcomes.pop(0) if self._outcomes else "completed"
+        return ExecutorOutcome(status, f"executor said {status}")
+
+
+class RecordingPublisher:
+    def __init__(self, *, sha: str | None = "abc12345", error: Exception | None = None):
+        self._sha = sha
+        self._error = error
+        self.followups: list[dict[str, Any]] = []
+        self.comments: list[tuple[int, str]] = []
+
+    def reconcilers(self) -> dict[str, Any]:
+        return {}
+
+    def publish(self, **kwargs: Any) -> PublishResult:
+        return PublishResult(
+            PullRequest(number=42, state="open", head="h", url="u"), "opened PR #42: u"
+        )
+
+    def push_followup(self, **kwargs: Any) -> str | None:
+        self.followups.append(kwargs)
+        if self._error is not None:
+            raise self._error
+        return self._sha
+
+    def comment(self, index: int, body: str) -> None:
+        self.comments.append((index, body))
+
+
+def build(
+    tmp_path: Path,
+    executor: Any,
+    publisher: Any,
+    gitea_operators: set[str] | None = None,
+) -> tuple[Orchestrator, StateStore, InMemoryTransport]:
+    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+    if hasattr(executor, "store"):
+        executor.store = store
+    transport = InMemoryTransport()
+    orch = Orchestrator(
+        store,
+        transport,
+        LLMPlanningAgent(ScriptedClient(), store, ROLES),
+        operators={"don"},
+        roles=ROLES,
+        executor=executor,
+        publisher=publisher,
+        gitea_operators=gitea_operators if gitea_operators is not None else {"don"},
+    )
+    orch.start()
+    transport.receive("don", "use demo")
+    transport.receive("don", "approve")
+    transport.receive("don", "approve")
+    transport.receive("don", "start milestone 1")
+    return orch, store, transport
+
+
+def review(
+    cid: int = 5, author: str = "don", body: str = "rename the module"
+) -> Comment:
+    return Comment(id=cid, author=author, body=body, created_at="t")
+
+
+def test_review_comment_triggers_a_fix_and_a_push(tmp_path: Path) -> None:
+    executor, publisher = RecordingExecutor(), RecordingPublisher()
+    orch, store, transport = build(tmp_path, executor, publisher)
+    assert store.get_state().state == PR_OPEN
+
+    orch.handle_review_comment(review())
+
+    assert executor.runs == [1, 1]  # milestone run, then the review response
+    assert publisher.followups[0]["milestone"] == 1
+    assert "abc12345" in transport.outbound[-1].text
+    assert publisher.comments[0][0] == 42  # replied on the PR
+
+
+def test_comment_enters_the_transcript_before_the_executor_runs(
+    tmp_path: Path,
+) -> None:
+    executor, publisher = RecordingExecutor(), RecordingPublisher()
+    orch, _, _ = build(tmp_path, executor, publisher)
+    orch.handle_review_comment(review(body="please add a test"))
+
+    # The executor must see the feedback as conversation, not as an
+    # out-of-band instruction it was never shown.
+    texts = [getattr(e.event, "text", "") for e in executor.seen_events[-1]]
+    assert any("please add a test" in t for t in texts)
+
+
+def test_non_operator_comments_are_ignored(tmp_path: Path) -> None:
+    executor, publisher = RecordingExecutor(), RecordingPublisher()
+    orch, store, _ = build(tmp_path, executor, publisher)
+    orch.handle_review_comment(review(author="drive-by"))
+
+    assert executor.runs == [1]  # only the original milestone run
+    assert publisher.followups == []
+    assert store.get_state().state == PR_OPEN
+
+
+def test_comments_outside_pr_open_are_ignored(tmp_path: Path) -> None:
+    executor, publisher = RecordingExecutor(), RecordingPublisher()
+    orch, _store, transport = build(tmp_path, executor, publisher)
+    transport.receive("don", "project complete")  # back to IDLE
+
+    orch.handle_review_comment(review())
+    assert executor.runs == [1]
+
+
+def test_failed_review_response_gates_to_the_operator(tmp_path: Path) -> None:
+    # Milestone succeeds; the response to the review comment does not.
+    executor = RecordingExecutor(outcomes=["completed", "escalate"])
+    publisher = RecordingPublisher()
+    orch, store, transport = build(tmp_path, executor, publisher)
+    orch.handle_review_comment(review())
+
+    assert store.get_state().state == AWAIT_INPUT
+    assert "could not address the review comment" in transport.outbound[-1].text
+    assert publisher.followups == []
+
+
+def test_no_code_change_needed_is_reported(tmp_path: Path) -> None:
+    executor, publisher = RecordingExecutor(), RecordingPublisher(sha=None)
+    orch, _, transport = build(tmp_path, executor, publisher)
+    orch.handle_review_comment(review(body="just curious, why sqlite?"))
+
+    assert "no code change was needed" in transport.outbound[-1].text
+
+
+def test_push_failure_is_reported_without_losing_pr_open(tmp_path: Path) -> None:
+    executor = RecordingExecutor()
+    publisher = RecordingPublisher(error=GiteaError("push rejected"))
+    orch, store, transport = build(tmp_path, executor, publisher)
+    orch.handle_review_comment(review())
+
+    assert "could not push the fix" in transport.outbound[-1].text
+    assert store.get_state().state == PR_OPEN  # still reviewable, can retry