nisse/nisse-dev/tests/test_approvals.py

"""Plan externalisation to Gitea and approvals arriving on either channel
(architecture L3, D2)."""

import io
from pathlib import Path
from typing import Any

from nisse_dev.approvals import PlanBoard
from nisse_dev.canonical import CanonicalEvent, ToolCall
from nisse_dev.control import (
    AWAIT_ARCH_SIGNOFF,
    AWAIT_MILESTONE_SIGNOFF,
    MILESTONE_PLANNING,
    Orchestrator,
)
from nisse_dev.gitea import Comment, GiteaError
from nisse_dev.llm import RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.planner import LLMPlanningAgent
from nisse_dev.state import StateStore
from nisse_dev.transport import InMemoryTransport

ROLES = RoleMap(executor="x", planner="y", helper="y", expert="y")


class FakeGitea:
    def __init__(self, *, fail: bool = False) -> None:
        self.issues: list[dict[str, str]] = []
        self.comments: list[tuple[int, str]] = []
        self._fail = fail
        self._next_id = 100

    def create_issue(self, repo: str, title: str, body: str) -> int:
        if self._fail:
            raise GiteaError("gitea is down")
        self.issues.append({"title": title, "body": body})
        return len(self.issues)

    def post_comment(self, repo: str, index: int, body: str) -> Comment:
        if self._fail:
            raise GiteaError("gitea is down")
        self.comments.append((index, body))
        self._next_id += 1
        return Comment(
            id=self._next_id, author="nisse-dev-testing", body=body, created_at="t"
        )

    def list_comments(self, repo: str, index: int, since: str | None = None):
        return []


class ScriptedClient:
    def __init__(self) -> None:
        self.calls = 0

    def complete(self, events, tools, model, system=None) -> list[CanonicalEvent]:
        self.calls += 1
        return [
            ToolCall(
                call_id=f"c{self.calls}",
                tool="submit_plan",
                arguments={"type": "x", "content": f"plan {self.calls}"},
            )
        ]


def build(
    tmp_path: Path, gitea: Any
) -> tuple[Orchestrator, StateStore, InMemoryTransport, PlanBoard]:
    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
    transport = InMemoryTransport()
    board = PlanBoard(gitea, "o/sandbox")
    orch = Orchestrator(
        store,
        transport,
        LLMPlanningAgent(ScriptedClient(), store, ROLES),
        operators={"don"},
        roles=ROLES,
        gitea_operators={"dongar"},
        board=board,
    )
    orch.start()
    return orch, store, transport, board


def comment(body: str, author: str = "dongar", cid: int = 900) -> Comment:
    return Comment(id=cid, author=author, body=body, created_at="t")


def test_project_start_opens_an_issue_and_posts_the_plan(tmp_path: Path) -> None:
    gitea = FakeGitea()
    _, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")

    assert gitea.issues[0]["title"] == "nisse-dev: demo"
    issue, body = gitea.comments[0]
    assert issue == 1
    assert "plan 1" in body
    assert "approve rev-1" in body  # the tag is printed for copy-paste (L3)
    # The store keeps a pointer, not the plan record itself (D2).
    assert store.get_state().data["board"]["issue"] == 1


def test_approval_by_gitea_comment_satisfies_the_gate(tmp_path: Path) -> None:
    gitea = FakeGitea()
    orch, store, _, _ = build(tmp_path, gitea)
    orch._transport.receive("don", "use demo")
    assert store.get_state().state == AWAIT_ARCH_SIGNOFF

    orch.handle_plan_comment(comment("approve rev-1"))

    assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF


def test_stale_revision_from_gitea_is_refused(tmp_path: Path) -> None:
    gitea = FakeGitea()
    orch, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")
    transport.receive("don", "too vague")  # now at rev-2

    orch.handle_plan_comment(comment("approve rev-1"))
    assert store.get_state().state == AWAIT_ARCH_SIGNOFF  # not advanced

    orch.handle_plan_comment(comment("approve rev-2", cid=901))
    assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF


def test_non_operator_gitea_comment_cannot_approve(tmp_path: Path) -> None:
    gitea = FakeGitea()
    orch, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")

    orch.handle_plan_comment(comment("approve rev-1", author="drive-by"))
    assert store.get_state().state == AWAIT_ARCH_SIGNOFF


def test_agents_own_comment_cannot_approve_its_plan(tmp_path: Path) -> None:
    # The agent posts plans as nisse-dev-testing; that login is not on the
    # operator allowlist, so its own comments can never satisfy its gate (L3).
    gitea = FakeGitea()
    orch, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")

    orch.handle_plan_comment(comment("approve rev-1", author="nisse-dev-testing"))
    assert store.get_state().state == AWAIT_ARCH_SIGNOFF


def test_rejection_by_gitea_comment_replans(tmp_path: Path) -> None:
    gitea = FakeGitea()
    orch, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")

    orch.handle_plan_comment(comment("needs more detail on the schema"))

    assert store.get_state().state == AWAIT_ARCH_SIGNOFF
    assert "plan 2" in transport.outbound[-1].text  # replanned


def test_matrix_approval_is_echoed_to_gitea(tmp_path: Path) -> None:
    gitea = FakeGitea()
    _, _, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")
    transport.receive("don", "approve")

    echoes = [body for _, body in gitea.comments if "approved by don" in body]
    assert echoes, "the Gitea record must not silently disagree with Matrix"


def test_gitea_approval_is_echoed_to_matrix(tmp_path: Path) -> None:
    gitea = FakeGitea()
    orch, _, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")
    orch.handle_plan_comment(comment("approve rev-1"))

    assert any("approved by dongar via gitea" in m.text for m in transport.outbound)


def test_gitea_outage_leaves_matrix_a_complete_channel(tmp_path: Path) -> None:
    gitea = FakeGitea(fail=True)
    _, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")

    # Planning proceeded and the gate is usable despite Gitea being down.
    assert store.get_state().state == AWAIT_ARCH_SIGNOFF
    assert any("Matrix-only" in m.text for m in transport.outbound)
    transport.receive("don", "approve")
    assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF


def test_comments_are_ignored_outside_a_gate(tmp_path: Path) -> None:
    gitea = FakeGitea()
    orch, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")
    transport.receive("don", "approve")  # arch approved; now at milestone gate
    transport.receive("don", "approve")  # milestone approved, awaiting start
    before = store.get_state().state

    orch.handle_plan_comment(comment("approve rev-1", cid=902))
    assert store.get_state().state == before


def test_board_records_the_milestone_plan_too(tmp_path: Path) -> None:
    gitea = FakeGitea()
    _, store, transport, _ = build(tmp_path, gitea)
    transport.receive("don", "use demo")
    transport.receive("don", "approve")

    assert store.get_state().state in (MILESTONE_PLANNING, AWAIT_MILESTONE_SIGNOFF)
    posted = [body for _, body in gitea.comments if "milestone-1" in body]
    assert posted, "the milestone plan must be externalised as well"