diff --git a/.env.example b/.env.example index f5861a8..7e708cc 100644 --- a/.env.example +++ b/.env.example @@ -11,6 +11,9 @@ NISSE_LLM_TOKEN=replace-me NISSE_MATRIX_TOKEN=replace-me # gitea.token: the Gitea service account's scoped token (B2) NISSE_GITEA_TOKEN=replace-me +# The end-to-end tests' scripted operator (nisse-testing-operator): scopes +# write:issue and write:repository, write access to the sandbox only. +NISSE_OPERATOR_TOKEN=replace-me # The live push test reads the sandbox's deploy key from the path in # tests/integration/nisse-test.yaml (gitea.push_key), not from here. diff --git a/nisse-dev/src/nisse_dev/loop/agent.py b/nisse-dev/src/nisse_dev/loop/agent.py index 7ed653a..0a3f633 100644 --- a/nisse-dev/src/nisse_dev/loop/agent.py +++ b/nisse-dev/src/nisse_dev/loop/agent.py @@ -70,8 +70,8 @@ After the go-ahead, work in a milestone: call start_milestone before changing anything (never create or switch branches yourself), make the change, run the project's tests, then call finish_milestone with a summary. The harness commits, pushes and opens a PR, and you wait for review. Review -comments arrive as messages: make the changes, run the tests, and call -finish_milestone again to push the fix. When the PR has merged, call +comments arrive as messages: make the changes within the same milestone (don't +start another), run the tests, and call finish_milestone again to push the fix. When the PR has merged, call close_item with what was done, what was verified, and what is left open. If the PR is closed without merging, ask the operator what next. @@ -561,8 +561,13 @@ class LoopAgent: f"PR #{loop.pr}: {fixed.note}", ) elif fixed.kind == "new-pr" and fixed.pr is not None: + # Later fixes go to the new PR's branch, never the merged one. loop = replace( - loop, pr=fixed.pr.number, pr_url=fixed.pr.url, fixes=loop.fixes + 1 + loop, + pr=fixed.pr.number, + pr_url=fixed.pr.url, + branch=fixed.branch, + fixes=loop.fixes + 1, ) self._wait( item, @@ -636,6 +641,15 @@ class LoopAgent: def _close(self, item: WorkItem, loop: Loop, outcome: str, note: str) -> None: self._fact(item, f"work item {outcome}: {note}") + if loop.pr and outcome == "abandoned": + # L2: abandoning closes the item's open PR with a note. + try: + self._gitea.post_comment(self._repo, loop.pr, f"abandoned: {note}") + self._gitea.close_pull_request(self._repo, loop.pr) + except GiteaError as exc: + self._transport.post( + f"could not close PR #{loop.pr}: {exc}", thread=loop.thread + ) save(self._store, item.id, CLOSED, loop, cause=outcome) self._store.close_work_item(item.id, outcome=outcome, cause=note) if loop.issue: diff --git a/nisse-dev/src/nisse_dev/loop/milestones.py b/nisse-dev/src/nisse_dev/loop/milestones.py index fa2cb11..efb6cea 100644 --- a/nisse-dev/src/nisse_dev/loop/milestones.py +++ b/nisse-dev/src/nisse_dev/loop/milestones.py @@ -32,6 +32,7 @@ class Fixed: kind: Literal["pushed", "nothing", "new-pr", "closed", "failed"] note: str pr: PullRequest | None = None + branch: str = "" # a new PR's branch, which later fixes go to class Milestones: @@ -177,7 +178,7 @@ class Milestones: summary=f"PR #{pr} merged before this review fix landed.", issue=0, ) - return Fixed("new-pr", new.note, new.pr) + return Fixed("new-pr", new.note, new.pr, follow) if state.state != "open": return Fixed("closed", f"PR #{pr} was closed", state) return Fixed("pushed", f"pushed {sha[:8]}", state) diff --git a/nisse-dev/src/nisse_dev/views.py b/nisse-dev/src/nisse_dev/views.py index 2789774..bf13941 100644 --- a/nisse-dev/src/nisse_dev/views.py +++ b/nisse-dev/src/nisse_dev/views.py @@ -90,8 +90,9 @@ def project(events: list[StoredEvent], viewer: str) -> list[StoredEvent]: shown = _SHOWN.get(viewer) if shown is None: # A role with no view defined (the expert is stateless, the - # classifier sees only an action) reads the log as it is. - return list(events) + # classifier sees only an action) reads the log as it is, with + # results kept after their calls. + return _results_follow_calls(list(events)) results = { e.event.call_id: e.event for e in events if isinstance(e.event, ToolResult) } @@ -128,4 +129,30 @@ def project(events: list[StoredEvent], viewer: str) -> list[StoredEvent]: out.append(_said(stored, actor, line)) elif isinstance(event, AssistantText) and event.text.strip(): out.append(_said(stored, actor, event.text.strip())) - return out + return _results_follow_calls(out) + + +def _results_follow_calls(events: list[StoredEvent]) -> list[StoredEvent]: + """Keep each tool result right after its call, as both API formats + require: anything recorded while a call was in flight (a fact the + harness wrote while handling it, say) moves to just after the results. + A call with no result yet keeps its place; the renderer answers it.""" + out: list[StoredEvent] = [] + pending: set[str] = set() + held: list[StoredEvent] = [] + for stored in events: + event = stored.event + if isinstance(event, ToolCall): + pending.add(event.call_id) + out.append(stored) + elif isinstance(event, ToolResult) and event.call_id in pending: + pending.discard(event.call_id) + out.append(stored) + if not pending: + out.extend(held) + held = [] + elif pending: + held.append(stored) + else: + out.append(stored) + return out + held diff --git a/nisse-dev/tests/model_contract.py b/nisse-dev/tests/model_contract.py new file mode 100644 index 0000000..55d97ec --- /dev/null +++ b/nisse-dev/tests/model_contract.py @@ -0,0 +1,58 @@ +"""What a real model API insists on, for scripted test models. + +Both formats the agent sends require a tool call's result to come right +after it: in Anthropic's, the next message is the user's and starts with a +tool_result for every tool_use, before any text; in OpenAI's, the assistant +message with tool_calls is followed directly by a tool message for each. A +scripted model that doesn't check this lets a broken transcript pass the +unit tests and be rejected live; this makes it fail here. +""" + +from typing import Any + +from nisse_dev.render import OpenAIChatRenderer +from nisse_dev.render_anthropic import AnthropicMessagesRenderer + + +def assert_renderable(events: list[Any]) -> None: + _anthropic(AnthropicMessagesRenderer().render(list(events), None, "m")["messages"]) + _openai(OpenAIChatRenderer().render(list(events), None, "m")["messages"]) + + +def _anthropic(messages: list[dict[str, Any]]) -> None: + for index, message in enumerate(messages): + if message["role"] != "assistant": + continue + used = [b["id"] for b in message["content"] if b.get("type") == "tool_use"] + if not used: + continue + following = messages[index + 1] if index + 1 < len(messages) else None + assert following is not None and following["role"] == "user", ( + f"anthropic messages.{index}: tool_use with no user message after it" + ) + blocks = following["content"] + leading = [] + for block in blocks: + if block.get("type") != "tool_result": + break + leading.append(block["tool_use_id"]) + assert set(used) <= set(leading), ( + f"anthropic messages.{index}: tool_use ids {sorted(set(used) - set(leading))} " + "lack a tool_result at the start of the next message" + ) + + +def _openai(messages: list[dict[str, Any]]) -> None: + for index, message in enumerate(messages): + calls = [c["id"] for c in message.get("tool_calls") or []] + if not calls: + continue + answered = [] + for after in messages[index + 1 :]: + if after["role"] != "tool": + break + answered.append(after["tool_call_id"]) + assert set(calls) <= set(answered), ( + f"openai messages.{index}: tool_calls {sorted(set(calls) - set(answered))} " + "are not answered by the tool messages right after them" + ) diff --git a/nisse-dev/tests/test_loop.py b/nisse-dev/tests/test_loop.py index cbba820..5a88a13 100644 --- a/nisse-dev/tests/test_loop.py +++ b/nisse-dev/tests/test_loop.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import Any import pytest +from model_contract import assert_renderable from nisse_dev.cancel import Cancel from nisse_dev.canonical import AssistantText, CanonicalEvent, Fact, ToolCall @@ -34,6 +35,7 @@ class ScriptedModel: self.seen: list[list[Any]] = [] def complete(self, events, tools, model, system=None) -> list[CanonicalEvent]: + assert_renderable(events) # as the real API would self.offered.append({t["function"]["name"] for t in tools or []}) self.seen.append(events) assert self.turns, "the model was called more often than scripted" diff --git a/nisse-dev/tests/test_loop_milestones.py b/nisse-dev/tests/test_loop_milestones.py index 4fff477..9a77b22 100644 --- a/nisse-dev/tests/test_loop_milestones.py +++ b/nisse-dev/tests/test_loop_milestones.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any import pytest +from model_contract import assert_renderable from nisse_dev.cancel import Cancel from nisse_dev.canonical import ToolCall @@ -42,6 +43,7 @@ class ScriptedModel: self.turns = list(turns) def complete(self, events, tools, model, system=None) -> list[Any]: + assert_renderable(events) # as the real API would assert self.turns, "the model was called more often than scripted" return self.turns.pop(0) @@ -96,6 +98,9 @@ class Forge: def close_issue(self, repo: str, index: int) -> None: self.closed_issues.append(index) + def close_pull_request(self, repo: str, index: int) -> None: + self.prs[index]["state"] = "closed" + # what the operator does in Gitea def merge(self, n: int) -> None: head = self.prs[n]["head"] @@ -429,3 +434,45 @@ def test_boot_asks_when_a_publish_never_reached_gitea(world: dict[str, Any]) -> assert store.pending_intents() == [] assert loop_of(item(world)).wait == ANSWER assert "publishing was cut off by a restart" in transport.outbound[-1].text + + +def test_a_fix_after_a_follow_up_pr_goes_to_the_follow_up( + world: dict[str, Any], +) -> None: + # Found by the property walk (seed 7): after a fix became a follow-up PR, + # the next fix was pushed to the original, merged branch and lost. + forge: Forge = world["forge"] + model = ScriptedModel( + RESTATE, + WORK + FINISH, + [call("write_file", path="fix1.py", content="a = 1\n"), *FINISH], + [call("write_file", path="fix2.py", content="b = 2\n"), *FINISH], + ) + agent, transport = loop_agent(world, model) + agent.intake() + say(transport, world, "go") + [first] = forge.prs + forge.comments[first] = [Comment(1, "don", "add fix1", "t")] + agent.watch() + forge.merge(first) # merged before the fix lands: a follow-up PR + agent._schedule() + follow = loop_of(item(world)).pr + assert follow != first + forge.comments[follow] = [Comment(2, "don", "add fix2", "t")] + agent.watch() + agent._schedule() + assert len(forge.prs) == 2 # pushed to the follow-up, not a third PR + assert "fix2.py" in forge.files(follow) + + +def test_abandoning_an_item_closes_its_open_pr(world: dict[str, Any]) -> None: + # Found by the property walk (seed 4); the model's fix 1 (L2). + forge: Forge = world["forge"] + agent, transport = loop_agent(world, ScriptedModel(RESTATE, WORK + FINISH)) + agent.intake() + say(transport, world, "go") + [pr] = forge.prs + say(transport, world, "abandon") + assert item(world).outcome == "abandoned" + assert forge.prs[pr]["state"] == "closed" + assert (pr, "abandoned: abandoned by @don:test") in forge.posted diff --git a/nisse-dev/tests/test_loop_properties.py b/nisse-dev/tests/test_loop_properties.py new file mode 100644 index 0000000..bbcd35f --- /dev/null +++ b/nisse-dev/tests/test_loop_properties.py @@ -0,0 +1,249 @@ +"""specs/loop.qnt's properties, checked against the real loop (M2.1, PR D). + +The Quint model checks the design; this checks the code. A seeded random +walk interleaves what the operator and Gitea can do (go-aheads, answers, +comments, merges, closes, abandon, a crash and restart) with an adaptive +scripted model, and checks every property after every step. A failure prints +the seed and the steps, so it replays exactly. +""" + +import random +import subprocess +from pathlib import Path +from typing import Any + +import pytest +from model_contract import assert_renderable +from test_loop_milestones import DON, Forge, call +from test_loop_milestones import world as world # noqa: PLC0414 -- the fixture + +from nisse_dev.cancel import Cancel +from nisse_dev.canonical import Fact, ToolCall +from nisse_dev.gitea import Comment +from nisse_dev.loop.agent import LoopAgent +from nisse_dev.loop.driver import Driver +from nisse_dev.loop.items import ANSWER, GO_AHEAD, MERGE, WAITING, loop_of +from nisse_dev.oplog import OpLog +from nisse_dev.tools import Workspace +from nisse_dev.transport import InboundMessage, InMemoryTransport +from nisse_dev.worker import InlineWorker + +SEEDS = range(32) +STEPS = 20 + + +class PolicyModel: + """Plays a plausible driver: restate first, then a milestone, fixes + after comments, close after a merge; sometimes asks a question or + replies in prose. Seeded, so a walk replays exactly.""" + + def __init__(self, rng: random.Random) -> None: + self.rng = rng + self.n = 0 + + def complete(self, events, tools, model, system=None) -> list[Any]: + assert_renderable(events) # as the real API would + self.n += 1 + offered = {t["function"]["name"] for t in tools} + facts = [getattr(e.event, "text", "") for e in events] + roll = self.rng.random() + if roll < 0.08: + return [call("ask_human", question=f"question {self.n}?")] + if roll < 0.12: + return [ToolCall(call_id=f"p{self.n}", tool="nonsense", arguments={})] + if "finish_milestone" not in offered: + return [call("restate", text=f"restatement {self.n}")] + merged = any("merged" in t for t in facts[-6:]) + if merged and self.rng.random() < 0.8: + return [call("close_item", summary=f"done {self.n}")] + return [ + call("start_milestone", name=f"m{self.n}"), + call("write_file", path=f"f{self.n}.py", content=f"x = {self.n}\n"), + call("finish_milestone", summary=f"work {self.n}"), + ] + + +class Walk: + def __init__(self, world: dict[str, Any], seed: int) -> None: + self.world = world + self.rng = random.Random(seed) + self.seed = seed + self.steps: list[str] = [] + self.shown_at_go: list[int] = [] # the revision on screen at each go + self.agent, self.transport = self._agent() + + def _agent(self) -> tuple[LoopAgent, InMemoryTransport]: + transport = InMemoryTransport() + cancel = Cancel() + w = self.world + agent = LoopAgent( + w["store"], + transport, + w["forge"], + repo="don/sandbox", + project="sandbox", + account="nisse-dev", + operators={DON}, + gitea_operators={"don"}, + driver=Driver(PolicyModel(self.rng), "m", cancel=cancel), # type: ignore[arg-type] + workspace=Workspace(w["host"], w["repo"].root, timeout_s=30), + worker=InlineWorker(), + cancel=cancel, + oplog=OpLog("nisse-dev", _Null()), + milestones=w["milestones"], + ) + transport.on_message(agent.on_message) + return agent, transport + + def items(self) -> list[Any]: + return self.world["store"].work_items() + + def step(self) -> None: + forge: Forge = self.world["forge"] + open_items = [i for i in self.items() if i.status == "open"] + # Weighted towards what an item can actually do, so review paths + # are walked often, not drowned out by intake and crashes. + weights = {"intake": 2 if not open_items else 1, "crash": 1} + for item in open_items: + loop = loop_of(item) + weights.update({"go": 3, "answer": 2, "abandon": 1, "go-stale": 1}) + if loop.pr: + weights.update( + {"comment": 4, "merge": 2, "close-pr": 1, "comment-merge": 2} + ) + action = self.rng.choices(list(weights), list(weights.values()))[0] + self.steps.append(action) + item = open_items[0] if open_items else None + loop = loop_of(item) if item else None + if action == "intake": + self.agent.intake() + elif action == "crash": + # Volatile state is lost: a new agent over the same store. + self.agent, self.transport = self._agent() + self.agent.boot() + self.agent._schedule() + elif action in ("go", "answer", "abandon") and loop: + text = {"go": "go", "answer": "an answer", "abandon": "abandon"}[action] + if action == "go" and item.position == WAITING and loop.wait == GO_AHEAD: + self.shown_at_go.append(loop.rev) + self.transport.receive(DON, text, thread=loop.thread) + elif action == "go-stale" and loop and loop.rev > 1: + # A go sent while an older revision was on screen. + self.agent.handle( + InboundMessage(DON, "go", thread=loop.thread), about=loop.rev - 1 + ) + elif action == "comment" and loop: + forge.comments.setdefault(loop.pr, []).append( + Comment(len(self.steps) * 10, "don", f"comment {len(self.steps)}", "t") + ) + self.agent.watch() + self.agent._schedule() + elif ( + action == "comment-merge" and loop and forge.prs[loop.pr]["state"] == "open" + ): + # The race: a comment is seen, then the PR merges before the + # driver's fix lands. + forge.comments.setdefault(loop.pr, []).append( + Comment(len(self.steps) * 10, "don", f"comment {len(self.steps)}", "t") + ) + self.agent.watch() + forge.merge(loop.pr) + self.agent._schedule() + elif action == "merge" and loop and forge.prs[loop.pr]["state"] == "open": + forge.merge(loop.pr) + self.agent.watch() + self.agent._schedule() + elif action == "close-pr" and loop and forge.prs[loop.pr]["state"] == "open": + forge.prs[loop.pr]["state"] = "closed" + self.agent.watch() + self.agent._schedule() + + # -- the properties (specs/loop.qnt) ------------------------------------- + + def check(self) -> None: + where = f"seed {self.seed}, steps {self.steps}" + store, forge = self.world["store"], self.world["forge"] + items = self.items() + issues = [loop_of(i).issue for i in items if loop_of(i).issue] + assert len(issues) == len(set(issues)), f"one_item_per_issue: {where}" + for item in items: + loop = loop_of(item) + if item.status == "open" and item.position == WAITING: + explained = ( + (loop.wait == GO_AHEAD and loop.rev > 0) + or (loop.wait == ANSWER and bool(loop.note)) + or (loop.wait == MERGE and loop.pr > 0) + ) + assert explained, ( + f"waits_are_explainable: {item.position} {loop}: {where}" + ) + gos = [ + int(f.event.text.split("rev-")[1].split()[0]) + for f in store.transcript(item.id, "orchestrator").read() + if isinstance(f.event, Fact) + and f.event.text.startswith("go-ahead on rev-") + ] + assert all(rev in self.shown_at_go for rev in gos), ( + f"go_binds: {gos}: {where}" + ) + mine = [p for p in forge.prs.values() if p["head"].startswith(item.id)] + assert sum(p["state"] == "open" for p in mine) <= 1, ( + f"milestone_after_merge: {where}" + ) + if item.status == "closed": + assert not any(p["state"] == "open" for p in mine), ( + f"closed_items_leave_nothing (PR): {where}" + ) + assert not [ + i for i in store.pending_intents() if i.work_item == item.id + ], f"closed_items_leave_nothing (intent): {where}" + heads = [p["head"] for p in forge.prs.values()] + assert len(heads) == len(set(heads)), f"one_pr_per_milestone: {where}" + bare = self.world["bare"] + for n, pr in forge.prs.items(): + if not pr["merged"]: + continue + head = _rev(bare, f"refs/heads/{pr['head']}") + # A fix pushed after the merge must reach the operator as a PR + # (loop.qnt's lostFix): in main, or in a later PR, whatever they + # then decide about it. + landed = _is_ancestor(bare, head, "refs/heads/main") or any( + _is_ancestor(bare, head, f"refs/heads/{other['head']}") + for m, other in forge.prs.items() + if m > n + ) + assert landed, f"review_fixes_land (PR {n}): {where}" + + +class _Null: + def write(self, _: str) -> None: ... + + def flush(self) -> None: ... + + +def _rev(bare: Path, ref: str) -> str: + return subprocess.run( + ["git", "--git-dir", str(bare), "rev-parse", ref], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + + +def _is_ancestor(bare: Path, commit: str, ref: str) -> bool: + return ( + subprocess.run( + ["git", "--git-dir", str(bare), "merge-base", "--is-ancestor", commit, ref], + capture_output=True, + check=False, + ).returncode + == 0 + ) + + +@pytest.mark.parametrize("seed", SEEDS) +def test_the_loop_keeps_its_properties(world: dict[str, Any], seed: int) -> None: + walk = Walk(world, seed) + for _ in range(STEPS): + walk.step() + walk.check() diff --git a/nisse-dev/tests/test_views.py b/nisse-dev/tests/test_views.py index 05f4558..838ad1a 100644 --- a/nisse-dev/tests/test_views.py +++ b/nisse-dev/tests/test_views.py @@ -282,3 +282,18 @@ def test_operator_messages_carry_their_channel(tmp_path: Path) -> None: "[operator, comment on plan issue #9] hi", "[operator, Gitea comment] hi", ] + + +def test_something_written_mid_call_moves_after_the_result(tmp_path: Path) -> None: + # Live, M2.1: start_milestone's fact landed between the call and its + # result, and the model API rejected every later request in the run. + store = new_store(tmp_path) + driver = store.transcript("a", "driver") + driver.append(ToolCall(call_id="c1", tool="start_milestone", arguments={})) + store.transcript("a", "orchestrator").append(Fact(text="milestone 1 started")) + driver.append(ToolResult(call_id="c1", result="started", success=True)) + kinds = [type(e.event).__name__ for e in driver.view()] + assert kinds == ["ToolCall", "ToolResult", "UserMessage"] + body = AnthropicMessagesRenderer().render(driver.view(), tools=None, model="m") + result_first = body["messages"][1]["content"][0] + assert result_first["type"] == "tool_result" diff --git a/scripts/e2e.sh b/scripts/e2e.sh index 4c356ee..a4871f2 100755 --- a/scripts/e2e.sh +++ b/scripts/e2e.sh @@ -17,4 +17,6 @@ else export NISSE_E2E_BUILD_HOST=${NISSE_E2E_BUILD_HOST:-nisse-dev.garrett-hall.com} export NISSE_E2E_BUILD_KEY=${NISSE_E2E_BUILD_KEY:-/root/nisse-e2e/ssh/id_ed25519} fi -exec uv run pytest -m e2e -rs tests/integration/test_end_to_end.py "$@" +# Each run's state store is kept here, for nisse_dev.dump after a failure. +export NISSE_E2E_RUNS=${NISSE_E2E_RUNS:-/root/nisse-e2e/test-runs} +exec uv run pytest -m e2e -rs tests/integration/test_end_to_end.py tests/integration/test_end_to_end_loop.py "$@" diff --git a/specs/README.md b/specs/README.md index 28bd402..5b071e5 100644 --- a/specs/README.md +++ b/specs/README.md @@ -65,6 +65,13 @@ Traces need about 80 steps to reach second milestones. The `w_*` values at the end are reachability witnesses: each should be *violated*, which shows the state it names is reachable. +**Against the real code.** `nisse-dev/tests/test_loop_properties.py` checks +the same properties against the new loop itself: a seeded random walk over +what the operator and Gitea can do (go-aheads, answers, comments, merges, +closes, abandon, a crash and restart), checking every property after every +step. It found two bugs the design model couldn't, because they were in the +code: abandon left an open PR, and a review fix after a follow-up PR went to +the merged branch. Each also has its own regression test. + **Keeping it true.** When the architecture changes a rule this models, change -the model in the same PR. When the new loop is built (Development Plan Phase 2), -each property gets a test against the real code, as `orchestrator.qnt`'s did. +the model and the walk in the same PR. diff --git a/tests/integration/e2e_support.py b/tests/integration/e2e_support.py new file mode 100644 index 0000000..91f0752 --- /dev/null +++ b/tests/integration/e2e_support.py @@ -0,0 +1,42 @@ +"""Shared by the end-to-end tests: each run gets work that can't already be +done, and a base branch of its own, so runs never depend on one another or +on what earlier runs merged.""" + +import contextlib +import os +from collections.abc import Iterator + +from nisse_dev.gitea import GiteaClient, GiteaError + + +def task(run: str) -> str: + """A request no earlier run can have fulfilled.""" + return ( + f"Add a function vowels_{run}(text) to the sandbox package that returns " + "how many of a, e, i, o, u (either case) the text contains, with tests." + ) + + +def operator_gitea(url: str) -> GiteaClient: + token = os.environ.get("NISSE_OPERATOR_TOKEN", "") + assert token.strip(), ( + "NISSE_OPERATOR_TOKEN is not set: the test operator's Gitea token" + ) + return GiteaClient(url, token) + + +@contextlib.contextmanager +def base_branch(operator: GiteaClient, repo: str, run: str) -> Iterator[str]: + """A branch off main for this run's PRs to target and merge into, so + nothing accumulates in main; deleted afterwards.""" + name = f"e2e-base-{run}" + operator._request( + "POST", + f"/repos/{repo}/branches", + json={"new_branch_name": name, "old_branch_name": "main"}, + ) + try: + yield name + finally: + with contextlib.suppress(GiteaError): + operator._request("DELETE", f"/repos/{repo}/branches/{name}") diff --git a/tests/integration/nisse-test.yaml b/tests/integration/nisse-test.yaml index 7ad7fe3..03e515e 100644 --- a/tests/integration/nisse-test.yaml +++ b/tests/integration/nisse-test.yaml @@ -41,7 +41,9 @@ gitea: url: https://gitea.dongar.net token: ${NISSE_GITEA_TOKEN} repo: dongar/nisse-dev-testing-sandbox # the sandbox, never a real project - operators: [dongar] + # nisse-testing-operator plays the operator in the end-to-end tests + # (NISSE_OPERATOR_TOKEN); an operator here and nowhere else. + operators: [dongar, nisse-testing-operator] # What pushes go by (D4): the sandbox's deploy key, whose private half CI # writes to this path from the NISSE_PUSH_KEY secret (locally: copy it # there). An HTTPS push token works too: push_user plus push_token. diff --git a/tests/integration/test_end_to_end.py b/tests/integration/test_end_to_end.py index 12d6ba7..22abe53 100644 --- a/tests/integration/test_end_to_end.py +++ b/tests/integration/test_end_to_end.py @@ -25,6 +25,8 @@ from pathlib import Path from typing import Any import pytest +from e2e_support import base_branch, operator_gitea +from e2e_support import task as make_task from nisse_dev.agent import Agent, build_agent from nisse_dev.canonical import UserMessage @@ -43,10 +45,6 @@ pytestmark = [pytest.mark.integration, pytest.mark.e2e] OPERATOR = "@dongar:dongar.net" # an allowlisted operator in nisse-test.yaml SETTLE_S = 1200 # a milestone, with real models, inside one message -TASK = ( - "Add a function `count_vowels(text)` to the sandbox package that returns " - "how many of a, e, i, o, u (either case) the text contains, with tests." -) def build_host(ssh_host: SSHHost) -> BuildHost: @@ -70,10 +68,14 @@ def agent( ssh_host: SSHHost, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, -) -> Iterator[tuple[Agent, InMemoryTransport]]: +) -> Iterator[tuple[Agent, InMemoryTransport, str]]: run = uuid.uuid4().hex[:8] + stack = contextlib.ExitStack() + operator = operator_gitea(live_config.gitea_url) + base = stack.enter_context(base_branch(operator, live_config.gitea_repo, run)) config = replace( live_config, + base_branch=base, state_db=tmp_path / "state.db", workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"), build_host=build_host(ssh_host), @@ -84,7 +86,7 @@ def agent( try: agent = build_agent(config) agent.start() - yield agent, transport + yield agent, transport, run finally: if agent is not None: agent.stop() @@ -97,6 +99,7 @@ def agent( state_dir=tmp_path / "ssh", ) host.execute(f"rm -rf /tmp/nisse-e2e-{run}") + stack.close() # the run's base branch class Operator: @@ -148,14 +151,16 @@ def is_artifact(path: str) -> bool: ) -def test_a_milestone_end_to_end(agent: tuple[Agent, InMemoryTransport]) -> None: - running, transport = agent +def test_a_milestone_end_to_end( + agent: tuple[Agent, InMemoryTransport, str], +) -> None: + running, transport, run = agent operator = Operator(running, transport) gitea, repo = running.gitea, running.config.gitea_repo pr: dict[str, Any] = {} try: # The task first, before any project: it must be kept, not lost. - operator.say(TASK) + operator.say(make_task(run)) operator.say(f"use {running.config.project}") state = operator.drive_to_pr() pr = state.data["pr"] diff --git a/tests/integration/test_end_to_end_loop.py b/tests/integration/test_end_to_end_loop.py new file mode 100644 index 0000000..7524cf7 --- /dev/null +++ b/tests/integration/test_end_to_end_loop.py @@ -0,0 +1,220 @@ +"""The end-to-end test on the new loop (Development Plan M2.1, PR D): the +whole slice, from an issue assigned in Gitea to a merged PR and a closed work +item, 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 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, operator_gitea, task + +from nisse_dev.config import AgentConfig +from nisse_dev.execution import SSHHost +from nisse_dev.gitea import GiteaError +from nisse_dev.loop.build import RunningLoop, build_loop_agent +from nisse_dev.loop.items import ANSWER, GO_AHEAD, MERGE, WAITING, 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 + + +@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, + loop="new", + base_branch=base, + state_db=keep / "state.db", + workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"), + build_host=build_host(ssh_host), + ) + 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]") + ) + + +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) -> Any: + items = self.agent.worker.call(self.agent.store.work_items) + return items[0] if items else None + + def say(self, text: str) -> None: + self.transport.receive(OPERATOR, text, thread=loop_of(self.item()).thread) + self.settle() + + def drive_until(self, wait: str) -> Any: + """Answer whatever the agent asks until it waits for ``wait``.""" + deadline = time.monotonic() + DEADLINE_S + answers = 0 + while time.monotonic() < deadline: + item = self.item() + 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 item.position == WAITING and loop.wait == GO_AHEAD: + self.say("go") + 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.") + else: + self.settle() + raise AssertionError(f"never waited for {wait}: {self.last()}") + + 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 --- + 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"}, + ) + as_operator._request( + "POST", f"/repos/{repo}/pulls/{pr}/merge", json={"Do": "merge"} + ) + operator.tick() + deadline = time.monotonic() + DEADLINE_S + nudges = 0 + while operator.item().status != "closed" and time.monotonic() < deadline: + if loop_of(operator.item()).wait == ANSWER: + nudges += 1 + assert nudges <= MAX_ANSWERS, f"never closed: {operator.last()}" + operator.say("It's merged; please close the item.") + operator.settle() + assert operator.item().outcome == "completed" + state = gitea._request("GET", f"/repos/{repo}/issues/{issue}")["state"] + assert state == "closed" + finally: + # Leave the sandbox tidy: whatever is still open gets closed. + item = operator.item() + if item is not None and item.status == "open": + operator.say("abandon") + if issue is not None: + with contextlib.suppress(GiteaError): + as_operator.close_issue(repo, issue) + if pr: + with contextlib.suppress(GiteaError): + head = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["ref"] + as_operator._request("DELETE", f"/repos/{repo}/branches/{head}")