nisse/nisse-dev/tests/test_pr_open.py

"""The B1 loop past EXECUTING: publish, review gate, then next milestone or
project completion."""

import io
from pathlib import Path
from typing import Any

import pytest

from nisse_dev.canonical import AssistantText, CanonicalEvent, StoredEvent, ToolCall
from nisse_dev.control import (
    AWAIT_INPUT,
    EXECUTING,
    IDLE,
    PR_OPEN,
    Orchestrator,
)
from nisse_dev.executor import ExecutorOutcome
from nisse_dev.gitea import GiteaError, PullRequest
from nisse_dev.llm import RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.planner import LLMPlanningAgent
from nisse_dev.publish import PublishResult
from nisse_dev.state import StateStore
from nisse_dev.transport import InMemoryTransport

ROLES = RoleMap(
    executor="llama3.2",
    planner="haiku",
    delegate="haiku",
    consultant="haiku",
)


class ScriptedClient:
    def __init__(self, plans: list[str]) -> None:
        self._plans = plans
        self.calls: list[dict[str, Any]] = []

    def complete(
        self,
        events: list[StoredEvent],
        tools: list[dict[str, Any]] | None,
        model: str,
        system: str | None = None,
    ) -> list[CanonicalEvent]:
        self.calls.append({"system": system})
        if not self._plans:
            return [AssistantText(text="no more plans", model=model)]
        return [
            ToolCall(
                call_id=f"c{len(self.calls)}",
                tool="submit_plan",
                arguments={"type": "x", "content": self._plans.pop(0)},
            )
        ]


class DoneExecutor:
    def run(self, milestone: int) -> ExecutorOutcome:
        return ExecutorOutcome("completed", f"milestone {milestone} done")


class FakePublisher:
    def __init__(
        self, *, result: PublishResult | None = None, error: Exception | None = None
    ) -> None:
        self._result = result or PublishResult(
            PullRequest(number=7, state="open", head="h", url="https://g/pulls/7"),
            "opened PR #7: https://g/pulls/7",
        )
        self._error = error
        self.calls: list[dict[str, Any]] = []

    def reconcilers(self) -> dict[str, Any]:
        return {"open_pr": lambda intent: True}

    def publish(self, **kwargs: Any) -> PublishResult:
        self.calls.append(kwargs)
        if self._error is not None:
            raise self._error
        return self._result


def build(
    tmp_path: Path, publisher: Any, plans: list[str] | None = None
) -> tuple[StateStore, InMemoryTransport]:
    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
    transport = InMemoryTransport()
    client = ScriptedClient(plans or ["arch plan", "m1 plan", "m2 plan"])
    orch = Orchestrator(
        store,
        transport,
        LLMPlanningAgent(client, store, ROLES),
        operators={"don"},
        roles=ROLES,
        executor=DoneExecutor(),
        publisher=publisher,
    )
    orch.start()
    return store, transport


def run_to_pr(tmp_path: Path, publisher: Any) -> tuple[StateStore, InMemoryTransport]:
    store, transport = build(tmp_path, publisher)
    transport.receive("don", "use demo")
    transport.receive("don", "approve")  # arch
    transport.receive("don", "approve")  # milestone plan
    transport.receive("don", "start milestone 1")
    return store, transport


def test_completed_milestone_publishes_and_enters_pr_open(tmp_path: Path) -> None:
    publisher = FakePublisher()
    store, transport = run_to_pr(tmp_path, publisher)

    assert store.get_state().state == PR_OPEN
    assert publisher.calls[0]["milestone"] == 1
    assert publisher.calls[0]["project"] == "demo"
    assert "m1 plan" in publisher.calls[0]["plan"]  # the approved plan travels
    assert "PR #7" in transport.outbound[-1].text


def test_pr_open_records_the_pull_request(tmp_path: Path) -> None:
    store, _ = run_to_pr(tmp_path, FakePublisher())
    assert store.get_state().data["pr"]["number"] == 7


def test_next_milestone_replans_and_bumps_the_counter(tmp_path: Path) -> None:
    store, transport = run_to_pr(tmp_path, FakePublisher())
    transport.receive("don", "next milestone")

    state = store.get_state()
    assert state.state != PR_OPEN
    assert state.data["milestone"] == 2
    assert "m2 plan" in transport.outbound[-1].text


def test_project_complete_returns_to_idle(tmp_path: Path) -> None:
    store, transport = run_to_pr(tmp_path, FakePublisher())
    transport.receive("don", "project complete")

    state = store.get_state()
    assert state.state == IDLE
    assert state.project is None
    assert "project closed" in transport.outbound[-1].text


def test_unknown_command_in_pr_open_explains_the_options(tmp_path: Path) -> None:
    store, transport = run_to_pr(tmp_path, FakePublisher())
    transport.receive("don", "looks good!")
    assert store.get_state().state == PR_OPEN
    assert "next milestone" in transport.outbound[-1].text


def test_no_changes_reports_without_entering_pr_open(tmp_path: Path) -> None:
    publisher = FakePublisher(result=PublishResult(None, "no changes to publish"))
    store, transport = run_to_pr(tmp_path, publisher)
    assert store.get_state().state == EXECUTING
    assert transport.outbound[-1].text == "no changes to publish"


def test_publish_failure_gates_to_the_operator(tmp_path: Path) -> None:
    publisher = FakePublisher(error=GiteaError("gitea is down"))
    store, transport = run_to_pr(tmp_path, publisher)
    assert store.get_state().state == AWAIT_INPUT
    assert "could not publish" in transport.outbound[-1].text
    # And the operator can still get back out (the deadlock regression).
    transport.receive("don", "gitea is back")
    assert store.get_state().state != AWAIT_INPUT


def test_publisher_reconciler_is_registered_at_boot(tmp_path: Path) -> None:
    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
    store.begin_intent("open_pr", {"branch": "b"}, cause="crash")
    transport = InMemoryTransport()
    orch = Orchestrator(
        store,
        transport,
        LLMPlanningAgent(ScriptedClient([]), store, ROLES),
        operators={"don"},
        roles=ROLES,
        publisher=FakePublisher(),
    )
    orch.start()
    # The publisher's reconciler answered "already opened", so boot resumed
    # instead of gating on an unknown intent kind.
    assert store.pending_intents() == []
    assert store.get_state().state != AWAIT_INPUT


@pytest.mark.parametrize("command", ["next milestone", "project complete"])
def test_pr_open_commands_are_operator_only(tmp_path: Path, command: str) -> None:
    store, transport = run_to_pr(tmp_path, FakePublisher())
    transport.receive("mallory", command)
    assert store.get_state().state == PR_OPEN