nisse/nisse-dev/tests/test_quint_counterexamples.py

"""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 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")


@XFAIL
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")


@XFAIL
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")


@XFAIL
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")


@XFAIL
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") == 1


@XFAIL
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
    agent.say("continue", "start milestone 1")
    # There is a PR for this milestone: the agent must end up watching it,
    # not trying to open another.
    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)