Files
nisse/tests/integration/test_end_to_end_loop.py
T
dongarandClaude Opus 5.5 3f54896f65
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 5m14s
Conversations on the new loop (M2.5, PR A)
- A top-level message that isn't a command is a conversation: an item in
  the `general` scope (L1), threaded under the operator's message, with
  `reply`, `ask_human` and `propose_work` and no workspace or execution.
  `general` is reserved as a project name.
- propose_work(project, title, summary) goes up as a revision for a
  go-ahead (or a plain yes). On go, under a file_issue intent (L8), the
  record issue is opened in that project, assigned to the agent and
  carrying the new item's ID; its item starts agreed, with the summary as
  its brief. Intake skips any issue carrying an ID (D1); boot adopts a
  filed issue or puts the proposal up again.
- A conversation closes after tuning.loop.general_idle_s (default a day);
  a reply in its thread then starts a new conversation that continues it.

Tests: conversation unit tests (threading, filing, the marker, an unknown
project, changes, idle close and continuation, a plain yes, both crash
cases, commands at top level); the walk with conversations, proposals and
idle time, and records_have_one_item / conversations_do_no_work; an e2e
conversation that becomes filed work.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
2026-10-06 04:26:40 +00:00

538 lines
21 KiB
Python

"""The end-to-end tests on the new loop (Development Plan M2.1, M2.2): the
whole slice, from an issue assigned in Gitea to a merged PR and a closed work
item, and the queue, with an issue blocked on another until it closes; with
real models, the Gitea sandbox and a real build host.
A scripted operator plays the operator's part: in Gitea through its own test
account (NISSE_OPERATOR_TOKEN; an operator only in nisse-test.yaml), and in
the item's Matrix thread through the in-memory transport. Run with
scripts/e2e.sh after each milestone.
"""
import base64
import contextlib
import os
import time
import uuid
from collections.abc import Iterator
from dataclasses import replace
from fnmatch import fnmatch
from pathlib import Path
from typing import Any
import pytest
from e2e_support import base_branch, dependent_task, operator_gitea, task
from nisse_dev.config import AgentConfig, LoopTuning
from nisse_dev.execution import SSHHost
from nisse_dev.gitea import GiteaClient, GiteaError
from nisse_dev.loop.build import RunningLoop, build_loop_agent
from nisse_dev.loop.items import (
ANSWER,
BLOCKED,
GO_AHEAD,
MERGE,
WAITING,
item_for_issue,
loop_of,
)
from nisse_dev.transport import InMemoryTransport
pytestmark = [pytest.mark.integration, pytest.mark.e2e]
OPERATOR = "@dongar:dongar.net" # an allowlisted Matrix operator in nisse-test.yaml
DEADLINE_S = 1200
MAX_ANSWERS = 5 # per wait for something else, before giving up
BUG_GRACE_S = 20
@pytest.fixture
def running(
live_config: AgentConfig,
ssh_host: SSHHost,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> Iterator[tuple[RunningLoop, InMemoryTransport, str]]:
from test_end_to_end import build_host
run = uuid.uuid4().hex[:8]
stack = contextlib.ExitStack()
# Kept outside pytest's temp directories, which pytest prunes, so a
# failed run's transcript can be read afterwards (nisse_dev.dump).
keep = Path(os.environ.get("NISSE_E2E_RUNS", "/tmp/nisse-e2e-runs")) / run
keep.mkdir(parents=True, exist_ok=True)
print(f"\ne2e run {run}: state store in {keep}")
operator = operator_gitea(live_config.gitea_url)
base = stack.enter_context(base_branch(operator, live_config.gitea_repo, run))
config = replace(
live_config.with_project(
base_branch=base,
workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
build_host=build_host(ssh_host),
),
loop="new",
state_db=keep / "state.db",
# Short, so a bug's grace period runs out within the test.
tuning=replace(live_config.tuning, loop=LoopTuning(bug_grace_s=BUG_GRACE_S)),
)
transport = InMemoryTransport()
monkeypatch.setattr("nisse_dev.loop.build.MatrixTransport", lambda cfg: transport)
agent = None
try:
agent = build_loop_agent(config)
# The test drives intake and PR watching itself, not on a timer.
agent.loop._interval = 3600
agent.start()
yield agent, transport, run
finally:
if agent is not None:
agent.stop()
host = SSHHost(
config.build_host.address,
key=config.build_host.key,
user=config.build_host.user,
port=config.build_host.port,
state_dir=tmp_path / "ssh",
)
host.execute(f"rm -rf /tmp/nisse-e2e-{run}")
stack.close() # the run's base branch
def is_artifact(path: str) -> bool:
parts = path.split("/")
return (
"__pycache__" in parts
or ".pytest_cache" in parts
or fnmatch(parts[-1], "*.py[co]")
)
def merge(gitea: GiteaClient, repo: str, pr: int, wait_s: float = 60) -> None:
"""Merge, waiting out Gitea's "Please try again later": it answers that
while it is still checking a freshly pushed PR can merge."""
deadline = time.monotonic() + wait_s
while True:
try:
gitea._request(
"POST", f"/repos/{repo}/pulls/{pr}/merge", json={"Do": "merge"}
)
return
except GiteaError as exc:
if "try again later" not in str(exc) or time.monotonic() > deadline:
raise
time.sleep(3)
class Operator:
def __init__(self, agent: RunningLoop, transport: InMemoryTransport) -> None:
self.agent = agent
self.transport = transport
def settle(self) -> None:
self.agent.worker.call(lambda: None, timeout=DEADLINE_S)
def tick(self) -> None:
"""Intake and PR watching, as the agent's timer would do."""
self.agent.worker.call(self.agent.loop.intake, timeout=DEADLINE_S)
def item(self, issue: int = 0) -> Any:
"""The work item for ``issue``; by default the first one."""
if issue:
store = self.agent.store
return self.agent.worker.call(lambda: item_for_issue(store, issue))
items = self.agent.worker.call(self.agent.store.work_items)
return items[0] if items else None
def say(self, text: str, issue: int = 0) -> None:
thread = loop_of(self.item(issue)).thread
self.transport.receive(OPERATOR, text, thread=thread)
self.settle()
def drive_until(self, wait: str, issue: int = 0, *, go: bool = True) -> Any:
"""Answer whatever the agent asks until it waits for ``wait``.
Without ``go``, a restatement is left for its grace period."""
deadline = time.monotonic() + DEADLINE_S
answers = 0
while time.monotonic() < deadline:
item = self.item(issue)
assert item is not None, "no work item"
loop = loop_of(item)
if item.position == WAITING and loop.wait == wait:
return item
if item.status == "closed":
raise AssertionError(f"closed early: {self.last()}")
if wait == GO_AHEAD and loop.started:
raise AssertionError(f"went past the go-ahead: {self.last()}")
if item.position == WAITING and loop.wait == GO_AHEAD and not go:
assert loop.grace_until, f"no grace period: {self.last()}"
time.sleep(2)
self.tick() # what the agent's timer would do
elif item.position == WAITING and loop.wait == GO_AHEAD:
self.say("go", issue)
elif item.position == WAITING and loop.wait == ANSWER:
answers += 1
# Answering forever is how a stuck agent looks: give up.
assert answers <= MAX_ANSWERS, f"stuck asking: {self.last()}"
self.say("Use your best judgement and proceed.", issue)
else:
self.settle()
raise AssertionError(f"never waited for {wait}: {self.last()}")
def merge_and_close(self, issue: int) -> None:
"""Approve and merge the item's PR, and see the agent close the item."""
gitea, repo = self.agent.gitea, self.agent.config.gitea_repo
as_operator = operator_gitea(self.agent.config.gitea_url)
pr = loop_of(self.item(issue)).pr
with contextlib.suppress(GiteaError): # approval may not be required
as_operator._request(
"POST",
f"/repos/{repo}/pulls/{pr}/reviews",
json={"event": "APPROVED", "body": "looks good"},
)
merge(as_operator, repo, pr)
self.tick()
deadline = time.monotonic() + DEADLINE_S
nudges = 0
while self.item(issue).status != "closed" and time.monotonic() < deadline:
if loop_of(self.item(issue)).wait == ANSWER:
nudges += 1
assert nudges <= MAX_ANSWERS, f"never closed: {self.last()}"
self.say("It's merged; please close the item.", issue)
self.settle()
assert self.item(issue).outcome == "completed"
state = gitea._request("GET", f"/repos/{repo}/issues/{issue}")["state"]
assert state == "closed"
def tidy(self, issues: list[int], prs: list[int]) -> None:
"""Leave the sandbox tidy: whatever is still open gets closed."""
gitea, repo = self.agent.gitea, self.agent.config.gitea_repo
as_operator = operator_gitea(self.agent.config.gitea_url)
for issue in issues:
item = self.item(issue)
if item is not None and item.status == "open":
self.say("abandon", issue)
with contextlib.suppress(GiteaError):
as_operator.close_issue(repo, issue)
for pr in prs:
with contextlib.suppress(GiteaError):
head = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["ref"]
as_operator._request("DELETE", f"/repos/{repo}/branches/{head}")
def last(self) -> str:
return self.transport.outbound[-1].text if self.transport.outbound else ""
def test_the_slice_end_to_end(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
as_operator = operator_gitea(agent.config.gitea_url)
issue = None
pr = 0
try:
# --- the operator files an issue and assigns it to the agent (D1) ---
issue = as_operator.create_issue(repo, f"vowels_{run}", task(run))
as_operator._request(
"PATCH",
f"/repos/{repo}/issues/{issue}",
json={"assignees": [agent.account]},
)
operator.tick()
item = operator.item()
assert item is not None and loop_of(item).issue == issue
# --- restatement, go-ahead, milestone, PR (L3, D4) ---
item = operator.drive_until(MERGE)
pr = loop_of(item).pr
files = gitea._request("GET", f"/repos/{repo}/pulls/{pr}/files")
changed = [f["filename"] for f in files]
assert changed and not [p for p in changed if is_artifact(p)], changed
body = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["body"]
assert f"Closes #{issue}" in body
# --- a review comment is answered with a fix (D5) ---
# A request that can't already be met, so a fix must follow.
marker = f"reviewed-{uuid.uuid4().hex[:8]}"
before = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["sha"]
as_operator.post_comment(
repo, pr, f"Please add the comment line `# {marker}` above vowels_{run}."
)
operator.tick()
operator.drive_until(MERGE)
after = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["sha"]
assert after != before, "no fix was pushed for the review comment"
diff = gitea._http.get(f"/repos/{repo}/pulls/{pr}.diff").text
assert marker in diff, "the pushed fix doesn't do what the comment asked"
# --- the operator approves and merges; the agent closes the item ---
operator.merge_and_close(issue)
finally:
operator.tidy([issue] if issue is not None else [], [pr] if pr else [])
def test_an_issue_blocked_on_another_waits_for_it(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
"""M2.2: B depends on A in Gitea. B is blocked until A's PR merges and A
closes; then B starts from a workspace that has A's work."""
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
as_operator = operator_gitea(agent.config.gitea_url)
owner, name = repo.split("/")
issues: list[int] = []
prs: list[int] = []
try:
a = as_operator.create_issue(repo, f"vowels_{run}", task(run))
issues.append(a)
b = as_operator.create_issue(repo, f"consonants_{run}", dependent_task(run))
issues.append(b)
as_operator._request(
"POST",
f"/repos/{repo}/issues/{b}/dependencies",
json={"owner": owner, "repo": name, "index": a},
)
for n in (b, a): # B first: the dependency, not the order, decides
as_operator._request(
"PATCH",
f"/repos/{repo}/issues/{n}",
json={"assignees": [agent.account]},
)
operator.tick()
assert operator.item(b).position == BLOCKED
assert loop_of(operator.item(b)).blocked_by == (f"{repo}#{a}",)
# --- A runs to a merged PR; B stays blocked meanwhile ---
item = operator.drive_until(MERGE, a)
prs.append(loop_of(item).pr)
assert operator.item(b).position == BLOCKED
operator.merge_and_close(a)
# --- A closed: B is unblocked and starts from A's work ---
operator.tick()
assert operator.item(b).position != BLOCKED
item = operator.drive_until(MERGE, b)
prs.append(loop_of(item).pr)
files = gitea._request("GET", f"/repos/{repo}/pulls/{prs[-1]}/files")
diff = gitea._http.get(f"/repos/{repo}/pulls/{prs[-1]}.diff").text
assert f"consonants_{run}" in diff, [f["filename"] for f in files]
# Built on A's merged work, so A's function isn't added again.
assert f"\n+def vowels_{run}" not in diff
finally:
operator.tidy(issues, prs)
def label(gitea: GiteaClient, repo: str, name: str) -> int:
"""The repository's label ``name``, made if it's missing."""
for existing in gitea._request("GET", f"/repos/{repo}/labels"):
if existing["name"] == name:
return int(existing["id"])
made = gitea._request(
"POST", f"/repos/{repo}/labels", json={"name": name, "color": "#ee0701"}
)
return int(made["id"])
def test_a_bug_goes_ahead_after_its_grace_period(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
"""M2.3: a `bug` issue is reproduced and restated, and work starts when
the grace period runs out, with no go from the operator."""
agent, transport, run = running
operator = Operator(agent, transport)
repo, base = agent.config.gitea_repo, agent.config.base_branch
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
try:
# A real bug on this run's base branch, for the agent to reproduce.
code = (
f"def last_word_{run}(text):\n"
' """The last word of text, or "" if it has none."""\n'
" return text.split()[-1]\n"
)
as_operator._request(
"POST",
f"/repos/{repo}/contents/sandbox/last_word_{run}.py",
json={
"branch": base,
"content": base64.b64encode(code.encode()).decode(),
"message": f"last_word_{run}, with a bug",
},
)
data = as_operator._request(
"POST",
f"/repos/{repo}/issues",
json={
"title": f"last_word_{run} crashes on empty text",
"body": f"`last_word_{run}('')` in sandbox/last_word_{run}.py "
'raises IndexError; it should return "" as its docstring says.',
"labels": [label(as_operator, repo, "bug")],
"assignees": [agent.account],
},
)
issues.append(int(data["number"]))
operator.tick()
assert loop_of(operator.item(issues[0])).labels == ("bug",)
item = operator.drive_until(MERGE, issues[0], go=False)
prs.append(loop_of(item).pr)
orchestrator = agent.worker.call(
lambda: agent.store.transcript(item.id, "orchestrator").read()
)
said = [getattr(e.event, "text", "") for e in orchestrator]
assert any(t.endswith("from the bug grace period") for t in said), said
finally:
operator.tidy(issues, prs)
def test_a_plain_yes_is_a_go_ahead(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
"""M2.3: "yep, that's right" to a restatement is read as a go-ahead,
and the agent says so before acting."""
agent, transport, run = running
operator = Operator(agent, transport)
repo = agent.config.gitea_repo
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
try:
issues.append(as_operator.create_issue(repo, f"vowels_{run}", task(run)))
as_operator._request(
"PATCH",
f"/repos/{repo}/issues/{issues[0]}",
json={"assignees": [agent.account]},
)
operator.tick()
for _ in range(MAX_ANSWERS):
operator.drive_until(GO_AHEAD, issues[0])
if "?" not in operator.last():
break
# A restatement with a question in it can't be agreed to as it
# stands: answer first (the reader rightly says no to a yes).
operator.say(
f"Not yet: a new function vowels_{run}, not reusing anything "
"that exists. Put up a new restatement with that.",
issues[0],
)
operator.say("Yep, that's right, sounds good.", issues[0])
said = [m.text for m in transport.outbound]
assert any(t.startswith("Taking that as a go-ahead on rev-") for t in said)
assert loop_of(operator.item(issues[0])).started
pr = loop_of(operator.item(issues[0])).pr
if pr:
prs.append(pr)
finally:
operator.tidy(issues, prs)
def test_a_two_milestone_feature(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
"""M2.4: work in two milestones. The first PR carries the build plan and
is part of the issue; the second starts from the merged first and closes
it; the item closes after both merge."""
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
try:
body = (
f"{task(run)}\n\nThen, separately, {dependent_task(run)[0].lower()}"
f"{dependent_task(run)[1:]}\n\nDo this in two milestones, one PR each: "
f"vowels_{run} first, then consonants_{run} after the first has merged. "
"Write a short build plan for it."
)
issues.append(as_operator.create_issue(repo, f"letters_{run}", body))
as_operator._request(
"PATCH",
f"/repos/{repo}/issues/{issues[0]}",
json={"assignees": [agent.account]},
)
operator.tick()
# --- milestone 1: the plan and vowels; part of the issue ---
item = operator.drive_until(MERGE, issues[0])
first = loop_of(item).pr
prs.append(first)
pr = gitea._request("GET", f"/repos/{repo}/pulls/{first}")
assert f"Part of #{issues[0]}" in pr["body"], pr["body"]
files = [
f["filename"]
for f in gitea._request("GET", f"/repos/{repo}/pulls/{first}/files")
]
assert any(f.startswith("docs/plans/") for f in files), files
merge(as_operator, repo, first)
operator.tick()
# --- milestone 2: from the merged first; closes the issue ---
item = operator.drive_until(MERGE, issues[0])
second = loop_of(item).pr
assert second != first
prs.append(second)
pr = gitea._request("GET", f"/repos/{repo}/pulls/{second}")
assert f"Closes #{issues[0]}" in pr["body"], pr["body"]
diff = gitea._http.get(f"/repos/{repo}/pulls/{second}.diff").text
assert f"consonants_{run}" in diff
assert f"\n+def vowels_{run}" not in diff # built on the merged first
operator.merge_and_close(issues[0])
finally:
operator.tidy(issues, prs)
def test_a_conversation_becomes_filed_work(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
"""M2.5: a top-level message is a conversation; the work it turns into
is proposed, agreed, filed as a record issue, and started as agreed."""
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
issues: list[int] = []
prs: list[int] = []
def conversation() -> Any:
items = agent.worker.call(agent.store.work_items)
return next(i for i in items if i.scope == "general")
try:
root = transport.receive(
OPERATOR,
f"I've been thinking the sandbox could use this: {task(run)} "
"Could you file that as an issue and start on it?",
)
operator.settle()
deadline = time.monotonic() + DEADLINE_S
answers = 0
while loop_of(conversation()).wait != GO_AHEAD:
assert time.monotonic() < deadline, f"never proposed: {operator.last()}"
answers += 1
assert answers <= MAX_ANSWERS, f"never proposed: {operator.last()}"
transport.receive(
OPERATOR,
"Yes, propose it as an issue in the sandbox project.",
thread=root,
)
operator.settle()
assert operator.last().startswith(f"rev-{loop_of(conversation()).rev}:")
transport.receive(OPERATOR, "go", thread=root)
operator.settle()
[filed] = loop_of(conversation()).became
number = int(filed.split("#")[1])
issues.append(number)
data = gitea._request("GET", f"/repos/{repo}/issues/{number}")
assert "<!-- nisse-item: " in data["body"]
assert agent.account in [a["login"] for a in data["assignees"] or []]
item = operator.drive_until(MERGE, number)
assert loop_of(item).started # no second restatement
prs.append(loop_of(item).pr)
pr = gitea._request("GET", f"/repos/{repo}/pulls/{prs[0]}")
assert f"Closes #{number}" in pr["body"]
finally:
operator.tidy(issues, prs)