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. A VM the
driver finds broken blocks its projects until the operator says it's ready
(M4.3)."""

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.gitea import Issue
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, **more: Any
) -> 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),
        **more,
    )
    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_blocks_the_vm_until_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()
    [block] = world["store"].vm_blocks().values()
    assert block.cause == "host key"
    [asked] = [t for t in said(transport) if "different host key" in t]
    assert f"`vm ready {project}`" in asked and "impersonating" in asked
    assert "has a new host key" in agent.status()
    say(transport, world, "it's fine")  # an answer isn't the operator's word
    assert world["store"].vm_blocks() and forgotten == []
    changed["on"] = False
    transport.receive(DON, f"vm ready {project}")
    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


# --- M4.3: a broken VM blocks its projects until it's ready ---------------------------


def reimage(reason: str = "pip crashes in every venv, even a new one") -> list[Any]:
    return [call("request_reimage", reason=reason)]


def by_issue(world: dict[str, Any], number: int) -> Any:
    [found] = [i for i in world["store"].work_items() if loop_of(i).issue == number]
    return found


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


def test_a_broken_vm_blocks_its_project_until_the_operator_says_ready(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    world["forge"].issues.append(Issue(15, "Add count_lines", "count lines", "don", ()))
    forgotten: list[str] = []
    monkeypatch.setattr(
        type(world["host"]), "forget_host_key", lambda self: forgotten.append("yes")
    )
    # Asked for before the go-ahead: it's a request, not an action.
    model = ScriptedModel(ready(), reimage(), ready(), RESTATE, RESTATE)
    agent, transport = agent_for(world, project, model)
    agent.intake()
    first, second = by_issue(world, 14), by_issue(world, 15)
    assert "build VM reimage requested: pip crashes in every venv, even a new one" in (
        facts_of(world, first)
    )
    [asked] = [
        m for m in transport.outbound if "reimage it from its template" in m.text
    ]
    assert asked.thread is None  # top level: it can cover several projects
    assert "pip crashes in every venv" in asked.text
    assert f"`vm ready {project}`" in asked.text
    assert any(f.startswith("Blocked: the build VM") for f in facts_of(world, second))
    # Neither runs while the VM is blocked, even after a restart.
    again, _ = agent_for(world, project, ScriptedModel())
    assert again._queue() == []
    assert "its build VM needs reimaging" in again.status()
    # The operator reimages it: the clone and the marker are gone.
    shutil.rmtree(world["repo"].root)
    world["host"].execute(f'rm -f "{preflight.marker_path(project, markers(world))}"')
    transport.receive(DON, f"vm ready {project}")
    assert forgotten == ["yes"]  # its new host key is trusted
    assert world["store"].vm_blocks() == {}
    assert any("cloned the workspace again" in f for f in facts_of(world, first))
    # The item that found it went first, and set the VM up again.
    assert loop_of(by_issue(world, 14)).wait == GO_AHEAD
    assert loop_of(by_issue(world, 15)).wait == GO_AHEAD
    setups = [
        f
        for item in (first, second)
        for f in facts_of(world, item)
        if "setting the build VM up from the repository" in f
    ]
    assert len(setups) == 2  # the first, and after the reimage
    assert setups == [f for f in facts_of(world, first) if "setting the build VM" in f]


def test_a_vm_said_ready_without_a_reimage_goes_straight_on(
    world: dict[str, Any], project: str
) -> None:
    model = ScriptedModel(
        ready(),
        RESTATE,
        [call("start_milestone", name="m"), *reimage("the build hangs")],
        [*WORK[1:], *FINISH],
    )
    agent, transport = agent_for(world, project, model)
    agent.intake()
    say(transport, world, "go")
    assert world["store"].vm_blocks()
    transport.receive(DON, "vm ready")  # the only blocked VM: no name needed
    # Not reimaged: the marker is still there, so no new setup step.
    assert (
        sum("setting the build VM up from the repository" in f for f in facts(world))
        == 1
    )
    assert world["forge"].prs  # the work went on to a PR


def test_vm_ready_with_nothing_blocked_says_so(
    world: dict[str, Any], project: str
) -> None:
    _, transport = agent_for(world, project, ScriptedModel())
    transport.receive(DON, "vm ready")
    transport.receive(DON, f"vm ready {project}")
    transport.receive(DON, "vm ready nowhere")
    assert said(transport) == [
        "No build VM is blocked.",
        f"Nothing is blocking {project}'s build VM.",
        "I don't know a project called nowhere.",
    ]


def test_work_in_progress_is_pushed_before_a_reimage(
    world: dict[str, Any], project: str
) -> None:
    model = ScriptedModel(
        ready(),
        RESTATE,
        [
            call("start_milestone", name="m"),
            call("write_file", path="w.py", content="x = 1\n"),
            *reimage(),
        ],
        ready(),  # set up again after the reimage
        [call("finish_milestone", summary="w.py, tested", final="yes")],
    )
    agent, transport = agent_for(world, project, model)
    agent.intake()
    say(transport, world, "go")
    [asked] = [t for t in said(transport) if "reimage it from its template" in t]
    assert "The work in progress is pushed" in asked
    shutil.rmtree(world["repo"].root)
    world["host"].execute(f'rm -f "{preflight.marker_path(project, markers(world))}"')
    transport.receive(DON, f"vm ready {project}")
    forge = world["forge"]
    [pr] = [n for n, p in forge.prs.items() if not p["head"].startswith("nisse-check")]
    assert "w.py" in forge.files(pr)  # it survived the reimage