Add PR comment intake and the review loop (B2) — step 6b #17
No files matched your search
@@ -31,3 +31,6 @@ NISSE_GITEA_URL=https://gitea.example
|
|||||||
NISSE_GITEA_TOKEN=your-scoped-service-account-token
|
NISSE_GITEA_TOKEN=your-scoped-service-account-token
|
||||||
# Sandbox repo for live tests; never the real project repo.
|
# Sandbox repo for live tests; never the real project repo.
|
||||||
NISSE_GITEA_TEST_REPO=owner/sandbox-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
|
||||||
@@ -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
|
||||||
|
]
|
||||||
@@ -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.
|
store before the loop continues, so a restart lands mid-gate, not mid-air.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import threading
|
||||||
import uuid
|
import uuid
|
||||||
from collections.abc import Iterable, Mapping
|
from collections.abc import Iterable, Mapping
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
@@ -86,6 +87,23 @@ class MilestonePublisher(Protocol):
|
|||||||
self, *, task_id: str, milestone: int, project: str, plan: str
|
self, *, task_id: str, milestone: int, project: str, plan: str
|
||||||
) -> Any: ...
|
) -> 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):
|
class PlanningAgent(Protocol):
|
||||||
def plan(
|
def plan(
|
||||||
@@ -126,6 +144,10 @@ def _fresh_data() -> dict[str, Any]:
|
|||||||
"task_id": None,
|
"task_id": None,
|
||||||
"roles": None, # A11 mapping, pinned at task start
|
"roles": None, # A11 mapping, pinned at task start
|
||||||
"pr": None, # the open PR for the current milestone, once published
|
"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,
|
roles: RoleMap | None = None,
|
||||||
executor: "MilestoneExecutor | None" = None,
|
executor: "MilestoneExecutor | None" = None,
|
||||||
publisher: "MilestonePublisher | None" = None,
|
publisher: "MilestonePublisher | None" = None,
|
||||||
|
gitea_operators: Iterable[str] = (),
|
||||||
|
board: "PlanBoard | None" = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
self._store = store
|
self._store = store
|
||||||
self._transport = transport
|
self._transport = transport
|
||||||
@@ -147,6 +171,14 @@ class Orchestrator:
|
|||||||
self._executor = executor
|
self._executor = executor
|
||||||
self._publisher = publisher
|
self._publisher = publisher
|
||||||
self._operators = set(operators)
|
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._data = _fresh_data()
|
||||||
self._project: str | None = None
|
self._project: str | None = None
|
||||||
transport.on_message(self._handle)
|
transport.on_message(self._handle)
|
||||||
@@ -197,6 +229,10 @@ class Orchestrator:
|
|||||||
)
|
)
|
||||||
|
|
||||||
def _handle(self, msg: InboundMessage) -> None:
|
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:
|
if msg.sender not in self._operators:
|
||||||
return
|
return
|
||||||
# Every operator message enters the canonical transcript, so planning
|
# 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._data = {**_fresh_data(), "task_id": task_id, "roles": pinned}
|
||||||
self._project = project
|
self._project = project
|
||||||
self._store.oplog.task_id = task_id
|
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._transition(PLANNING_ARCH, cause=f"operator command: use {project}")
|
||||||
self._run_planning_turn("arch")
|
self._run_planning_turn("arch")
|
||||||
|
|
||||||
@@ -289,6 +338,7 @@ class Orchestrator:
|
|||||||
f"{action.content}\n\nreply 'approve' to approve rev-{rev}; "
|
f"{action.content}\n\nreply 'approve' to approve rev-{rev}; "
|
||||||
"any other reply is rejection feedback"
|
"any other reply is rejection feedback"
|
||||||
)
|
)
|
||||||
|
self._post_plan_to_board(gate_name, rev, action.content)
|
||||||
self._store.oplog.emit(
|
self._store.oplog.emit(
|
||||||
"gate.open",
|
"gate.open",
|
||||||
cause=f"plan submitted (rev-{rev})",
|
cause=f"plan submitted (rev-{rev})",
|
||||||
@@ -298,6 +348,12 @@ class Orchestrator:
|
|||||||
)
|
)
|
||||||
|
|
||||||
def _handle_gate_reply(self, msg: InboundMessage, text: str) -> None:
|
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"]
|
gate_data = self._data["gate"]
|
||||||
name, rev = gate_data["name"], gate_data["rev"]
|
name, rev = gate_data["name"], gate_data["rev"]
|
||||||
milestone = self._data["milestone"]
|
milestone = self._data["milestone"]
|
||||||
@@ -313,9 +369,8 @@ class Orchestrator:
|
|||||||
)
|
)
|
||||||
return
|
return
|
||||||
revision = PlanRevision(gate=name, rev=rev, content=gate_data["plan"])
|
revision = PlanRevision(gate=name, rev=rev, content=gate_data["plan"])
|
||||||
result = Gate(revision, self._operators).evaluate(
|
allowlist = self._operators if channel == "transport" else self._gitea_operators
|
||||||
msg.sender, text, channel="transport"
|
result = Gate(revision, allowlist).evaluate(sender, text, channel=channel)
|
||||||
)
|
|
||||||
if isinstance(result, Ignored):
|
if isinstance(result, Ignored):
|
||||||
if result.notice:
|
if result.notice:
|
||||||
self._transport.send_message(result.notice)
|
self._transport.send_message(result.notice)
|
||||||
@@ -328,12 +383,19 @@ class Orchestrator:
|
|||||||
rev=f"rev-{rev}",
|
rev=f"rev-{rev}",
|
||||||
channel=result.channel,
|
channel=result.channel,
|
||||||
)
|
)
|
||||||
|
self._echo_outcome(
|
||||||
|
channel,
|
||||||
|
f"{name} rev-{rev} rejected by {result.approver}: {result.feedback}",
|
||||||
|
)
|
||||||
self._data["gate"] = None
|
self._data["gate"] = None
|
||||||
back = PLANNING_ARCH if name == "arch" else MILESTONE_PLANNING
|
back = PLANNING_ARCH if name == "arch" else MILESTONE_PLANNING
|
||||||
self._transition(back, cause=f"plan rejected ({name} rev-{rev})")
|
self._transition(back, cause=f"plan rejected ({name} rev-{rev})")
|
||||||
self._run_planning_turn(name, feedback=result.feedback)
|
self._run_planning_turn(name, feedback=result.feedback)
|
||||||
return
|
return
|
||||||
assert isinstance(result, Approved)
|
assert isinstance(result, Approved)
|
||||||
|
self._echo_outcome(
|
||||||
|
channel, f"{name} rev-{rev} approved by {result.approver} via {channel}"
|
||||||
|
)
|
||||||
self._store.oplog.emit(
|
self._store.oplog.emit(
|
||||||
"gate.approve",
|
"gate.approve",
|
||||||
cause=f"approved by {result.approver}",
|
cause=f"approved by {result.approver}",
|
||||||
@@ -376,6 +438,118 @@ class Orchestrator:
|
|||||||
pending={"kind": "milestone_stopped", "milestone": milestone},
|
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:
|
def _publish_milestone(self, milestone: int) -> None:
|
||||||
if self._publisher is None:
|
if self._publisher is None:
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -30,6 +30,20 @@ plain-text summary and no tool call — that ends the milestone."""
|
|||||||
DEFAULT_MAX_TURNS = 24
|
DEFAULT_MAX_TURNS = 24
|
||||||
DEFAULT_FAILURE_LIMIT = 2
|
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)
|
@dataclass(frozen=True)
|
||||||
class ExecutorOutcome:
|
class ExecutorOutcome:
|
||||||
@@ -59,6 +73,8 @@ class LLMExecutor:
|
|||||||
model = self._roles.resolve("executor")
|
model = self._roles.resolve("executor")
|
||||||
system = f"{EXECUTOR_SYSTEM}\n\nYou are executing milestone {milestone}."
|
system = f"{EXECUTOR_SYSTEM}\n\nYou are executing milestone {milestone}."
|
||||||
consecutive_failures = 0
|
consecutive_failures = 0
|
||||||
|
did_work = False
|
||||||
|
challenged = False
|
||||||
|
|
||||||
for _turn in range(self._max_turns):
|
for _turn in range(self._max_turns):
|
||||||
try:
|
try:
|
||||||
@@ -78,9 +94,24 @@ class LLMExecutor:
|
|||||||
calls.append(event)
|
calls.append(event)
|
||||||
|
|
||||||
if not calls:
|
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:
|
for call in calls:
|
||||||
|
if call.tool in MUTATING_TOOLS:
|
||||||
|
did_work = True
|
||||||
outcome = self._workspace.dispatch(call.tool, call.arguments)
|
outcome = self._workspace.dispatch(call.tool, call.arguments)
|
||||||
self._store.append_event(
|
self._store.append_event(
|
||||||
ToolResult(
|
ToolResult(
|
||||||
|
|||||||
@@ -151,6 +151,12 @@ class GiteaClient:
|
|||||||
created_at=data.get("created_at", ""),
|
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:
|
def close_pull_request(self, repo: str, index: int) -> None:
|
||||||
self._request("PATCH", f"/repos/{repo}/pulls/{index}", json={"state": "closed"})
|
self._request("PATCH", f"/repos/{repo}/pulls/{index}", json={"state": "closed"})
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
@@ -10,7 +10,12 @@ an idempotent transition reconciled against external truth.
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
from nisse_dev.git import GitRepo, authenticated_url
|
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.reconcile import Reconciler
|
||||||
from nisse_dev.state import StateStore
|
from nisse_dev.state import StateStore
|
||||||
|
|
||||||
@@ -86,6 +91,32 @@ class MilestonePublisher:
|
|||||||
self._store.commit_intent(intent_id, cause=f"opened PR #{pr.number}")
|
self._store.commit_intent(intent_id, cause=f"opened PR #{pr.number}")
|
||||||
return PublishResult(pr, f"opened PR #{pr.number}: {pr.url}")
|
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:
|
def _pr_body(milestone: int, plan: str, service_url: str | None) -> str:
|
||||||
parts = [
|
parts = [
|
||||||
|
|||||||
@@ -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"
|
||||||
@@ -88,11 +88,46 @@ def test_uses_executor_role_and_reduced_manifest(
|
|||||||
assert "milestone 2" in client.calls[0]["system"]
|
assert "milestone 2" in client.calls[0]["system"]
|
||||||
|
|
||||||
|
|
||||||
def test_no_tool_call_ends_the_milestone(store: StateStore, ws: Workspace) -> None:
|
def test_no_tool_call_after_work_ends_the_milestone(
|
||||||
client = ScriptedClient([[AssistantText(text="all steps done", model="m")]])
|
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"
|
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(
|
def test_two_consecutive_command_failures_escalate(
|
||||||
store: StateStore, ws: Workspace
|
store: StateStore, ws: Workspace
|
||||||
) -> None:
|
) -> 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(
|
def test_failed_reads_do_not_count_toward_escalation(
|
||||||
store: StateStore, ws: Workspace
|
store: StateStore, ws: Workspace
|
||||||
) -> None:
|
) -> 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(
|
client = ScriptedClient(
|
||||||
[
|
[
|
||||||
[call("read_file", path="missing-a.txt")],
|
[call("read_file", path="missing-a.txt")],
|
||||||
[call("read_file", path="missing-b.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 == (
|
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:
|
def test_turn_cap_is_enforced(store: StateStore, ws: Workspace) -> None:
|
||||||
forever = [[call("list_dir", path=".")] for _ in range(50)]
|
forever = [[call("list_dir", path=".")] for _ in range(50)]
|
||||||
client = ScriptedClient(forever)
|
client = ScriptedClient(forever)
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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
|
||||||
@@ -1,4 +1,5 @@
|
|||||||
import os
|
import os
|
||||||
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@@ -97,6 +98,12 @@ def test_timeout_kills_the_whole_process_group(tmp_path: Path) -> None:
|
|||||||
assert pidfile.exists()
|
assert pidfile.exists()
|
||||||
pid = int(pidfile.read_text().strip())
|
pid = int(pidfile.read_text().strip())
|
||||||
assert _is_running(os.getpid()), "liveness check is broken, not the kill"
|
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)
|
assert not _is_running(pid)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ from pathlib import Path
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from nisse_dev.envfile import load_env_file
|
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.gitea import GiteaClient, pull_request_reconciler
|
||||||
from nisse_dev.state import Intent
|
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.
|
# Now the same reconciler must see the effect and refuse to repeat it.
|
||||||
assert reconcile(intent_for(branch)) is True
|
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:
|
finally:
|
||||||
if pr is not None:
|
if pr is not None:
|
||||||
client.close_pull_request(repo_path, pr.number)
|
client.close_pull_request(repo_path, pr.number)
|
||||||
|
|||||||
Reference in new issue
Block a user