nisse/nisse-dev/tests/test_loop_multi.py

"""Multi-milestone work on the new loop (M2.4): each milestone starts from
the latest default branch, the next waits for the previous PR's merge (read
from Gitea when asked), earlier PRs are part of the issue and the last closes
it, and the item closes only with nothing left open (D1, D2, D4). Real git on
the throwaway build host; a fake Gitea."""

from typing import Any

from test_loop_milestones import (
    RESTATE,
    Forge,
    ScriptedModel,
    call,
    git,
    item,
    loop_agent,
    say,
)
from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture

from nisse_dev.canonical import ToolCall, ToolResult
from nisse_dev.loop.items import MERGE, WAITING, loop_of


def milestone(path: str, final: str) -> list[Any]:
    return [
        call("start_milestone", name=path),
        call("write_file", path=path, content=f"# {path}\n"),
        call("finish_milestone", summary=f"adds {path}", title=path, final=final),
    ]


def tree(world: dict[str, Any], branch: str) -> list[str]:
    bare = str(world["bare"])
    return git("--git-dir", bare, "ls-tree", "--name-only", branch).split()


def result_of(world: dict[str, Any], tool: str) -> str:
    events = world["store"].transcript(item(world).id, "driver").read()
    ids = {
        e.event.call_id
        for e in events
        if isinstance(e.event, ToolCall) and e.event.tool == tool
    }
    results = [
        e.event.result
        for e in events
        if isinstance(e.event, ToolResult) and e.event.call_id in ids
    ]
    return results[-1]


def test_the_second_milestone_builds_on_the_merged_first(world: dict[str, Any]) -> None:
    forge: Forge = world["forge"]
    model = ScriptedModel(
        RESTATE,
        milestone("a.py", final="no"),
        milestone("b.py", final="yes"),
        [call("close_item", summary="both done")],
    )
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    [first] = forge.prs
    assert forge.prs[first]["body"].endswith("Part of #14")
    forge.merge(first)
    agent.watch()
    agent._schedule()

    second = max(forge.prs)
    assert second != first
    assert "Closes #14" in forge.prs[second]["body"]
    assert forge.files(second) == ["b.py"]  # only its own work
    head = forge.prs[second]["head"]
    assert "a.py" in tree(world, head)  # cut from main, with the first merged
    assert loop_of(item(world)).milestone == 2

    forge.merge(second)
    agent.watch()
    agent._schedule()
    assert item(world).status == "closed"


def test_the_next_milestone_waits_for_the_merge(world: dict[str, Any]) -> None:
    forge: Forge = world["forge"]
    model = ScriptedModel(
        RESTATE,
        milestone("a.py", final="no"),
        [  # a message arrives while the PR is open: try to move on
            call("start_milestone", name="b"),
            call("wait_for_review", note="waiting for the merge"),
        ],
    )
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    say(transport, world, "how's it going?")
    assert "hasn't merged yet" in result_of(world, "start_milestone")
    assert (item(world).position, loop_of(item(world)).wait) == (WAITING, MERGE)
    assert len(forge.prs) == 1


def test_a_merge_not_yet_watched_is_read_from_gitea(world: dict[str, Any]) -> None:
    forge: Forge = world["forge"]
    model = ScriptedModel(
        RESTATE,
        milestone("a.py", final="no"),
        milestone("b.py", final="yes"),
    )
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    [first] = forge.prs
    forge.merge(first)  # merged in Gitea; no watch round has seen it
    say(transport, world, "merged, carry on")
    assert len(forge.prs) == 2
    assert "a.py" in tree(world, forge.prs[max(forge.prs)]["head"])


def test_a_redo_after_a_closed_pr_drops_the_rejected_work(
    world: dict[str, Any],
) -> None:
    forge: Forge = world["forge"]
    model = ScriptedModel(
        RESTATE,
        milestone("a.py", final="yes"),
        milestone("c.py", final="yes"),
    )
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    [first] = forge.prs
    forge.prs[first]["state"] = "closed"  # rejected
    agent.watch()
    agent._schedule()
    redo = max(forge.prs)
    assert forge.files(redo) == ["c.py"]
    assert "a.py" not in tree(world, forge.prs[redo]["head"])


def test_closing_with_a_pr_open_waits_for_the_merge(world: dict[str, Any]) -> None:
    forge: Forge = world["forge"]
    model = ScriptedModel(
        RESTATE,
        milestone("a.py", final="yes"),
        [call("close_item", summary="done, I think")],  # too soon
        [call("close_item", summary="done")],
    )
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    say(transport, world, "thanks")
    assert item(world).status == "open"
    assert (item(world).position, loop_of(item(world)).wait) == (WAITING, MERGE)
    [pr] = forge.prs
    forge.merge(pr)
    agent.watch()
    agent._schedule()
    assert item(world).status == "closed"


def test_an_unsaid_final_closes_the_issue(world: dict[str, Any]) -> None:
    forge: Forge = world["forge"]
    model = ScriptedModel(
        RESTATE,
        [
            call("start_milestone", name="a"),
            call("write_file", path="a.py", content="x = 1\n"),
            call("finish_milestone", summary="adds a"),
        ],
    )
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    [pr] = forge.prs
    assert "Closes #14" in forge.prs[pr]["body"]