nisse/nisse-dev/tests/test_views.py

"""Per-role transcript views (architecture L6; Development Plan M1.2)."""

import io
from pathlib import Path
from typing import Any

from nisse_dev.canonical import (
    AssistantText,
    CanonicalEvent,
    DelegationCompleted,
    DelegationStarted,
    Fact,
    PlanSubmitted,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)
from nisse_dev.control import MILESTONE_PLANNING, PR_OPEN, Orchestrator
from nisse_dev.executor import ExecutorOutcome
from nisse_dev.gitea import Comment, 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.render import OpenAIChatRenderer
from nisse_dev.render_anthropic import AnthropicMessagesRenderer
from nisse_dev.state import StateStore
from nisse_dev.transport import InMemoryTransport

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


def new_store(tmp_path: Path) -> StateStore:
    return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))


def texts(events: list[StoredEvent]) -> list[str]:
    return [getattr(e.event, "text", "") for e in events]


# --- the orchestrator: run 2's planner, after a milestone ----------------------


class PlanningClient:
    """Always submits a plan, recording the events of each request."""

    def __init__(self) -> None:
        self.requests: list[list[StoredEvent]] = []

    def complete(self, events, tools, model, system=None) -> list[CanonicalEvent]:
        self.requests.append(events)
        n = len(self.requests)
        return [
            ToolCall(
                call_id=f"plan-{n}",
                tool="submit_plan",
                arguments={"content": f"plan {n}"},
            )
        ]


class WorkingExecutor:
    """Leaves the trail a real executor run does: text, calls, results."""

    def __init__(self) -> None:
        self.views: list[list[StoredEvent]] = []

    def run(self, transcript: Any, milestone: int) -> ExecutorOutcome:
        self.views.append(transcript.view())
        transcript.append(AssistantText(text="I'll write the helper", model="x"))
        call = ToolCall(
            call_id=f"run-{len(self.views)}",
            tool="run_command",
            arguments={"command": "pytest"},
        )
        transcript.append(call)
        transcript.append(
            ToolResult(call_id=call.call_id, result="1 passed", success=True)
        )
        return ExecutorOutcome("completed", "added truncate() with a test")


class Publisher:
    def reconcilers(self) -> dict[str, Any]:
        return {}

    def publish(self, **kwargs: Any) -> PublishResult:
        return PublishResult(
            PullRequest(number=165, state="open", head="h", url="https://g/pr/165"),
            "opened PR #165",
        )

    def push_followup(self, **kwargs: Any) -> str | None:
        return None

    def comment(self, index: int, body: str) -> None:
        pass


def run_one_milestone(
    tmp_path: Path,
) -> tuple[Orchestrator, StateStore, InMemoryTransport, PlanningClient, Any]:
    store = new_store(tmp_path)
    transport = InMemoryTransport()
    client = PlanningClient()
    executor = WorkingExecutor()
    orch = Orchestrator(
        store,
        transport,
        LLMPlanningAgent(client, ROLES),
        operators={"don"},
        roles=ROLES,
        executor=executor,
        publisher=Publisher(),
        gitea_operators={"don"},
    )
    orch.start()
    for text in ("add a truncate helper", "use demo", "approve", "approve"):
        transport.receive("don", text)
    transport.receive("don", "start milestone 1")
    assert store.get_state().state == PR_OPEN
    return orch, store, transport, client, executor


def test_the_planner_after_a_milestone_sees_facts_not_executor_turns(
    tmp_path: Path,
) -> None:
    # Run 2: shown the executor's calls as its own turns, the planner took
    # milestone 1's work for its own and planned around it (#78).
    _, store, transport, client, _ = run_one_milestone(tmp_path)
    transport.receive("don", "next milestone")
    assert store.get_state().state != MILESTONE_PLANNING  # it planned

    events = client.requests[-1]
    body = OpenAIChatRenderer().render(events, tools=None, model="y")
    assistant = [m for m in body["messages"] if m["role"] == "assistant"]
    called = [c["function"]["name"] for m in assistant for c in m["tool_calls"]]
    assert set(called) == {"submit_plan"}  # its own turns, and only those
    seen = "\n".join(texts(events))
    assert "I'll write the helper" not in seen
    assert "[operator, Matrix, sent while idle] add a truncate helper" in seen
    assert "[orchestrator] arch plan rev-1 approved by don" in seen
    assert "milestone 1 finished: added truncate() with a test" in seen
    assert "PR #165" in seen and "https://g/pr/165" in seen
    assert "planning milestone 2" in seen


def test_the_executor_sees_the_milestone_start_and_the_comment_source(
    tmp_path: Path,
) -> None:
    orch, _, _, _, executor = run_one_milestone(tmp_path)
    assert "[orchestrator] milestone 1 started" in texts(executor.views[0])

    orch.handle_review_comment(
        Comment(id=7, author="don", body="add a docstring", created_at="t")
    )
    review_view = texts(executor.views[-1])
    assert "[operator, comment on PR #165] add a docstring" in review_view
    # Its own earlier turn is still its own, not relabelled.
    own = [e for e in executor.views[-1] if isinstance(e.event, AssistantText)]
    assert [e.event.text for e in own] == ["I'll write the helper"]


def test_the_outcome_is_recorded_as_a_fact_when_the_task_closes(
    tmp_path: Path,
) -> None:
    _, store, transport, _, _ = run_one_milestone(tmp_path)
    item = store.active_item()
    transport.receive("don", "project complete")
    facts = [
        e.event.text
        for e in store.transcript(item.id, "orchestrator").read()
        if isinstance(e.event, Fact)
    ]
    assert facts[-1] == "work item completed: don said project complete"


# --- the projection, per role ----------------------------------------------------


def delegated_milestone(store: StateStore) -> None:
    """A milestone in which the executor failed and handed off to a helper."""
    store.transcript("a", "planner").append(PlanSubmitted(plan="1. fix it", model="y"))
    executor = store.transcript("a", "executor")
    executor.append(
        ToolCall(call_id="e1", tool="run_command", arguments={"command": "make"})
    )
    executor.append(ToolResult(call_id="e1", result="error: missing ;", success=False))
    helper = executor.as_actor("helper")
    helper.append(
        DelegationStarted(brief="make fails", role="helper", model="y", caps={})
    )
    helper.append(AssistantText(text="looking at main.c", model="y"))
    helper.append(
        ToolCall(call_id="h1", tool="read_file", arguments={"path": "main.c"})
    )
    helper.append(ToolResult(call_id="h1", result="int main()", success=True))
    helper.append(
        DelegationCompleted(
            handback="fixed the missing semicolon",
            artifacts=["main.c"],
            tokens=0,
            duration_s=1.0,
            exit_status="completed",
        )
    )


def test_the_executor_sees_the_helpers_report_not_its_turns(tmp_path: Path) -> None:
    store = new_store(tmp_path)
    delegated_milestone(store)
    view = store.transcript("a", "executor").view()
    assert [type(e.event).__name__ for e in view] == [
        "ToolCall",
        "ToolResult",
        "UserMessage",
    ]
    assert view[-1].event.text == (
        "[helper report, completed] fixed the missing semicolon"
    )


def test_the_helper_sees_the_plan_and_the_executors_work_as_narration(
    tmp_path: Path,
) -> None:
    store = new_store(tmp_path)
    delegated_milestone(store)
    view = store.transcript("a", "helper").view()
    said = texts(view)
    assert said[0] == "[approved plan] 1. fix it"
    assert said[1] == "[executor] run_command(make) → failed: error: missing ;"
    # Its own events are unchanged, tool calls and all.
    assert any(isinstance(e.event, ToolCall) and e.event.call_id == "h1" for e in view)
    assert not any(
        isinstance(e.event, ToolCall) and e.event.call_id == "e1" for e in view
    )


def test_the_planner_sees_no_other_roles_work(tmp_path: Path) -> None:
    store = new_store(tmp_path)
    delegated_milestone(store)
    store.transcript("a", "orchestrator").append(Fact(text="milestone 1 finished"))
    assert texts(store.transcript("a", "planner").view()) == [
        "[orchestrator] milestone 1 finished"
    ]


def test_a_view_keeps_log_ids_so_cursors_still_work(tmp_path: Path) -> None:
    store = new_store(tmp_path)
    delegated_milestone(store)
    raw = store.transcript("a", "executor").read()
    view = store.transcript("a", "executor").view(after=raw[2].id)
    assert [e.id for e in view] == [raw[-1].id]  # the helper's report


# --- rendering ---------------------------------------------------------------------


def test_both_renderers_show_a_fact_as_a_labelled_user_message() -> None:
    events = [StoredEvent(id=1, ts="", event=Fact(text="PR #3 opened"))]
    openai = OpenAIChatRenderer().render(events, tools=None, model="m")
    assert openai["messages"] == [
        {"role": "user", "content": "[orchestrator] PR #3 opened"}
    ]
    anthropic = AnthropicMessagesRenderer().render(events, tools=None, model="m")
    assert anthropic["messages"] == [
        {
            "role": "user",
            "content": [{"type": "text", "text": "[orchestrator] PR #3 opened"}],
        }
    ]


def test_operator_messages_carry_their_channel(tmp_path: Path) -> None:
    store = new_store(tmp_path)
    operator = store.transcript("a", "operator")
    for source in ("transport:@don", "gitea:issue:9:comment:4", "gitea:comment:5"):
        operator.append(UserMessage(text="hi", source=source))
    assert texts(store.transcript("a", "planner").view()) == [
        "[operator, Matrix] hi",
        "[operator, comment on plan issue #9] hi",
        "[operator, Gitea comment] hi",
    ]