nisse/nisse-dev/tests/test_loop_live_fixes.py

"""What the first live session of the new loop found (2026-10-07), each as a
test: a request ending on the assistant's side (refused by models without
prefill: the driver's prose nudge, a helper's first request), `status` in an
item's thread taken as a reply, no way for the driver to answer a remark;
later, near-miss commands, the driver claiming a stop, a reassigned issue
ignored, an unexplained failed abandon and restating an agreed item.
"""

from pathlib import Path
from typing import Any

from model_contract import assert_renderable
from test_loop import DON, RESTATED, FakeGitea, Harness, ScriptedModel, call, issue
from test_loop_help import ROLES, hand_back
from test_loop_milestones import RESTATE, Forge, loop_agent, 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.canonical import AssistantText, Fact
from nisse_dev.delegation import Caps
from nisse_dev.gitea import Comment
from nisse_dev.loop.help import Helpers
from nisse_dev.loop.items import GO_AHEAD, MERGE, WAITING, loop_of


class Clock:
    def __init__(self) -> None:
        self.now = 1000.0

    def __call__(self) -> float:
        return self.now


def test_the_prose_nudge_keeps_requests_ending_on_the_user_side(
    tmp_path: Path,
) -> None:
    # ScriptedModel checks every request as the API would, now including
    # "the conversation must end with a user message".
    h = Harness(
        tmp_path,
        ScriptedModel([AssistantText(text="Let me think.", model="m")], RESTATED),
        FakeGitea(issue()),
    )
    h.agent.intake()
    assert loop_of(h.item()).wait == GO_AHEAD  # the nudge worked
    facts = [
        e.event.text
        for e in h.store.transcript(h.item().id, "orchestrator").read()
        if isinstance(e.event, Fact)
    ]
    assert any("Your last reply was prose" in f for f in facts)


class ContractHelper:
    """A helper that checks its requests as the API would."""

    def __init__(self, *turns: list[Any]) -> None:
        self.turns = list(turns)

    def complete(self, events, tools, model, system=None) -> list[Any]:
        assert_renderable(events)
        return self.turns.pop(0)


def test_a_helpers_requests_end_on_the_user_side(world: dict[str, Any]) -> None:
    helper = ContractHelper(
        [AssistantText(text="Looking around first.", model="h")],  # prose
        [call("write_file", path="helped.py", content="H = 1\n")],
        [hand_back()],
    )
    model = ScriptedModel(
        RESTATE,
        [
            call("start_milestone", name="words"),
            call("delegate", task="write helped.py"),
            call("finish_milestone", summary="with help", final="yes"),
        ],
    )
    helpers = Helpers(
        client=helper,  # type: ignore[arg-type]
        roles=ROLES,
        client_for=lambda model: helper,  # type: ignore[arg-type,return-value]
        caps=Caps(turns=5),
    )
    agent, transport = loop_agent(world, model, helpers=helpers)
    agent.intake()
    say(transport, world, "go")
    forge: Forge = world["forge"]
    [pr] = forge.prs
    assert forge.files(pr) == ["helped.py"]


def test_status_in_a_thread_is_not_a_reply(tmp_path: Path) -> None:
    clock = Clock()
    model = ScriptedModel(RESTATED)  # called once only: status runs nothing
    h = Harness(
        tmp_path,
        model,
        FakeGitea(issue(labels=("bug",))),
        clock=clock,
        bug_grace_s=600,
    )
    h.agent.intake()
    grace = loop_of(h.item()).grace_until
    h.say("status")
    assert h.posted()[-1].startswith("#14 Add count_words: waiting for go-ahead")
    loop = loop_of(h.item())
    assert (h.item().position, loop.wait, loop.grace_until) == (
        WAITING,
        GO_AHEAD,
        grace,
    )
    h.say("details")
    assert loop_of(h.item()).grace_until == grace


def test_a_remark_gets_a_reply_and_the_restatement_stays_up(tmp_path: Path) -> None:
    h = Harness(
        tmp_path,
        ScriptedModel(
            RESTATED,
            [call("reply", text="Noted: I'll keep it to text.py.")],
        ),
        FakeGitea(issue()),
    )
    h.agent.intake()
    h.say("just a thought: keep it small")
    loop = loop_of(h.item())
    assert (loop.wait, loop.rev) == (GO_AHEAD, 1)
    assert h.posted()[-1] == "Noted: I'll keep it to text.py."


def test_a_review_comment_can_be_answered_without_a_fix(
    world: dict[str, Any],
) -> None:
    forge: Forge = world["forge"]
    model = ScriptedModel(
        RESTATE,
        [
            call("start_milestone", name="words"),
            call("write_file", path="w.py", content="W = 1\n"),
            call("finish_milestone", summary="words", final="yes"),
        ],
        [call("reply", text="It's a constant on purpose, for the test.")],
    )
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    [pr] = forge.prs
    forge.comments[pr] = [Comment(7, "don", "why a constant", "t")]
    agent.watch()
    agent._schedule()
    loop = loop_of(the_item(world))
    assert (loop.wait, loop.pr) == (MERGE, pr)
    assert forge.posted[-1][0] == pr  # answered on the PR, where asked
    assert "It's a constant on purpose" in forge.posted[-1][1]


# --- the session's later findings ----------------------------------------------


def test_a_near_miss_for_abandon_is_asked_about(tmp_path: Path) -> None:
    model = ScriptedModel(RESTATED)  # called once only: nothing reaches it
    h = Harness(tmp_path, model, FakeGitea(issue()))
    h.agent.intake()
    h.say("abort")
    assert h.posted()[-1].startswith("Did you mean `abandon`?")
    assert (h.item().status, loop_of(h.item()).wait) == ("open", GO_AHEAD)
    h.say("abandon")
    assert h.item().status == "closed"


def test_the_driver_is_told_it_cant_stop_or_close(tmp_path: Path) -> None:
    seen: list[str] = []

    class Spy(ScriptedModel):
        def complete(self, events, tools, model, system=None):  # type: ignore[no-untyped-def]
            seen.append(system or "")
            return super().complete(events, tools, model, system)

    h = Harness(tmp_path, Spy(RESTATED), FakeGitea(issue()))
    h.agent.intake()
    assert "never say you have stopped or closed anything yourself" in seen[0]


def test_an_issue_assigned_again_after_its_item_closed_is_taken_again(
    tmp_path: Path,
) -> None:
    gitea = FakeGitea(issue())
    h = Harness(tmp_path, ScriptedModel(RESTATED, RESTATED), gitea)
    h.agent.intake()
    h.say("abandon")
    [old] = h.store.work_items()
    gitea.closed.remove(14)  # the operator reopened it
    h.agent.intake()
    assert len(h.store.work_items()) == 1  # the same assignment: not again
    gitea.reassigned[14] = "2999-01-01T00:00:00Z"  # assigned anew, later
    h.agent.intake()
    items = h.store.work_items()
    assert len(items) == 2
    new = next(i for i in items if i.id != old.id)
    assert new.status == "open" and loop_of(new).follows == old.id


def test_abandoning_a_blocked_issue_says_why_it_stays_open(tmp_path: Path) -> None:
    from nisse_dev.gitea import GiteaError

    class Strict(FakeGitea):
        def close_issue(self, repo: str, index: int) -> None:
            if self.open_dependencies(repo, index):
                raise GiteaError("gitea PATCH returned 500: blocked")
            super().close_issue(repo, index)

    gitea = Strict(issue(14), issue(15, title="other"))
    gitea.deps[14] = [15]
    h = Harness(tmp_path, ScriptedModel(RESTATED), gitea)
    h.agent._running = True  # hold the queue: only the abandon matters
    h.agent.intake()
    item = next(i for i in h.store.work_items() if loop_of(i).issue == 14)
    h.transport.receive(DON, "abandon", thread=loop_of(item).thread)
    said = " ".join(h.posted())
    assert "won't close an issue while it depends on open ones" in said
    assert "don/sandbox#15" in said


def test_an_agreed_item_isnt_offered_restate(world: dict[str, Any]) -> None:
    offered: list[set[str]] = []

    class Spy(ScriptedModel):
        def complete(self, events, tools, model, system=None):  # type: ignore[no-untyped-def]
            offered.append({t["function"]["name"] for t in tools})
            return super().complete(events, tools, model, system)

    model = Spy(RESTATE, [call("ask_human", question="anything else?")])
    agent, transport = loop_agent(world, model)
    agent.intake()
    say(transport, world, "go")
    assert "restate" in offered[0] and "restate" not in offered[1]