nisse/nisse-dev/tests/test_quint_counterexamples.py

"""The counterexamples found by specs/orchestrator.qnt, replayed against the
real orchestrator. Each names the scenario in specs/orchestrator_scenarios.qnt
it reproduces; all seven are fixed, and the tests stay as regressions.

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)


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


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 the
    # operator is told, and it is delivered and answered again.
    assert "answering it again" in agent.transport.outbound[-1].text
    [delivered] = agent.pollers[20].poll_once(20)
    assert delivered.body == "please add a test"
    assert "pushed" in agent.transport.outbound[-1].text
    assert "reviewing" not in agent.store.get_state().data["pr"]


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


class Crash(BaseException):
    """The process dying: no handler runs."""


def crash_mid_review(agent: Agent, gitea: FakeGitea) -> None:
    def dies(milestone: int) -> ExecutorOutcome:
        raise Crash()

    agent.executor.run = dies  # type: ignore[method-assign]
    with pytest.raises(Crash):
        agent.pollers[20].poll_once(20)


def test_a_second_crash_while_answering_again_still_redelivers(tmp_path: Path) -> None:
    gitea = FakeGitea()
    agent = Agent(tmp_path, gitea)
    agent.say("use demo", "approve", "approve", "start milestone 1")
    gitea.add(20, "don", "please add a test")
    crash_mid_review(agent, gitea)
    agent = Agent(tmp_path, gitea)
    crash_mid_review(agent, gitea)  # dies again while answering it again
    agent = Agent(tmp_path, gitea)
    [delivered] = agent.pollers[20].poll_once(20)
    assert delivered.body == "please add a test"


def test_an_executor_error_answers_the_comment_so_it_is_not_redelivered(
    tmp_path: Path,
) -> None:
    gitea = FakeGitea()
    agent = Agent(tmp_path, gitea)
    agent.say("use demo", "approve", "approve", "start milestone 1")
    gitea.add(20, "don", "please add a test")

    def fails(milestone: int) -> ExecutorOutcome:
        raise RuntimeError("bug in the executor")

    agent.executor.run = fails  # type: ignore[method-assign]
    agent.pollers[20].poll_once(20)
    assert "could not handle a review comment" in agent.transport.outbound[-1].text
    agent = Agent(tmp_path, gitea)  # restart: nothing to answer again
    assert "still watching" in agent.transport.outbound[-1].text
    assert agent.pollers[20].poll_once(20) == []


def test_project_complete_still_completes_when_gitea_is_down(tmp_path: Path) -> None:
    agent = Agent(tmp_path, FakeGitea())

    def down(issue: int, note: str) -> None:
        raise GiteaError("gitea is down")

    agent.orch._board.close = down  # type: ignore[attr-defined]
    agent.say("use demo", "approve", "approve", "start milestone 1", "project complete")
    assert agent.state == IDLE
    assert "could not close plan issue #10" in agent.transport.outbound[-1].text