nisse/nisse-dev/tests/test_loop_talk.py

"""Conversation on the new loop (M2.7, L5, #42, #77): questions answered
without changing the item, revisions that say what changed, progress in
prose at a readable rate, raw tool activity on request, and a note for a
message that can't be read yet."""

from pathlib import Path
from typing import Any

from test_loop import DON, RESTATED, FakeGitea, Harness, ScriptedModel, call, issue

from nisse_core.canonical import AssistantText
from nisse_dev.loop.items import ANSWER, GO_AHEAD, WAITING, loop_of


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

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


def harness(tmp_path: Path, *turns: list[Any], **kw: Any) -> Harness:
    return Harness(tmp_path, ScriptedModel(*turns), FakeGitea(issue()), **kw)


# --- questions never change state (#77) -------------------------------------


def test_a_question_about_a_restatement_leaves_it_up(tmp_path: Path) -> None:
    h = harness(
        tmp_path,
        RESTATED,
        [call("reply", text="Because text.py already holds the string helpers.")],
        [call("ask_human", question="started; anything else?")],
    )
    h.agent.intake()
    h.say("Why text.py?")
    loop = loop_of(h.item())
    assert (h.item().position, loop.wait, loop.rev) == (WAITING, GO_AHEAD, 1)
    assert not loop.question and not loop.started
    assert h.posted()[-1] == "Because text.py already holds the string helpers."
    h.say("go")  # still bound to rev-1
    assert loop_of(h.item()).started


def test_a_question_back_keeps_the_agents_own_question_open(tmp_path: Path) -> None:
    h = harness(
        tmp_path,
        [call("ask_human", question="One function or two?")],
        [call("reply", text="One handles words, two would split lines too.")],
    )
    h.agent.intake()
    h.say("What would the second one do?")
    loop = loop_of(h.item())
    assert (loop.wait, loop.note) == (ANSWER, "One function or two?")


def test_a_question_that_isnt_answered_still_goes_back(tmp_path: Path) -> None:
    h = harness(
        tmp_path,
        RESTATED,
        [call("restate", text="sneaky v2", changed="x")],  # not offered
        [AssistantText(text="hmm", model="m")],
        [AssistantText(text="hmm", model="m")],  # prose twice: the run fails
    )
    h.agent.intake()
    h.say("why?")
    loop = loop_of(h.item())
    # restate isn't offered in a question run: no new revision.
    assert (loop.wait, loop.rev) == (GO_AHEAD, 1)
    assert h.posted()[-1].startswith("I couldn't answer that")


# --- explanations ------------------------------------------------------------


def test_a_revision_says_what_changed(tmp_path: Path) -> None:
    h = harness(
        tmp_path,
        RESTATED,
        [call("restate", text="count_words in words.py", changed="")],
        [
            call(
                "restate",
                text="count_words in words.py",
                changed="a module of its own, as you asked",
            )
        ],
    )
    h.agent.intake()
    h.say("put it in its own module")
    loop = loop_of(h.item())
    assert loop.rev == 2
    assert "Changed from rev-1: a module of its own, as you asked" in h.posted()[-1]


# --- progress and details (#42) ----------------------------------------------


def test_progress_is_posted_at_a_readable_rate(tmp_path: Path) -> None:
    clock = Clock()

    class Ticking(ScriptedModel):
        def complete(self, events, tools, model, system=None):  # type: ignore[no-untyped-def]
            clock.now += 30  # each turn takes a while
            return super().complete(events, tools, model, system)

    h = Harness(
        tmp_path,
        Ticking(
            [
                AssistantText(text="Looking at text.py first.", model="m"),
                call("list_dir", path="."),
            ],
            [
                AssistantText(text="Reading the tests too.", model="m"),
                call("list_dir", path="tests"),
            ],
            [
                AssistantText(text="x" * 1000, model="m"),
                call("list_dir", path="."),
            ],
            [call("restate", text="I'll add count_words.", changed="")],
        ),
        FakeGitea(issue()),
        clock=clock,
    )
    h.agent.intake()
    said = h.posted()
    assert "Looking at text.py first." in said
    assert "Reading the tests too." not in said  # 30 s later: folded away
    long = [p for p in said if p.startswith("xxx")]
    assert long and len(long[0]) < 420 and long[0].endswith(" …")  # 60 s on


def test_details_shows_the_last_tool_calls(tmp_path: Path) -> None:
    h = harness(
        tmp_path,
        [
            call("list_dir", path="src"),
            call("read_file", path="text.py"),
            call("restate", text="I'll add count_words.", changed=""),
        ],
    )
    h.agent.intake()
    h.say("details")
    shown = h.posted()[-1]
    assert shown.startswith("The last tool calls on #14:")
    assert "list_dir(src)" in shown and "read_file(text.py)" in shown
    assert loop_of(h.item()).wait == GO_AHEAD  # nothing changed


def test_a_message_mid_run_is_acknowledged(tmp_path: Path) -> None:
    h = harness(tmp_path, RESTATED)
    h.agent.intake()
    h.agent._running = True  # as if a driver run were under way
    h.agent._active_thread = "elsewhere"
    h.agent._active_name = "#15"
    h.transport.receive(DON, "one more thing", thread=h.thread())
    assert "Noted; I'm in the middle of #15" in h.posted()[-1]
    h.transport.receive(DON, "and another", thread="elsewhere")
    assert h.posted()[-1] == "Noted; I'll read it when this turn ends."