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

Merged
dev-agent merged 5 commits from review-loop into main 2026-09-07 22:55:12 +00:00
13 changed files with 1036 additions and 9 deletions

No files matched your search

+3
View File
@@ -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
+80
View File
@@ -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
]
+177 -3
View File
@@ -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
+32 -1
View File
@@ -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(
+6
View File
@@ -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"})
+104
View File
@@ -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)
+32 -1
View File
@@ -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 = [
+210
View File
@@ -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"
+55 -3
View File
@@ -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)
+132
View File
@@ -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
+184
View File
@@ -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
+7
View File
@@ -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)
+14 -1
View File
@@ -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)