diff --git a/.env.example b/.env.example index 13dc32f..51a4464 100644 --- a/.env.example +++ b/.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 diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index f01077f..c33b884 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/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 diff --git a/nisse-dev/src/nisse_dev/poller.py b/nisse-dev/src/nisse_dev/poller.py new file mode 100644 index 0000000..d486707 --- /dev/null +++ b/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) diff --git a/nisse-dev/src/nisse_dev/publish.py b/nisse-dev/src/nisse_dev/publish.py index 93b613f..19060af 100644 --- a/nisse-dev/src/nisse_dev/publish.py +++ b/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 = [ diff --git a/nisse-dev/tests/test_poller.py b/nisse-dev/tests/test_poller.py new file mode 100644 index 0000000..4e94c9d --- /dev/null +++ b/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 diff --git a/nisse-dev/tests/test_review_loop.py b/nisse-dev/tests/test_review_loop.py new file mode 100644 index 0000000..2cc67c5 --- /dev/null +++ b/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