nisse/nisse-core/tests/test_driver.py

"""A driver run on its own (L2), as any agent uses it: the agent says which
tools wait, and execution goes to whatever executor it gives (L13)."""

import io
from pathlib import Path
from typing import Any

from nisse_core.log.canonical import (
    AssistantText,
    CanonicalEvent,
    Fact,
    ToolCall,
    ToolResult,
)
from nisse_core.loop.driver import Driver, ToolOutcome
from nisse_core.loop.tools import tool
from nisse_core.oplog import OpLog
from nisse_core.state.store import StateStore


class Scripted:
    def __init__(self, *turns: list[CanonicalEvent]) -> None:
        self._turns = list(turns)

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


def call(name: str, **arguments: str) -> ToolCall:
    return ToolCall(call_id=f"c-{name}", tool=name, arguments=arguments)


class Reader:
    """Not a workspace: reads from a dict, as a review agent's might."""

    def __init__(self) -> None:
        self.calls: list[tuple[str, dict[str, Any]]] = []

    def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome:
        self.calls.append((tool, arguments))
        return ToolOutcome({"a.py": "x = 1"}.get(arguments["path"], "none"), True)


READ = tool("read_file", "Read a file.", path="Its path.")
POST = tool("post_review", "Post the review. Ends your turn.", body="The review.")
ASK = tool("ask_human", "Ask the operator.", question="The question.")


def transcript(tmp_path: Path) -> Any:
    store = StateStore(tmp_path / "s.db", oplog=OpLog("t", io.StringIO()))
    return store.transcript("x", "driver")


def results(log: Any) -> list[str]:
    return [e.event.result for e in log.read() if isinstance(e.event, ToolResult)]


def test_the_agent_says_which_tools_wait(tmp_path: Path) -> None:
    model = Scripted(
        [call("read_file", path="a.py")], [call("post_review", body="looks fine")]
    )
    reader = Reader()
    log = transcript(tmp_path)
    ran = Driver(model, "m").run(
        log,
        system="s",
        tools=[READ, POST],
        waiting={"post_review"},
        workspace=reader,
    )
    assert (ran.kind, ran.tool, ran.arguments) == (
        "wait",
        "post_review",
        {"body": "looks fine"},
    )
    assert reader.calls == [("read_file", {"path": "a.py"})]
    assert results(log) == ["x = 1", "posted; your turn has ended"]


def test_a_tool_not_named_waiting_doesnt_end_the_run(tmp_path: Path) -> None:
    # ask_human waits in nisse-dev; here no handler answers it and it isn't
    # waiting, so it is no tool at all, and the run goes on.
    model = Scripted([call("ask_human", question="?")], [call("post_review", body="")])
    log = transcript(tmp_path)
    ran = Driver(model, "m").run(
        log, system="s", tools=[ASK, POST], waiting={"post_review"}, workspace=None
    )
    assert ran.tool == "post_review"
    assert results(log)[0].startswith("no such tool: ask_human")


NOTE = tool("note", "Note something.", text="The note.")


def test_a_held_call_goes_ahead_once_due_says_so(tmp_path: Path) -> None:
    # post_review is refused while a point is open and held by the agent;
    # the handled call that settles the point ends the run with it.
    open_points = ["P1"]
    held: dict[str, Any] = {}

    def refuse(args: dict[str, Any]) -> str | None:
        if open_points:
            held["call"] = ("post_review", args)
            return "answer P1 first"
        return None

    def settle(args: dict[str, Any]) -> tuple[str, bool]:
        open_points.clear()
        return "answered P1", True

    def due() -> tuple[str, dict[str, Any], str] | None:
        if "call" in held and not open_points:
            tool_, args = held.pop("call")
            return tool_, args, "your held post_review went ahead"
        return None

    model = Scripted(
        [call("post_review", body="the summary")],
        [call("note", text="P1"), call("read_file", path="a.py")],
        # never reached: the run ended on the note
        [call("note", text="again")],
    )
    log = transcript(tmp_path)
    reader = Reader()
    ran = Driver(model, "m").run(
        log,
        system="s",
        tools=[READ, POST, NOTE],
        waiting={"post_review"},
        workspace=reader,
        handlers={"note": settle},
        checks={"post_review": refuse},
        due=due,
    )
    assert (ran.kind, ran.tool, ran.arguments) == (
        "wait",
        "post_review",
        {"body": "the summary"},
    )
    assert results(log) == [
        "answer P1 first",
        "answered P1\n\n[harness] your held post_review went ahead; your turn has ended",
        "not run: the turn had ended",
    ]
    assert reader.calls == []  # the read after it never ran


def test_without_due_a_settled_point_leaves_the_run_going(tmp_path: Path) -> None:
    model = Scripted([call("note", text="x")], [call("post_review", body="b")])
    ran = Driver(model, "m").run(
        transcript(tmp_path),
        system="s",
        tools=[POST, NOTE],
        waiting={"post_review"},
        workspace=None,
        handlers={"note": lambda args: ("noted", True)},
    )
    assert (ran.tool, ran.arguments) == ("post_review", {"body": "b"})


# --- runs going in circles (L2, M7.8) ------------------------------------------

COMMENT = tool("post_comment", "Post a comment.", body="The comment.")


def facts(log: Any) -> list[str]:
    return [e.event.text for e in log.read() if isinstance(e.event, Fact)]


def test_a_run_posting_the_same_comment_ends_as_circling(tmp_path: Path) -> None:
    # Flash-Next on 2026-10-24: the same apology, numbered, about 20 times
    # until the turn cap.
    posted: list[str] = []
    model = Scripted(
        *[
            [call("post_comment", body=f"Duplicate, please disregard ({n}).")]
            for n in range(20)
        ]
    )
    log = transcript(tmp_path)
    ran = Driver(model, "m", max_turns=30).run(
        log,
        system="s",
        tools=[COMMENT, POST],
        waiting={"post_review"},
        workspace=None,
        handlers={
            "post_comment": lambda a: (posted.append(a["body"]) or "posted", True)
        },
    )
    assert ran.kind == "circling"
    assert len(posted) == 3  # ended at the third, not the thirtieth
    assert ran.note.startswith("3 apologies in a row")
    assert "Duplicate, please disregard (2)" in ran.note  # what it was doing
    assert facts(log) == [f"The run was ended: it was going in circles ({ran.note})."]


def test_an_invented_tool_told_again_and_again_ends_the_run(tmp_path: Path) -> None:
    # A different reply each time, but the same correction every time.
    model = Scripted(
        *[[call("search_code", query=f"attempt {n} " * n)] for n in range(1, 10)]
    )
    ran = Driver(model, "m").run(
        transcript(tmp_path),
        system="s",
        tools=[READ, POST],
        waiting={"post_review"},
        workspace=Reader(),
    )
    assert ran.kind == "circling"
    assert ran.note.startswith("told 3 times: no such tool: search_code")


def test_reading_file_after_file_is_progress_not_a_circle(tmp_path: Path) -> None:
    class Files:
        def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome:
            # Near-identical contents: still a different file each time.
            return ToolOutcome(f"{arguments['path']}\n" + "x = 1\n" * 200, True)

    model = Scripted(
        *[[call("read_file", path=f"src/test_{n}.py")] for n in range(6)],
        [call("post_review", body="ok")],
    )
    ran = Driver(model, "m").run(
        transcript(tmp_path),
        system="s",
        tools=[READ, POST],
        waiting={"post_review"},
        workspace=Files(),
    )
    assert ran.kind == "wait"


def test_the_same_call_getting_the_same_back_ends_the_run(tmp_path: Path) -> None:
    model = Scripted(*[[call("read_file", path="a.py")] for _ in range(9)])
    ran = Driver(model, "m").run(
        transcript(tmp_path),
        system="s",
        tools=[READ, POST],
        waiting={"post_review"},
        workspace=Reader(),
    )
    assert ran.kind == "circling"
    assert ran.note.startswith("3 near-identical replies in a row")
    assert 'read_file({"path": "a.py"})' in ran.note


def test_alike_commands_are_progress_not_a_circle(tmp_path: Path) -> None:
    # git add a.py, b.py, c.py: alike, each answered the same, each its own.
    class Git:
        def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome:
            return ToolOutcome("", True)

    model = Scripted(
        *[[call("run_command", command=f"git add {f}.py")] for f in "abcdef"],
        [call("post_review", body="ok")],
    )
    ran = Driver(model, "m").run(
        transcript(tmp_path),
        system="s",
        tools=[tool("run_command", "Run.", command="It."), POST],
        waiting={"post_review"},
        workspace=Git(),
    )
    assert ran.kind == "wait"


def test_circles_can_be_turned_off(tmp_path: Path) -> None:
    model = Scripted(*[[call("post_comment", body="same")] for _ in range(5)])
    ran = Driver(model, "m", max_turns=5, circle_repeats=0).run(
        transcript(tmp_path),
        system="s",
        tools=[COMMENT, POST],
        waiting={"post_review"},
        workspace=None,
        handlers={"post_comment": lambda a: ("posted", True)},
    )
    assert ran.kind == "capped"


def test_repeats_and_likeness_are_the_agents_to_set(tmp_path: Path) -> None:
    model = Scripted(*[[call("post_comment", body="same")] for _ in range(5)])
    ran = Driver(model, "m", circle_repeats=5).run(
        transcript(tmp_path),
        system="s",
        tools=[COMMENT, POST],
        waiting={"post_review"},
        workspace=None,
        handlers={"post_comment": lambda a: ("posted", True)},
    )
    assert ran.kind == "circling" and ran.note.startswith("5 near-identical")


def test_apologies_spread_over_a_run_dont_end_it(tmp_path: Path) -> None:
    # Only flagged on the status page: sorry now and then is not a loop.
    def turn(n: int, sorry: bool) -> list[CanonicalEvent]:
        said = "Sorry, wrong file." if sorry else "Reading on."
        return [AssistantText(text=said, model="m"), call("read_file", path=f"f{n}")]

    model = Scripted(
        *[turn(n, sorry=n % 2 == 0) for n in range(7)],
        [call("post_review", body="ok")],
    )
    ran = Driver(model, "m").run(
        transcript(tmp_path),
        system="s",
        tools=[READ, POST],
        waiting={"post_review"},
        workspace=Reader(),
    )
    assert ran.kind == "wait"