nisse/nisse-dev/tests/test_loop_toolchain.py

"""The toolchain step in the loop (D3; M4.1, M4.2): before the driver plans,
the build VM is checked (baseline, clone, marker); without a marker the harness
starts a toolchain step that sets the VM up from the repository and edits
nothing there; the driver can ask for one once work is agreed, whose edits are
part of the work; later runs check versions again; a rebuilt VM is re-cloned
and set up again, and a changed host key waits for the operator."""

import io
import shutil
import uuid
from pathlib import Path
from typing import Any

import pytest
from test_loop_milestones import (
    DON,
    FINISH,
    RESTATE,
    WORK,
    ScriptedModel,
    call,
    say,
)
from test_loop_milestones import item as the_item
from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture

from nisse_dev.cancel import Cancel
from nisse_dev.canonical import Fact, ToolResult
from nisse_dev.execution import HostKeyChanged
from nisse_dev.loop import preflight
from nisse_dev.loop.agent import LoopAgent, Project
from nisse_dev.loop.driver import Driver
from nisse_dev.loop.items import ANSWER, GO_AHEAD, WAITING, loop_of
from nisse_dev.oplog import OpLog
from nisse_dev.tools import Workspace
from nisse_dev.transport import InMemoryTransport
from nisse_dev.worker import InlineWorker

PYTHON = {"name": "python", "version_command": "python3 --version"}


def ready(**more: Any) -> list[Any]:
    """A preflight that finds Python and reports it."""
    return [
        call("run_command", command="python3 --version"),
        call(
            "toolchain_ready",
            summary="Python is there",
            toolchain=[PYTHON],  # type: ignore[arg-type]
            **more,
        ),
    ]


@pytest.fixture
def project() -> str:
    return f"pf-{uuid.uuid4().hex[:8]}"


def markers(world: dict[str, Any]) -> str:
    """Markers in the test's own directory, not this host's home."""
    return str(Path(world["repo"].root).parent / "markers")


def agent_for(
    world: dict[str, Any], project: str, model: ScriptedModel
) -> tuple[LoopAgent, InMemoryTransport]:
    transport = InMemoryTransport()
    cancel = Cancel()
    loop = LoopAgent(
        world["store"],
        transport,
        world["forge"],
        projects=[
            Project(
                project,
                "don/sandbox",
                Workspace(world["host"], world["repo"].root, timeout_s=30),
                world["milestones"],
            )
        ],
        account="nisse-dev",
        operators={DON},
        gitea_operators={"don"},
        driver=Driver(model, "m", cancel=cancel),  # type: ignore[arg-type]
        worker=InlineWorker(),
        cancel=cancel,
        oplog=OpLog("nisse-dev", io.StringIO()),
        preflight=True,
        preflight_turns=5,
        markers=markers(world),
    )
    transport.on_message(loop.on_message)
    return loop, transport


def facts(world: dict[str, Any]) -> list[str]:
    events = world["store"].transcript(the_item(world).id, "orchestrator").read()
    return [e.event.text for e in events if isinstance(e.event, Fact)]


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


def test_the_setup_step_comes_before_planning(
    world: dict[str, Any], project: str
) -> None:
    model = ScriptedModel(ready(), RESTATE)
    agent, transport = agent_for(world, project, model)
    agent.intake()
    item = the_item(world)
    assert (item.position, loop_of(item).wait) == (WAITING, GO_AHEAD)
    known = facts(world)
    toolchain = [
        f for f in known if f.startswith("toolchain step: toolchain: python Python 3.")
    ]
    assert toolchain, known
    assert known.index(
        "toolchain step: setting the build VM up from the repository"
    ) < known.index("restatement rev-1 put up for a go-ahead")
    assert loop_of(item).toolchain[0][:2] == ("python", "python3 --version")
    marker = preflight.check(
        world["host"], project, str(world["repo"].root), markers(world)
    ).marker
    assert marker is not None and marker["complete"]
    assert not [t for t in said(transport) if t.startswith("Toolchain step:")]  # quiet


def test_the_setup_step_runs_once_then_versions_are_checked(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    model = ScriptedModel(ready(), RESTATE, [*WORK, *FINISH])
    agent, transport = agent_for(world, project, model)
    agent.intake()
    real = preflight.versions

    def newer(host: Any, tools: list[Any]) -> list[Any]:
        return [
            preflight.Tool(t.name, t.command, "Python 3.99.0")
            for t in real(host, tools)
        ]

    monkeypatch.setattr(preflight, "versions", newer)
    say(transport, world, "go")  # no second preflight: the script has none
    changed = [f for f in facts(world) if f.startswith("the toolchain changed")]
    assert changed and "→ Python 3.99.0" in changed[0]
    assert any("toolchain changed" in t for t in said(transport))
    [pr] = [n for n in world["forge"].prs]
    assert "Tested with: python Python 3.99.0" in world["forge"].prs[pr]["body"]


def test_only_version_commands_are_accepted(
    world: dict[str, Any], project: str
) -> None:
    bad = call(
        "toolchain_ready",
        summary="s",
        toolchain=[{"name": "py", "version_command": "python3 -c 'print(1)'"}],  # type: ignore[arg-type]
    )
    model = ScriptedModel([bad], ready(), RESTATE)
    agent, _ = agent_for(world, project, model)
    agent.intake()
    results = [
        e.event.result
        for e in world["store"].transcript(the_item(world).id, "driver").read()
        if isinstance(e.event, ToolResult)
    ]
    assert any("plain version commands" in r for r in results)
    assert loop_of(the_item(world)).toolchain


def test_what_was_installed_or_is_missing_is_said(
    world: dict[str, Any], project: str
) -> None:
    model = ScriptedModel(
        ready(
            installed=["libfoo-dev"],  # type: ignore[arg-type]
            missing=[{"name": "protoc", "why": "no package for this OS"}],  # type: ignore[arg-type]
        ),
        RESTATE,
    )
    agent, transport = agent_for(world, project, model)
    agent.intake()
    [line] = [t for t in said(transport) if t.startswith("Toolchain step:")]
    assert "installed: libfoo-dev" in line
    assert "missing: protoc (no package for this OS)" in line


def test_an_unfinished_step_is_said_and_work_goes_on(
    world: dict[str, Any], project: str
) -> None:
    from nisse_dev.canonical import AssistantText

    prose = [AssistantText(text="hmm", model="m")]
    model = ScriptedModel(prose, prose, RESTATE)
    agent, transport = agent_for(world, project, model)
    agent.intake()
    assert loop_of(the_item(world)).wait == GO_AHEAD  # it planned anyway
    assert any("the step didn't finish" in t for t in said(transport))
    marker = preflight.check(
        world["host"], project, str(world["repo"].root), markers(world)
    ).marker
    assert marker is not None and not marker["complete"]


def test_a_wiped_vm_is_cloned_again_and_set_up(
    world: dict[str, Any], project: str
) -> None:
    model = ScriptedModel(ready(), RESTATE, ready(), [*WORK, *FINISH])
    agent, transport = agent_for(world, project, model)
    agent.intake()
    # The VM is rebuilt: the clone and the marker are gone.
    shutil.rmtree(world["repo"].root)
    world["host"].execute(f'rm -f "{preflight.marker_path(project, markers(world))}"')
    say(transport, world, "go")
    assert any("cloned the workspace again" in f for f in facts(world))
    assert (Path(world["repo"].root) / ".git").exists()
    assert (
        sum("setting the build VM up from the repository" in f for f in facts(world))
        == 2
    )
    assert world["forge"].prs  # and the work went on to a PR


def test_a_missing_baseline_waits_on_the_operator(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(preflight, "BASELINE", ("git", "no-such-tool-xyz"))
    agent, _ = agent_for(world, project, ScriptedModel())
    agent.intake()
    loop = loop_of(the_item(world))
    assert loop.wait == ANSWER and "lacks no-such-tool-xyz" in loop.note


def test_a_changed_host_key_waits_for_the_operators_word(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    real = preflight.check
    changed = {"on": True}

    def check(host: Any, name: str, root: str, *rest: Any) -> Any:
        if changed["on"]:
            raise HostKeyChanged("the key changed")
        return real(host, name, root, *rest)

    forgotten: list[str] = []
    monkeypatch.setattr(preflight, "check", check)
    monkeypatch.setattr(
        type(world["host"]), "forget_host_key", lambda self: forgotten.append("yes")
    )
    agent, transport = agent_for(world, project, ScriptedModel(ready(), RESTATE))
    agent.intake()
    loop = loop_of(the_item(world))
    assert loop.host_key_changed and "trust new host key" in loop.note
    say(transport, world, "it's fine")  # an answer isn't the operator's word
    still = loop_of(the_item(world))
    assert (still.wait, still.host_key_changed, forgotten) == (ANSWER, True, [])
    changed["on"] = False
    say(transport, world, "trust new host key")
    assert forgotten == ["yes"]
    assert loop_of(the_item(world)).wait == GO_AHEAD  # set up, then planned


# --- M4.2: the step's modes, and the go-ahead ----------------------------------------


def test_the_setup_step_edits_nothing_in_the_repository(
    world: dict[str, Any], project: str
) -> None:
    step = [
        call("run_command", command="echo stray > stray.txt"),
        *ready(),
    ]
    agent, _ = agent_for(world, project, ScriptedModel(step, RESTATE))
    agent.intake()
    [said] = [f for f in facts(world) if f.startswith("toolchain step: toolchain")]
    assert "changed files it shouldn't have, left for the work: stray.txt" in said
    tools = [
        e.event.tool
        for e in world["store"].transcript(the_item(world).id, "toolchain").read()
        if e.actor == "toolchain" and hasattr(e.event, "tool")
    ]
    assert "write_file" not in tools  # not offered in setup


def test_a_requested_step_brings_the_projects_files_in_step(
    world: dict[str, Any], project: str
) -> None:
    requested = [
        call("write_file", path="README.md", content="# demo\n\nNeeds Python 3.12.\n"),
        *ready(installed=["httpx"]),  # type: ignore[arg-type]
    ]
    model = ScriptedModel(
        ready(),
        RESTATE,
        [
            call("start_milestone", name="m"),
            call("toolchain_step", reason="it needs httpx, and the README says so"),
        ],
        requested,
        [
            call("write_file", path="w.py", content="import httpx\n"),
            call("finish_milestone", summary="uses httpx", final="yes"),
        ],
    )
    agent, transport = agent_for(world, project, model)
    agent.intake()
    say(transport, world, "go")
    forge = world["forge"]
    [pr] = [n for n, p in forge.prs.items() if not p["head"].startswith("nisse-check")]
    assert sorted(forge.files(pr)) == ["README.md", "w.py"]  # reviewed with the work
    driver = [
        e.event for e in world["store"].transcript(the_item(world).id, "driver").view()
    ]
    results = [getattr(e, "result", "") for e in driver]
    assert any(r.startswith("toolchain step done: toolchain: python") for r in results)
    # The step's own turns stay in its transcript.
    calls = [getattr(e, "arguments", {}).get("path") for e in driver]
    assert "README.md" not in calls


def test_before_the_go_ahead_the_driver_only_looks(
    world: dict[str, Any], project: str
) -> None:
    looking = [
        call("run_command", command="python3 --version"),
        call("run_command", command="pip install httpx"),
    ]
    restate = [call("restate", text="add count_words to text.py; it needs httpx")]
    agent, _ = agent_for(world, project, ScriptedModel(ready(), looking, restate))
    agent.intake()
    results = [
        e.event.result
        for e in world["store"].transcript(the_item(world).id, "driver").read()
        if isinstance(e.event, ToolResult) and e.actor == "driver"
    ]
    assert "Python 3." in results[0]  # looking runs
    assert results[1].startswith("not run: not before the go-ahead")
    assert loop_of(the_item(world)).wait == GO_AHEAD