nisse/nisse-dev/tests/test_control.py

import io
import json
from pathlib import Path

import pytest

from nisse_dev.canonical import UserMessage
from nisse_dev.control import (
    AWAIT_ARCH_SIGNOFF,
    AWAIT_INPUT,
    AWAIT_MILESTONE_SIGNOFF,
    EXECUTING,
    IDLE,
    PLANNING_ARCH,
    AskHuman,
    Orchestrator,
    PlanningError,
    StubAgent,
    SubmitPlan,
)
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
from nisse_dev.transport import InMemoryTransport


class Harness:
    def __init__(self, tmp_path: Path, agent: object | None = None) -> None:
        self.stream = io.StringIO()
        self.store = StateStore(
            tmp_path / "state.db", oplog=OpLog("nisse-dev", self.stream)
        )
        self.transport = InMemoryTransport()
        self.orch = Orchestrator(
            self.store, self.transport, agent or StubAgent(), operators={"don"}
        )
        self.orch.start()

    @property
    def state(self) -> str:
        state = self.store.get_state()
        assert state is not None
        return state.state

    def events(self, kind: str) -> list[dict]:
        return [
            e
            for e in map(json.loads, self.stream.getvalue().splitlines())
            if e["event"] == kind
        ]

    def outbound_texts(self) -> list[str]:
        return [m.text for m in self.transport.outbound]


@pytest.fixture
def h(tmp_path: Path) -> Harness:
    return Harness(tmp_path)


def test_boots_idle_and_greets(h: Harness) -> None:
    assert h.state == IDLE
    assert "ready" in h.outbound_texts()[0]


def test_use_project_submits_arch_plan_and_opens_gate(h: Harness) -> None:
    h.transport.receive("don", "use demo")
    assert h.state == AWAIT_ARCH_SIGNOFF
    request = h.transport.outbound[-1]
    assert request.kind == "approval_request"
    assert "rev-1" in request.text
    [opened] = h.events("gate.open")
    assert opened["gate"] == "arch"
    assert opened["rev"] == "rev-1"
    assert opened["task_id"] is not None


def test_full_happy_path_to_executing(h: Harness) -> None:
    h.transport.receive("don", "use demo")
    h.transport.receive("don", "approve")
    assert h.state == AWAIT_MILESTONE_SIGNOFF
    h.transport.receive("don", "approve")
    assert h.state == AWAIT_MILESTONE_SIGNOFF  # approved, awaiting start
    h.transport.receive("don", "start milestone 1")
    assert h.state == EXECUTING
    approvals = h.events("gate.approve")
    assert [e["gate"] for e in approvals] == ["arch", "milestone-1"]


def test_reject_replans_with_feedback_and_bumps_revision(h: Harness) -> None:
    h.transport.receive("don", "use demo")
    h.transport.receive("don", "too vague, add tests")
    assert h.state == AWAIT_ARCH_SIGNOFF
    request = h.transport.outbound[-1]
    assert "rev-2" in request.text
    assert "too vague, add tests" in request.text
    [rejected] = h.events("gate.reject")
    assert rejected["rev"] == "rev-1"


def test_stale_approval_is_refused_then_current_works(h: Harness) -> None:
    h.transport.receive("don", "use demo")
    h.transport.receive("don", "needs work")  # now at rev-2
    h.transport.receive("don", "approve rev-1")
    assert h.state == AWAIT_ARCH_SIGNOFF
    assert "not applied" in h.outbound_texts()[-1]
    assert h.events("gate.approve") == []
    h.transport.receive("don", "approve rev-2")
    assert h.state == AWAIT_MILESTONE_SIGNOFF


def test_non_operator_messages_are_ignored(h: Harness) -> None:
    h.transport.receive("don", "use demo")
    before = len(h.transport.outbound)
    h.transport.receive("mallory", "approve")
    assert h.state == AWAIT_ARCH_SIGNOFF
    assert len(h.transport.outbound) == before


def test_start_milestone_requires_approval_first(h: Harness) -> None:
    h.transport.receive("don", "use demo")
    h.transport.receive("don", "approve")
    h.transport.receive("don", "start milestone 1")
    assert h.state == AWAIT_MILESTONE_SIGNOFF  # it was rejection feedback
    h.transport.receive("don", "approve rev-2")
    h.transport.receive("don", "start milestone 1")
    assert h.state == EXECUTING


def test_a_message_is_matched_as_stripped_but_recorded_as_sent(
    h: Harness,
) -> None:
    h.transport.receive("don", "use demo")
    h.transport.receive("don", "approve", original="Nisse Dev: approve")
    assert h.state == AWAIT_MILESTONE_SIGNOFF
    recorded = [
        e.event.text for e in h.store.read_events() if isinstance(e.event, UserMessage)
    ]
    assert "Nisse Dev: approve" in recorded
    assert "approve" not in recorded


def test_status_reports_state_and_gate(h: Harness) -> None:
    h.transport.receive("don", "use demo")
    h.transport.receive("don", "status")
    status = h.outbound_texts()[-1]
    assert "AWAIT_ARCH_SIGNOFF" in status
    assert "demo" in status
    assert "rev-1" in status


class QuestionThenPlan:
    def plan(
        self,
        gate: str,
        project: str,
        feedback: str | None = None,
        answer: str | None = None,
    ) -> SubmitPlan | AskHuman:
        if answer is None:
            return AskHuman("monorepo or polyrepo?")
        return SubmitPlan(f"[stub] {gate} plan using {answer}")


def test_ask_human_pauses_and_answer_resumes(tmp_path: Path) -> None:
    h = Harness(tmp_path, agent=QuestionThenPlan())
    h.transport.receive("don", "use demo")
    assert h.state == AWAIT_INPUT
    assert h.transport.outbound[-1].kind == "question"
    h.transport.receive("don", "monorepo")
    assert h.state == AWAIT_ARCH_SIGNOFF
    assert "monorepo" in h.transport.outbound[-1].text


# --- restarts (A3 steps 7 and 8) --------------------------------------------


class FailsOnce:
    """A planner whose first turn fails, as Opus 5.5's did on its first live
    run when asked for forced tool use."""

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

    def plan(
        self,
        gate: str,
        project: str,
        feedback: str | None = None,
        answer: str | None = None,
    ) -> SubmitPlan | AskHuman:
        self.calls += 1
        if self.calls == 1:
            raise PlanningError("planner returned 400")
        return SubmitPlan(f"[stub] {gate} plan for {project}")


def test_restart_while_awaiting_input_asks_the_same_question(
    tmp_path: Path,
) -> None:
    agent = FailsOnce()
    h = Harness(tmp_path, agent=agent)
    h.transport.receive("don", "use demo")
    assert h.state == AWAIT_INPUT
    h = Harness(tmp_path, agent=agent)  # restart on the same state
    question = h.transport.outbound[-1]
    assert question.kind == "question"
    assert "restarted while waiting on you" in question.text
    assert "planning for arch failed" in question.text
    h.transport.receive("don", "Please retry.")
    assert h.state == AWAIT_ARCH_SIGNOFF


def test_restart_mid_planning_offers_continue_and_it_replans(
    tmp_path: Path,
) -> None:
    h = Harness(tmp_path)
    h.store.transition(PLANNING_ARCH, cause="test", project="demo", data={})
    h = Harness(tmp_path)
    assert h.state == AWAIT_INPUT
    assert "'continue'" in h.outbound_texts()[-1]
    h.transport.receive("don", "Please retry.")  # neither choice: asked again
    assert h.state == AWAIT_INPUT
    assert h.transport.outbound[-1].kind == "question"
    h.transport.receive("don", "continue")
    assert h.state == AWAIT_ARCH_SIGNOFF


def test_restart_mid_execution_continue_returns_to_the_approved_plan(
    h: Harness, tmp_path: Path
) -> None:
    h.transport.receive("don", "use demo")
    h.transport.receive("don", "approve")
    h.transport.receive("don", "approve")
    h.transport.receive("don", "start milestone 1")
    assert h.state == EXECUTING
    h = Harness(tmp_path)
    assert h.state == AWAIT_INPUT
    h.transport.receive("don", "continue")
    assert h.state == AWAIT_MILESTONE_SIGNOFF
    assert "start milestone 1" in h.outbound_texts()[-1]
    h.transport.receive("don", "start milestone 1")
    assert h.state == EXECUTING


def test_restart_question_without_its_interrupted_state_still_replans(
    tmp_path: Path,
) -> None:
    # What the first live run left behind: a restart question recorded
    # before the question kept its prompt or the state it interrupted.
    h = Harness(tmp_path)
    h.store.transition(
        AWAIT_INPUT, cause="test", project="demo", data={"pending": {"kind": "boot"}}
    )
    h = Harness(tmp_path)
    assert "my last question still stands" in h.outbound_texts()[-1]
    h.transport.receive("don", "re-plan")
    assert h.state == AWAIT_ARCH_SIGNOFF


def test_use_starts_over_when_nothing_is_pending(h: Harness) -> None:
    h.store.transition(AWAIT_INPUT, cause="forced", project="demo", data={})
    h.transport.receive("don", "use demo")
    assert h.state == AWAIT_ARCH_SIGNOFF