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/approvals.py b/nisse-dev/src/nisse_dev/approvals.py new file mode 100644 index 0000000..5b5edf1 --- /dev/null +++ b/nisse-dev/src/nisse_dev/approvals.py @@ -0,0 +1,80 @@ +"""Plan externalisation and Gitea-backed approvals (spec A4, Part H). + +Plans are posted as comments on a per-project Gitea issue. The approval +comment — its author, timestamp, and revision tag — is the durable off-box +record, authenticated by Gitea's own auth. The state store keeps only a +pointer to it, so boot reconciliation can re-derive gate state from Gitea +even after the state volume is lost: external truth wins. + +Approvals arriving here bind to a revision exactly as they do over transport +(A4). The first valid approval on any channel wins; the orchestrator echoes +the outcome to the other channel so the records converge. +""" + +from dataclasses import dataclass + +from nisse_dev.gitea import Comment, GiteaClient, GiteaError + +APPROVAL_HINT = ( + "Reply `approve {tag}` to approve this revision, or comment with feedback " + "to reject it." +) + + +@dataclass(frozen=True) +class PlanPost: + issue: int + comment_id: int + + +class PlanBoard: + """The Gitea side of plan gating: one issue per project, one comment per + plan revision.""" + + def __init__(self, client: GiteaClient, repo: str) -> None: + self._client = client + self._repo = repo + + @property + def repo(self) -> str: + return self._repo + + def open_project_issue(self, project: str, task_id: str) -> int: + return self._client.create_issue( + self._repo, + title=f"nisse-dev: {project}", + body=( + f"Plan gate for project `{project}` (task `{task_id}`).\n\n" + "Each plan revision is posted here as a comment. Approving a " + "revision by comment is equivalent to approving it in Matrix; " + "the first valid approval on either channel wins." + ), + ) + + def post_plan(self, issue: int, gate: str, rev: int, plan: str) -> PlanPost: + tag = f"rev-{rev}" + body = "\n".join( + [ + f"## Plan for `{gate}` ({tag})", + "", + plan, + "", + "---", + APPROVAL_HINT.format(tag=tag), + ] + ) + comment = self._client.post_comment(self._repo, issue, body) + return PlanPost(issue=issue, comment_id=comment.id) + + def echo(self, issue: int, text: str) -> None: + """Mirror an outcome decided on another channel, so the Gitea record + does not silently disagree with Matrix.""" + try: + self._client.post_comment(self._repo, issue, text) + except GiteaError: + pass + + def comments_since(self, issue: int, after_id: int) -> list[Comment]: + return [ + c for c in self._client.list_comments(self._repo, issue) if c.id > after_id + ] diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index f01077f..ff9ffd0 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,23 @@ 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 PlanBoard(Protocol): + """The Gitea side of plan gating; the concrete one is in + nisse_dev.approvals.""" + + def open_project_issue(self, project: str, task_id: str) -> int: ... + + def post_plan(self, issue: int, gate: str, rev: int, plan: str) -> Any: ... + + def echo(self, issue: int, text: str) -> None: ... + class PlanningAgent(Protocol): def plan( @@ -126,6 +144,10 @@ def _fresh_data() -> dict[str, Any]: "task_id": None, "roles": None, # A11 mapping, pinned at task start "pr": None, # the open PR for the current milestone, once published + # Pointer to the Gitea plan record (Part H): the state store holds + # only this, so gate state can be re-derived from Gitea after a + # state-volume loss. + "board": None, } @@ -139,6 +161,8 @@ class Orchestrator: roles: RoleMap | None = None, executor: "MilestoneExecutor | None" = None, publisher: "MilestonePublisher | None" = None, + gitea_operators: Iterable[str] = (), + board: "PlanBoard | None" = None, ) -> None: self._store = store self._transport = transport @@ -147,6 +171,14 @@ 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) + self._board = board + # 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 +229,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 @@ -241,6 +277,19 @@ class Orchestrator: self._data = {**_fresh_data(), "task_id": task_id, "roles": pinned} self._project = project self._store.oplog.task_id = task_id + if self._board is not None: + try: + self._data["board"] = { + "issue": self._board.open_project_issue(project, task_id), + "last_comment": 0, + } + except GiteaError as exc: + # Gitea being down must not block planning; Matrix remains a + # complete approval channel on its own. + self._transport.send_message( + f"note: could not open the Gitea plan issue ({exc}); " + "approvals will be Matrix-only for this task" + ) self._transition(PLANNING_ARCH, cause=f"operator command: use {project}") self._run_planning_turn("arch") @@ -289,6 +338,7 @@ class Orchestrator: f"{action.content}\n\nreply 'approve' to approve rev-{rev}; " "any other reply is rejection feedback" ) + self._post_plan_to_board(gate_name, rev, action.content) self._store.oplog.emit( "gate.open", cause=f"plan submitted (rev-{rev})", @@ -298,6 +348,12 @@ class Orchestrator: ) def _handle_gate_reply(self, msg: InboundMessage, text: str) -> None: + self._apply_gate_reply(sender=msg.sender, text=text, channel="transport") + + def _apply_gate_reply(self, *, sender: str, text: str, channel: str) -> None: + """Shared by every approval source (A4). The allowlist is per channel: + a Matrix MXID and a Gitea login are different names, and holding one + says nothing about holding the other.""" gate_data = self._data["gate"] name, rev = gate_data["name"], gate_data["rev"] milestone = self._data["milestone"] @@ -313,9 +369,8 @@ class Orchestrator: ) return revision = PlanRevision(gate=name, rev=rev, content=gate_data["plan"]) - result = Gate(revision, self._operators).evaluate( - msg.sender, text, channel="transport" - ) + allowlist = self._operators if channel == "transport" else self._gitea_operators + result = Gate(revision, allowlist).evaluate(sender, text, channel=channel) if isinstance(result, Ignored): if result.notice: self._transport.send_message(result.notice) @@ -328,12 +383,19 @@ class Orchestrator: rev=f"rev-{rev}", channel=result.channel, ) + self._echo_outcome( + channel, + f"{name} rev-{rev} rejected by {result.approver}: {result.feedback}", + ) self._data["gate"] = None back = PLANNING_ARCH if name == "arch" else MILESTONE_PLANNING self._transition(back, cause=f"plan rejected ({name} rev-{rev})") self._run_planning_turn(name, feedback=result.feedback) return assert isinstance(result, Approved) + self._echo_outcome( + channel, f"{name} rev-{rev} approved by {result.approver} via {channel}" + ) self._store.oplog.emit( "gate.approve", cause=f"approved by {result.approver}", @@ -376,6 +438,118 @@ class Orchestrator: pending={"kind": "milestone_stopped", "milestone": milestone}, ) + def _echo_outcome(self, deciding_channel: str, text: str) -> None: + """A gate satisfied on one channel is echoed to the other, so neither + record silently disagrees about what was approved (A4).""" + board_state = self._data.get("board") + if deciding_channel != "gitea" and self._board is not None and board_state: + self._board.echo(board_state["issue"], text) + elif deciding_channel == "gitea": + self._transport.send_message(text) + + def _post_plan_to_board(self, gate_name: str, rev: int, plan: str) -> None: + board_state = self._data.get("board") + if self._board is None or not board_state: + return + try: + post = self._board.post_plan(board_state["issue"], gate_name, rev, plan) + except GiteaError: + # Matrix already carries the plan; a Gitea outage degrades the + # durability of the record, not the ability to approve. + return + board_state["comment_id"] = post.comment_id + + def handle_plan_comment(self, comment: Any) -> None: + """A comment on the project issue may satisfy an open plan gate (A4). + + Approvals bind to a revision here exactly as over transport, and the + first valid approval on either channel wins. + """ + with self._lock: + self._handle_plan_comment_locked(comment) + + def _handle_plan_comment_locked(self, comment: Any) -> None: + if comment.author not in self._gitea_operators: + return + state = self._store.get_state() + if state is None or state.state not in ( + AWAIT_ARCH_SIGNOFF, + AWAIT_MILESTONE_SIGNOFF, + ): + return + self._store.append_event( + UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}") + ) + self._apply_gate_reply( + sender=comment.author, text=comment.body.strip(), channel="gitea" + ) + + 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/executor.py b/nisse-dev/src/nisse_dev/executor.py index 28c78ab..4e647ff 100644 --- a/nisse-dev/src/nisse_dev/executor.py +++ b/nisse-dev/src/nisse_dev/executor.py @@ -30,6 +30,20 @@ plain-text summary and no tool call — that ends the milestone.""" DEFAULT_MAX_TURNS = 24 DEFAULT_FAILURE_LIMIT = 2 +# Tools that change or verify something. Reading a file or listing a +# directory is the model looking around, not doing the work, so neither +# counts as evidence that a milestone was carried out. +MUTATING_TOOLS = frozenset({"write_file", "run_command"}) + +# A turn with no tool call means "the plan is done". Believing that before +# any real work has happened would record an empty milestone as a success, so +# the claim is challenged once first. +_NUDGE = ( + "\n\nYou have not made any change yet, so the plan cannot be complete. " + "Carry out the next step of the plan with a tool call, or run the tests " + "to show the work is already done." +) + @dataclass(frozen=True) class ExecutorOutcome: @@ -59,6 +73,8 @@ class LLMExecutor: model = self._roles.resolve("executor") system = f"{EXECUTOR_SYSTEM}\n\nYou are executing milestone {milestone}." consecutive_failures = 0 + did_work = False + challenged = False for _turn in range(self._max_turns): try: @@ -78,9 +94,24 @@ class LLMExecutor: calls.append(event) if not calls: - return ExecutorOutcome("completed", "executor reported the plan done") + if did_work: + return ExecutorOutcome( + "completed", "executor reported the plan done" + ) + if not challenged: + # Claiming completion without having touched anything is + # not evidence of completion; ask once before believing it. + challenged = True + system += _NUDGE + continue + return ExecutorOutcome( + "escalate", + "executor claimed the milestone was done without making any change", + ) for call in calls: + if call.tool in MUTATING_TOOLS: + did_work = True outcome = self._workspace.dispatch(call.tool, call.arguments) self._store.append_event( ToolResult( diff --git a/nisse-dev/src/nisse_dev/gitea.py b/nisse-dev/src/nisse_dev/gitea.py index 80ab20f..d5fb62b 100644 --- a/nisse-dev/src/nisse_dev/gitea.py +++ b/nisse-dev/src/nisse_dev/gitea.py @@ -151,6 +151,12 @@ class GiteaClient: created_at=data.get("created_at", ""), ) + def create_issue(self, repo: str, title: str, body: str) -> int: + data = self._request( + "POST", f"/repos/{repo}/issues", json={"title": title, "body": body} + ) + return int(data["number"]) + def close_pull_request(self, repo: str, index: int) -> None: self._request("PATCH", f"/repos/{repo}/pulls/{index}", json={"state": "closed"}) 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_approvals.py b/nisse-dev/tests/test_approvals.py new file mode 100644 index 0000000..0aa5032 --- /dev/null +++ b/nisse-dev/tests/test_approvals.py @@ -0,0 +1,210 @@ +"""Plan externalisation to Gitea and approvals arriving on either channel +(spec A4, Part H).""" + +import io +from pathlib import Path +from typing import Any + +from nisse_dev.approvals import PlanBoard +from nisse_dev.canonical import CanonicalEvent, ToolCall +from nisse_dev.control import ( + AWAIT_ARCH_SIGNOFF, + AWAIT_MILESTONE_SIGNOFF, + MILESTONE_PLANNING, + Orchestrator, +) +from nisse_dev.gitea import Comment, GiteaError +from nisse_dev.llm import RoleMap +from nisse_dev.oplog import OpLog +from nisse_dev.planner import LLMPlanningAgent +from nisse_dev.state import StateStore +from nisse_dev.transport import InMemoryTransport + +ROLES = RoleMap(executor="x", planner="y", delegate="y", consultant="y") + + +class FakeGitea: + def __init__(self, *, fail: bool = False) -> None: + self.issues: list[dict[str, str]] = [] + self.comments: list[tuple[int, str]] = [] + self._fail = fail + self._next_id = 100 + + def create_issue(self, repo: str, title: str, body: str) -> int: + if self._fail: + raise GiteaError("gitea is down") + self.issues.append({"title": title, "body": body}) + return len(self.issues) + + def post_comment(self, repo: str, index: int, body: str) -> Comment: + if self._fail: + raise GiteaError("gitea is down") + self.comments.append((index, body)) + self._next_id += 1 + return Comment( + id=self._next_id, author="nisse-dev-testing", body=body, created_at="t" + ) + + def list_comments(self, repo: str, index: int, since: str | None = None): + return [] + + +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}"}, + ) + ] + + +def build( + tmp_path: Path, gitea: Any +) -> tuple[Orchestrator, StateStore, InMemoryTransport, PlanBoard]: + store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + transport = InMemoryTransport() + board = PlanBoard(gitea, "o/sandbox") + orch = Orchestrator( + store, + transport, + LLMPlanningAgent(ScriptedClient(), store, ROLES), + operators={"don"}, + roles=ROLES, + gitea_operators={"dongar"}, + board=board, + ) + orch.start() + return orch, store, transport, board + + +def comment(body: str, author: str = "dongar", cid: int = 900) -> Comment: + return Comment(id=cid, author=author, body=body, created_at="t") + + +def test_project_start_opens_an_issue_and_posts_the_plan(tmp_path: Path) -> None: + gitea = FakeGitea() + _, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + + assert gitea.issues[0]["title"] == "nisse-dev: demo" + issue, body = gitea.comments[0] + assert issue == 1 + assert "plan 1" in body + assert "approve rev-1" in body # the tag is printed for copy-paste (A4) + # The store keeps a pointer, not the plan record itself (Part H). + assert store.get_state().data["board"]["issue"] == 1 + + +def test_approval_by_gitea_comment_satisfies_the_gate(tmp_path: Path) -> None: + gitea = FakeGitea() + orch, store, _, _ = build(tmp_path, gitea) + orch._transport.receive("don", "use demo") + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + + orch.handle_plan_comment(comment("approve rev-1")) + + assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF + + +def test_stale_revision_from_gitea_is_refused(tmp_path: Path) -> None: + gitea = FakeGitea() + orch, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + transport.receive("don", "too vague") # now at rev-2 + + orch.handle_plan_comment(comment("approve rev-1")) + assert store.get_state().state == AWAIT_ARCH_SIGNOFF # not advanced + + orch.handle_plan_comment(comment("approve rev-2", cid=901)) + assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF + + +def test_non_operator_gitea_comment_cannot_approve(tmp_path: Path) -> None: + gitea = FakeGitea() + orch, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + + orch.handle_plan_comment(comment("approve rev-1", author="drive-by")) + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + + +def test_agents_own_comment_cannot_approve_its_plan(tmp_path: Path) -> None: + # The agent posts plans as nisse-dev-testing; that login is not on the + # operator allowlist, so its own comments can never satisfy its gate (A4). + gitea = FakeGitea() + orch, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + + orch.handle_plan_comment(comment("approve rev-1", author="nisse-dev-testing")) + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + + +def test_rejection_by_gitea_comment_replans(tmp_path: Path) -> None: + gitea = FakeGitea() + orch, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + + orch.handle_plan_comment(comment("needs more detail on the schema")) + + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + assert "plan 2" in transport.outbound[-1].text # replanned + + +def test_matrix_approval_is_echoed_to_gitea(tmp_path: Path) -> None: + gitea = FakeGitea() + _, _, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + transport.receive("don", "approve") + + echoes = [body for _, body in gitea.comments if "approved by don" in body] + assert echoes, "the Gitea record must not silently disagree with Matrix" + + +def test_gitea_approval_is_echoed_to_matrix(tmp_path: Path) -> None: + gitea = FakeGitea() + orch, _, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + orch.handle_plan_comment(comment("approve rev-1")) + + assert any("approved by dongar via gitea" in m.text for m in transport.outbound) + + +def test_gitea_outage_leaves_matrix_a_complete_channel(tmp_path: Path) -> None: + gitea = FakeGitea(fail=True) + _, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + + # Planning proceeded and the gate is usable despite Gitea being down. + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + assert any("Matrix-only" in m.text for m in transport.outbound) + transport.receive("don", "approve") + assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF + + +def test_comments_are_ignored_outside_a_gate(tmp_path: Path) -> None: + gitea = FakeGitea() + orch, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + transport.receive("don", "approve") # arch approved; now at milestone gate + transport.receive("don", "approve") # milestone approved, awaiting start + before = store.get_state().state + + orch.handle_plan_comment(comment("approve rev-1", cid=902)) + assert store.get_state().state == before + + +def test_board_records_the_milestone_plan_too(tmp_path: Path) -> None: + gitea = FakeGitea() + _, store, transport, _ = build(tmp_path, gitea) + transport.receive("don", "use demo") + transport.receive("don", "approve") + + assert store.get_state().state in (MILESTONE_PLANNING, AWAIT_MILESTONE_SIGNOFF) + posted = [body for _, body in gitea.comments if "milestone-1" in body] + assert posted, "the milestone plan must be externalised as well" diff --git a/nisse-dev/tests/test_executor.py b/nisse-dev/tests/test_executor.py index 70b43f4..5d91924 100644 --- a/nisse-dev/tests/test_executor.py +++ b/nisse-dev/tests/test_executor.py @@ -88,11 +88,46 @@ def test_uses_executor_role_and_reduced_manifest( assert "milestone 2" in client.calls[0]["system"] -def test_no_tool_call_ends_the_milestone(store: StateStore, ws: Workspace) -> None: - client = ScriptedClient([[AssistantText(text="all steps done", model="m")]]) +def test_no_tool_call_after_work_ends_the_milestone( + store: StateStore, ws: Workspace +) -> None: + client = ScriptedClient( + [ + [call("write_file", path="a.py", content="x")], + [AssistantText(text="all steps done", model="m")], + ] + ) assert build(client, store, ws).run(milestone=1).status == "completed" +def test_completion_claimed_without_any_work_is_challenged( + store: StateStore, ws: Workspace +) -> None: + # A model that answers with prose on turn one has done nothing; believing + # it would record an empty milestone as a success. + client = ScriptedClient( + [ + [AssistantText(text="looks done to me", model="m")], + [call("write_file", path="a.py", content="x")], + [AssistantText(text="now it is done", model="m")], + ] + ) + outcome = build(client, store, ws).run(milestone=1) + assert outcome.status == "completed" + assert (ws.root / "a.py").exists() # the nudge produced actual work + assert "not made any change yet" in client.calls[1]["system"] + + +def test_persisting_in_an_empty_claim_escalates( + store: StateStore, ws: Workspace +) -> None: + prose = [AssistantText(text="nothing to do", model="m")] + client = ScriptedClient([prose, prose]) + outcome = build(client, store, ws).run(milestone=1) + assert outcome.status == "escalate" + assert "without making any change" in outcome.note + + def test_two_consecutive_command_failures_escalate( store: StateStore, ws: Workspace ) -> None: @@ -126,11 +161,14 @@ def test_a_success_resets_the_failure_count(store: StateStore, ws: Workspace) -> def test_failed_reads_do_not_count_toward_escalation( store: StateStore, ws: Workspace ) -> None: + # Two failed reads in a row must not trip the failure limit: looking for + # a file that is not there is the model orienting itself, not failing. client = ScriptedClient( [ [call("read_file", path="missing-a.txt")], [call("read_file", path="missing-b.txt")], - [AssistantText(text="looked around, done", model="m")], + [call("write_file", path="a.py", content="x")], + [AssistantText(text="done", model="m")], ] ) assert build(client, store, ws, failure_limit=2).run(milestone=1).status == ( @@ -138,6 +176,20 @@ def test_failed_reads_do_not_count_toward_escalation( ) +def test_reading_alone_is_not_doing_the_work(store: StateStore, ws: Workspace) -> None: + # Exploration is not evidence a milestone was carried out, so a + # completion claim backed only by reads is still challenged. + client = ScriptedClient( + [ + [call("list_dir", path=".")], + [AssistantText(text="looks done to me", model="m")], + [AssistantText(text="really, done", model="m")], + ] + ) + outcome = build(client, store, ws).run(milestone=1) + assert outcome.status == "escalate" + + def test_turn_cap_is_enforced(store: StateStore, ws: Workspace) -> None: forever = [[call("list_dir", path=".")] for _ in range(50)] client = ScriptedClient(forever) 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 diff --git a/nisse-dev/tests/test_tools.py b/nisse-dev/tests/test_tools.py index 5e66c4f..6827c6b 100644 --- a/nisse-dev/tests/test_tools.py +++ b/nisse-dev/tests/test_tools.py @@ -1,4 +1,5 @@ import os +import time from pathlib import Path import pytest @@ -97,6 +98,12 @@ def test_timeout_kills_the_whole_process_group(tmp_path: Path) -> None: assert pidfile.exists() pid = int(pidfile.read_text().strip()) assert _is_running(os.getpid()), "liveness check is broken, not the kill" + # SIGKILL to a process group is asynchronous, and the grandchild is an + # orphan by now, so its exit is not synchronised with our return. Wait + # for death rather than assuming it is instantaneous. + deadline = time.monotonic() + 5 + while _is_running(pid) and time.monotonic() < deadline: + time.sleep(0.05) assert not _is_running(pid) diff --git a/tests/integration/test_gitea_live.py b/tests/integration/test_gitea_live.py index f13dbf0..c1f3d0f 100644 --- a/tests/integration/test_gitea_live.py +++ b/tests/integration/test_gitea_live.py @@ -12,7 +12,7 @@ from pathlib import Path import pytest from nisse_dev.envfile import load_env_file -from nisse_dev.git import GitRepo, authenticated_url +from nisse_dev.git import GitError, GitRepo, authenticated_url from nisse_dev.gitea import GiteaClient, pull_request_reconciler from nisse_dev.state import Intent @@ -88,6 +88,19 @@ def test_push_open_pr_and_reconcile(tmp_path: Path) -> None: # Now the same reconciler must see the effect and refuse to repeat it. assert reconcile(intent_for(branch)) is True + + # A4 is only real if the server enforces it. The agent's own token + # must not be able to merge the PR it just opened, and must not be + # able to push to the protected default branch at all. + merged = client._http.post( + f"/repos/{repo_path}/pulls/{pr.number}/merge", json={"Do": "merge"} + ) + assert merged.status_code >= 400, ( + "the agent merged its own PR: branch protection is not enforcing " + "the human gate" + ) + with pytest.raises(GitError): + repo.push(url, "main", token=token) finally: if pr is not None: client.close_pull_request(repo_path, pr.number)