dongarandClaude Opus 5.5
8bd1ac0fa2
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m12s
Fixes counterexample 5 from specs/orchestrator.qnt. A crash after the PR was opened but before PR_OPEN was recorded left the agent in EXECUTING; the restart question's 'continue' and 'start' then published again, Gitea refused a second PR, and there was no way back to PR_OPEN. On the live run the executor model happened to notice the existing work; recovery must not depend on that. - Boot (A3 step 4): in EXECUTING, if Gitea shows the milestone's PR open, adopt it, watch it from the start, and resume in PR_OPEN with a notice. Gitea unreachable or the PR closed: the restart question, as before. - MilestonePublisher.publish is idempotent: after pushing, an open PR for the branch is adopted (and so updated) rather than a second one opened. This also covers a PR opened whose response was lost. - The model adopts too, gains the lost-response outcome, and now counts PRs opened per milestone (one_pr_per_milestone was a flag the fixed model could never set). A mutant without adoption fails the new scenario; the properties held over 200,000 traces each. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
350 lines
12 KiB
Python
350 lines
12 KiB
Python
"""The counterexamples found by specs/orchestrator.qnt, replayed against the
|
|
real orchestrator. Each is marked xfail until its fix lands, and names the
|
|
scenario in specs/orchestrator_scenarios.qnt it reproduces.
|
|
|
|
The model is a hand-written reading of control.py; reproducing each trace
|
|
here is what shows the flaw is in the design, not in the model.
|
|
"""
|
|
|
|
import io
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from nisse_dev.control import (
|
|
AWAIT_INPUT,
|
|
AWAIT_MILESTONE_SIGNOFF,
|
|
EXECUTING,
|
|
IDLE,
|
|
MILESTONE_PLANNING,
|
|
PR_OPEN,
|
|
Orchestrator,
|
|
StubAgent,
|
|
)
|
|
from nisse_dev.executor import ExecutorOutcome
|
|
from nisse_dev.gitea import Comment, GiteaError, PullRequest
|
|
from nisse_dev.oplog import OpLog
|
|
from nisse_dev.poller import CommentPoller
|
|
from nisse_dev.publish import PublishResult
|
|
from nisse_dev.state import StateStore
|
|
from nisse_dev.transport import InMemoryTransport
|
|
from nisse_dev.worker import SerialWorker
|
|
|
|
AGENT = "nisse-bot"
|
|
|
|
|
|
class QueueingWorker(SerialWorker):
|
|
"""Holds submitted jobs until released, so a test can queue two
|
|
messages before either runs, as a real worker does under load."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.held: list[Any] = []
|
|
self.holding = True
|
|
|
|
def submit(self, job: Any) -> None:
|
|
if self.holding:
|
|
self.held.append(job)
|
|
else:
|
|
job()
|
|
|
|
def call(self, job: Any, timeout: float | None = None) -> Any:
|
|
return job()
|
|
|
|
@property
|
|
def busy(self) -> bool:
|
|
return False
|
|
|
|
def release(self) -> None:
|
|
self.holding = False
|
|
jobs, self.held = self.held, []
|
|
for job in jobs:
|
|
job()
|
|
|
|
|
|
class FakeGitea:
|
|
def __init__(self) -> None:
|
|
self.threads: dict[int, list[Comment]] = {}
|
|
self.prs: set[str] = set()
|
|
self._next = 100
|
|
|
|
def add(self, index: int, author: str, body: str) -> int:
|
|
self._next += 1
|
|
self.threads.setdefault(index, []).append(
|
|
Comment(id=self._next, author=author, body=body, created_at="t")
|
|
)
|
|
return self._next
|
|
|
|
def list_comments(self, repo: str, index: int, since: str | None = None) -> list:
|
|
return list(self.threads.get(index, []))
|
|
|
|
|
|
class Board:
|
|
def __init__(self, gitea: FakeGitea) -> None:
|
|
self._gitea = gitea
|
|
|
|
def open_project_issue(self, project: str, task_id: str) -> int:
|
|
return 10
|
|
|
|
def post_plan(self, issue: int, gate: str, rev: int, plan: str) -> Any:
|
|
class Posted:
|
|
comment_id = self._gitea.add(issue, AGENT, f"{gate} rev-{rev}")
|
|
|
|
return Posted()
|
|
|
|
def echo(self, issue: int, text: str) -> None:
|
|
pass
|
|
|
|
def close(self, issue: int, note: str) -> None:
|
|
pass
|
|
|
|
|
|
class Executor:
|
|
def __init__(self) -> None:
|
|
self.outcome = ExecutorOutcome("completed", "done")
|
|
|
|
def run(self, milestone: int) -> ExecutorOutcome:
|
|
return self.outcome
|
|
|
|
|
|
class Publisher:
|
|
"""Opens one PR per branch, as Gitea does; a second attempt fails."""
|
|
|
|
def __init__(self, gitea: FakeGitea, store: StateStore) -> None:
|
|
self._gitea = gitea
|
|
self._store = store
|
|
self.crash_after_opening = False
|
|
self.attempts = 0
|
|
|
|
def reconcilers(self) -> dict[str, Any]:
|
|
return {"open_pr": lambda intent: intent.payload["branch"] in self._gitea.prs}
|
|
|
|
def publish(self, **kwargs: Any) -> PublishResult:
|
|
branch = f"{kwargs['task_id']}/milestone-{kwargs['milestone']}"
|
|
self.attempts += 1
|
|
intent = self._store.begin_intent("open_pr", {"branch": branch}, cause="t")
|
|
if branch in self._gitea.prs:
|
|
raise GiteaError("pull request already exists for these targets")
|
|
self._gitea.prs.add(branch)
|
|
if self.crash_after_opening:
|
|
raise SystemExit("crash") # the process dies here
|
|
self._store.commit_intent(intent, cause="opened")
|
|
return PublishResult(
|
|
PullRequest(number=20, state="open", head=branch, url="u"), "opened PR #20"
|
|
)
|
|
|
|
def published_pull_request(self, task_id: str, milestone: int) -> Any:
|
|
branch = f"{task_id}/milestone-{milestone}"
|
|
if branch in self._gitea.prs:
|
|
return PullRequest(number=20, state="open", head=branch, url="u")
|
|
return None
|
|
|
|
def push_followup(self, **kwargs: Any) -> str | None:
|
|
return "abc12345"
|
|
|
|
def comment(self, index: int, body: str) -> None:
|
|
pass
|
|
|
|
|
|
class Agent:
|
|
def __init__(self, tmp_path: Path, gitea: FakeGitea, *, worker: Any = None) -> None:
|
|
self.store = StateStore(
|
|
tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())
|
|
)
|
|
self.transport = InMemoryTransport()
|
|
self.gitea = gitea
|
|
self.executor = Executor()
|
|
self.publisher = Publisher(gitea, self.store)
|
|
self.pollers: dict[int, CommentPoller] = {}
|
|
|
|
def factory(handler: Any) -> CommentPoller:
|
|
poller = CommentPoller(gitea, "o/r", handler, ignore_authors=[AGENT])
|
|
|
|
def start(index: int) -> None:
|
|
self.pollers[index] = poller
|
|
|
|
poller.start = start # type: ignore[method-assign]
|
|
poller.stop = lambda: None # type: ignore[method-assign]
|
|
return poller
|
|
|
|
self.orch = Orchestrator(
|
|
self.store,
|
|
self.transport,
|
|
StubAgent(),
|
|
operators={"don"},
|
|
gitea_operators={"don"},
|
|
board=Board(gitea),
|
|
executor=self.executor,
|
|
publisher=self.publisher,
|
|
poller_factory=factory,
|
|
**({"worker": worker} if worker else {}),
|
|
)
|
|
self.orch.start()
|
|
|
|
@property
|
|
def state(self) -> str:
|
|
return self.store.get_state().state
|
|
|
|
def gate(self) -> dict:
|
|
return self.store.get_state().data.get("gate") or {}
|
|
|
|
def say(self, *texts: str) -> None:
|
|
for text in texts:
|
|
self.transport.receive("don", text)
|
|
|
|
|
|
XFAIL = pytest.mark.xfail(strict=True, reason="counterexample; fix pending")
|
|
|
|
|
|
def test_double_approve_approves_an_unseen_plan(tmp_path: Path) -> None:
|
|
"""doubleApproveApprovesAnUnseenPlanTest"""
|
|
worker = QueueingWorker()
|
|
agent = Agent(tmp_path, FakeGitea(), worker=worker)
|
|
worker.holding = False
|
|
agent.say("use demo") # arch rev-1 is up
|
|
worker.holding = True
|
|
agent.say("approve", "approve") # both queued, looking at arch rev-1
|
|
worker.release()
|
|
# The second approval must not approve the milestone plan, never seen.
|
|
assert not agent.gate().get("approved")
|
|
|
|
|
|
def test_named_revision_approves_the_wrong_gate(tmp_path: Path) -> None:
|
|
"""namedRevisionApprovesTheWrongGateTest"""
|
|
worker = QueueingWorker()
|
|
agent = Agent(tmp_path, FakeGitea(), worker=worker)
|
|
worker.holding = False
|
|
agent.say("use demo")
|
|
worker.holding = True
|
|
agent.say("approve", "approve rev-1") # "rev-1" meant the architecture
|
|
worker.release()
|
|
assert not agent.gate().get("approved")
|
|
|
|
|
|
def test_approval_on_both_channels_approves_the_next_plan(tmp_path: Path) -> None:
|
|
"""approvalOnBothChannelsApprovesTheNextPlanTest"""
|
|
gitea = FakeGitea()
|
|
worker = QueueingWorker()
|
|
agent = Agent(tmp_path, gitea, worker=worker)
|
|
worker.holding = False
|
|
agent.say("use demo")
|
|
worker.holding = True
|
|
agent.say("approve") # Matrix: queued
|
|
gitea.add(10, "don", "approve") # and on the issue, about the same plan
|
|
# The poller delivers it; it waits behind the Matrix approval.
|
|
delivered = agent.pollers[10]
|
|
worker.held.append(lambda: delivered.poll_once(10))
|
|
worker.release()
|
|
assert agent.state == AWAIT_MILESTONE_SIGNOFF
|
|
assert not agent.gate().get("approved")
|
|
|
|
|
|
def test_an_afterthought_rejects_the_next_plan(tmp_path: Path) -> None:
|
|
"""anAfterthoughtRejectsTheNextPlanTest"""
|
|
worker = QueueingWorker()
|
|
agent = Agent(tmp_path, FakeGitea(), worker=worker)
|
|
worker.holding = False
|
|
agent.say("use demo")
|
|
worker.holding = True
|
|
agent.say("approve", "also, keep it in one file")
|
|
worker.release()
|
|
# The note was about the architecture; the milestone plan was not
|
|
# rejected by it.
|
|
assert agent.state == AWAIT_MILESTONE_SIGNOFF
|
|
assert agent.gate().get("rev") == 2 # arch was rev-1; not re-planned
|
|
|
|
|
|
def test_crash_after_opening_the_pr_publishes_again(tmp_path: Path) -> None:
|
|
"""crashAfterOpeningThePrPublishesAgainTest"""
|
|
gitea = FakeGitea()
|
|
agent = Agent(tmp_path, gitea)
|
|
agent.say("use demo", "approve", "approve")
|
|
agent.publisher.crash_after_opening = True
|
|
with pytest.raises(SystemExit):
|
|
agent.say("start milestone 1")
|
|
assert agent.state == EXECUTING # the PR exists; PR_OPEN was never recorded
|
|
|
|
agent = Agent(tmp_path, gitea) # restart: the intent is verified
|
|
# There is a PR for this milestone: the agent adopts it on boot and ends
|
|
# up watching it, never trying to open another.
|
|
assert agent.state == PR_OPEN
|
|
assert "PR #20 is open" in agent.transport.outbound[-1].text
|
|
agent.say("continue", "start milestone 1") # no longer means anything
|
|
assert agent.publisher.attempts == 0
|
|
assert agent.state == PR_OPEN
|
|
|
|
|
|
@XFAIL
|
|
def test_crash_during_a_review_loses_the_comment(tmp_path: Path) -> None:
|
|
"""crashDuringAReviewLosesTheCommentTest"""
|
|
gitea = FakeGitea()
|
|
agent = Agent(tmp_path, gitea)
|
|
agent.say("use demo", "approve", "approve", "start milestone 1")
|
|
assert agent.state == PR_OPEN
|
|
gitea.add(20, "don", "please add a test")
|
|
|
|
class Crash(BaseException):
|
|
pass
|
|
|
|
def dies(milestone: int) -> ExecutorOutcome:
|
|
raise Crash() # the process dies mid-review
|
|
|
|
agent.executor.run = dies # type: ignore[method-assign]
|
|
with pytest.raises(Crash):
|
|
agent.pollers[20].poll_once(20)
|
|
|
|
agent = Agent(tmp_path, gitea) # restart
|
|
assert agent.state == PR_OPEN
|
|
# The comment was taken but never answered: after the restart it must
|
|
# be handled again, or the operator told it was dropped.
|
|
delivered = agent.pollers[20].poll_once(20)
|
|
told = any(
|
|
"please add a test" in m.text or "review" in m.text
|
|
for m in agent.transport.outbound[1:]
|
|
)
|
|
assert delivered or told
|
|
|
|
|
|
@XFAIL
|
|
def test_project_complete_leaves_the_plan_issue_open(tmp_path: Path) -> None:
|
|
"""projectCompleteLeavesThePlanIssueOpenTest"""
|
|
closed: list[int] = []
|
|
gitea = FakeGitea()
|
|
agent = Agent(tmp_path, gitea)
|
|
agent.orch._board.close = lambda issue, note: closed.append(issue) # type: ignore[attr-defined]
|
|
agent.say("use demo", "approve", "approve", "start milestone 1", "project complete")
|
|
assert agent.state == IDLE
|
|
assert closed == [10]
|
|
|
|
|
|
def test_the_harness_reaches_pr_open_without_faults(tmp_path: Path) -> None:
|
|
"""Guard for the harness itself: with nothing queued or crashed, the
|
|
same flow is correct."""
|
|
agent = Agent(tmp_path, FakeGitea())
|
|
agent.say("use demo", "approve")
|
|
assert agent.state == AWAIT_MILESTONE_SIGNOFF
|
|
agent.say("approve", "start milestone 1")
|
|
assert agent.state == PR_OPEN
|
|
assert agent.publisher.attempts == 1
|
|
_ = (AWAIT_INPUT, MILESTONE_PLANNING)
|
|
|
|
|
|
def test_boot_asks_as_before_when_gitea_cannot_say(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
gitea = FakeGitea()
|
|
agent = Agent(tmp_path, gitea)
|
|
agent.say("use demo", "approve", "approve")
|
|
agent.publisher.crash_after_opening = True
|
|
with pytest.raises(SystemExit):
|
|
agent.say("start milestone 1")
|
|
|
|
def unreachable(self: Publisher, task_id: str, milestone: int) -> Any:
|
|
raise GiteaError("gitea is down")
|
|
|
|
monkeypatch.setattr(Publisher, "published_pull_request", unreachable)
|
|
agent = Agent(tmp_path, gitea)
|
|
assert agent.state == AWAIT_INPUT
|
|
assert "restarted mid-EXECUTING" in agent.transport.outbound[-1].text
|