"""Plan externalisation to Gitea and approvals arriving on either channel (spec A4, Part H).""" 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", delegate="y", consultant="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 (A4) # The store keeps a pointer, not the plan record itself (Part H). 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 (A4). 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"