nisse/tests/integration/test_end_to_end.py

"""The end-to-end test (Development Plan M1.6): the whole agent, assembled as
it runs in production, against real models, the Gitea sandbox and a real
build host, driven by a scripted operator.

It automates the checks of bring-up run 3:
- a milestone reaches a PR on the sandbox that contains only the work, no
  build artifacts (#80), pushed with the push credential;
- the clone on the build host holds no credential (#89);
- after the milestone, the planner's view carries the orchestrator's facts
  and none of the executor's turns (#78).

Only the Matrix transport is replaced, by the in-memory one: the operator's
Matrix credentials are not something a test should hold. Run it with
scripts/e2e.sh after each milestone; it spends model tokens and takes minutes.
"""

import contextlib
import os
import shlex
import uuid
from collections.abc import Iterator
from dataclasses import replace
from fnmatch import fnmatch
from pathlib import Path
from typing import Any

import pytest
from e2e_support import base_branch, operator_gitea
from e2e_support import task as make_task

from nisse_dev.agent import Agent, build_agent
from nisse_dev.canonical import UserMessage
from nisse_dev.config import AgentConfig, BuildHost
from nisse_dev.control import (
    AWAIT_ARCH_SIGNOFF,
    AWAIT_INPUT,
    AWAIT_MILESTONE_SIGNOFF,
    PR_OPEN,
)
from nisse_dev.execution import SSHHost
from nisse_dev.gitea import GiteaError
from nisse_dev.transport import InMemoryTransport

pytestmark = [pytest.mark.integration, pytest.mark.e2e]

OPERATOR = "@dongar:dongar.net"  # an allowlisted operator in nisse-test.yaml
SETTLE_S = 1200  # a milestone, with real models, inside one message


def build_host(ssh_host: SSHHost) -> BuildHost:
    """The operator's build VM when scripts/e2e.sh names one, otherwise the
    throwaway local sshd."""
    address = os.environ.get("NISSE_E2E_BUILD_HOST")
    if address:
        key = Path(os.environ["NISSE_E2E_BUILD_KEY"])
        return BuildHost(address=address, key=key)
    return BuildHost(
        address=ssh_host.address,
        key=ssh_host._key,
        user=ssh_host.user,
        port=ssh_host.port,
    )


@pytest.fixture
def agent(
    live_config: AgentConfig,
    ssh_host: SSHHost,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> Iterator[tuple[Agent, InMemoryTransport, str]]:
    run = uuid.uuid4().hex[:8]
    stack = contextlib.ExitStack()
    operator = operator_gitea(live_config.gitea_url)
    base = stack.enter_context(base_branch(operator, live_config.gitea_repo, run))
    config = replace(
        live_config.with_project(
            base_branch=base,
            workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
            build_host=build_host(ssh_host),
        ),
        state_db=tmp_path / "state.db",
    )
    transport = InMemoryTransport()
    monkeypatch.setattr("nisse_dev.agent.MatrixTransport", lambda cfg: transport)
    agent = None
    try:
        agent = build_agent(config)
        agent.start()
        yield agent, transport, run
    finally:
        if agent is not None:
            agent.stop()
        # The run's clone on the build host goes too, even if startup failed.
        host = SSHHost(
            config.build_host.address,
            key=config.build_host.key,
            user=config.build_host.user,
            port=config.build_host.port,
            state_dir=tmp_path / "ssh",
        )
        host.execute(f"rm -rf /tmp/nisse-e2e-{run}")
        stack.close()  # the run's base branch


class Operator:
    """Says things to the agent and waits for it to finish with each."""

    def __init__(self, agent: Agent, transport: InMemoryTransport) -> None:
        self.agent = agent
        self.transport = transport

    def say(self, text: str) -> None:
        self.transport.receive(OPERATOR, text)
        # Messages are handled in order on the agent's worker, so a no-op
        # behind this one returns when it, and the work it started, is done.
        self.agent.worker.call(lambda: None, timeout=SETTLE_S)

    def state(self) -> Any:
        return self.agent.worker.call(self.agent.store.get_state)

    def drive_to_pr(self, max_steps: int = 12) -> Any:
        """Answer whatever the agent actually does: a live model may ask a
        question at any gate, so a fixed script would break."""
        for _ in range(max_steps):
            state = self.state()
            gate = state.data.get("gate") or {}
            if state.state == PR_OPEN:
                return state
            if state.state == AWAIT_INPUT:
                self.say("Use your best judgement and proceed.")
            elif state.state == AWAIT_MILESTONE_SIGNOFF and gate.get("approved"):
                self.say("start milestone 1")
            elif state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF):
                self.say("approve")
            else:
                raise AssertionError(f"stuck in {state.state}: {self.last()}")
        raise AssertionError(f"no PR after {max_steps} steps: {self.last()}")

    def last(self) -> str:
        return self.transport.outbound[-1].text if self.transport.outbound else ""


# The test's own idea of an artifact, independent of the code's list, so
# emptying that list can't make this pass.
def is_artifact(path: str) -> bool:
    parts = path.split("/")
    return (
        "__pycache__" in parts
        or ".pytest_cache" in parts
        or fnmatch(parts[-1], "*.py[co]")
    )


def test_a_milestone_end_to_end(
    agent: tuple[Agent, InMemoryTransport, str],
) -> None:
    running, transport, run = agent
    operator = Operator(running, transport)
    gitea, repo = running.gitea, running.config.gitea_repo
    pr: dict[str, Any] = {}
    try:
        # The task first, before any project: it must be kept, not lost.
        operator.say(make_task(run))
        operator.say(f"use {running.config.project}")
        state = operator.drive_to_pr()
        pr = state.data["pr"]
        task = state.data["task_id"]

        # --- the PR holds the work and only the work (D4, #80) ---
        files = gitea._request("GET", f"/repos/{repo}/pulls/{pr['number']}/files")
        changed = [f["filename"] for f in files]
        assert changed, "the PR is empty"
        assert not [p for p in changed if is_artifact(p)], changed
        found = gitea._request("GET", f"/repos/{repo}/pulls/{pr['number']}")
        assert found["head"]["ref"] == f"{task}-milestone-1"

        # --- no credential in the clone on the build host (#89) ---
        host = SSHHost(
            running.config.build_host.address,
            key=running.config.build_host.key,
            user=running.config.build_host.user,
            port=running.config.build_host.port,
            state_dir=running.config.state_db.parent / "ssh",
        )
        config = host.execute(
            f"cat {shlex.quote(str(running.config.workspace / '.git' / 'config'))}"
        ).stdout
        assert "PRIVATE KEY" not in config
        assert running.config.gitea_token not in config

        # --- the planner sees facts, not the executor's turns (#78) ---
        operator.say("next milestone")
        store = running.store
        view = running.worker.call(lambda: store.transcript(task, "planner").view())
        said = [e.event.text for e in view if isinstance(e.event, UserMessage)]
        assert any("[orchestrator] milestone 1 finished" in t for t in said), said
        assert any("[orchestrator] milestone 1 published" in t for t in said), said
        assert not [e for e in view if e.actor == "executor"], "executor turns leaked"
    finally:
        # Leave the sandbox as it was: closes the PR and the plan issue.
        operator.say("abandon")
        if pr:
            branch = gitea._request("GET", f"/repos/{repo}/pulls/{pr['number']}")[
                "head"
            ]["ref"]
            with contextlib.suppress(GiteaError):  # cleanup is best effort
                gitea._request("DELETE", f"/repos/{repo}/branches/{branch}")