nisse/nisse-dev/tests/test_loop_help.py

"""Help for the driver on the new loop (M2.6, L9): roles pinned per item,
delegation on a branch off the milestone's (merged back on a handback,
discarded otherwise), the stuck trigger, ask_expert and its budget, and
briefings from capability profiles. Real git on the throwaway build host."""

import json
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_core.log.canonical import (
    AssistantText,
    DelegationStarted,
    Fact,
    ToolCall,
    ToolResult,
)
from nisse_core.models.llm import RoleMap
from nisse_dev.config import Profile
from nisse_dev.delegation import DELEGATION_INTENT, Caps
from nisse_dev.loop.help import Helpers
from nisse_dev.loop.items import ANSWER, MERGE, loop_of

ROLES = RoleMap(driver="drv", helper="hlp", expert="exp")


class Spy:
    """A scripted model that records what each call was told."""

    def __init__(self, *turns: list[Any]) -> None:
        self.turns = list(turns)
        self.calls: list[tuple[str, str]] = []  # (model, system)

    def complete(self, events, tools, model, system=None) -> list[Any]:
        self.calls.append((model, system or ""))
        assert self.turns, f"{model} was called more often than scripted"
        return self.turns.pop(0)


def hand_back(succeeded: bool = True) -> ToolCall:
    return ToolCall(
        call_id=f"hb-{succeeded}",
        tool="hand_back",
        arguments={
            "changed": "wrote helped.py",
            "now_true": "the helper's part is done",
            "next": "finish the milestone",
            "succeeded": succeeded,
        },
    )


def helpers(helper: Spy, expert: Spy | None = None, **kw: Any) -> Helpers:
    return Helpers(
        client=expert or Spy(),  # type: ignore[arg-type]
        roles=ROLES,
        client_for=lambda model: helper,  # type: ignore[arg-type,return-value]
        caps=Caps(turns=5),
        profiles={
            "hlp": Profile(tools="reliable", strengths="tricky tests"),
            "drv": Profile(tools="fair", weaknesses="long refactors"),
        },
        **kw,
    )


def driver_results(world: dict[str, Any], tool: str) -> list[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
    }
    return [
        e.event.result
        for e in events
        if isinstance(e.event, ToolResult) and e.event.call_id in ids
    ]


def local_branches(world: dict[str, Any]) -> list[str]:
    return world["repo"].local_branches()


def test_a_helpers_work_lands_on_the_milestone(world: dict[str, Any]) -> None:
    forge: Forge = world["forge"]
    helper = Spy(
        [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 with H = 1"),
            call("finish_milestone", summary="with help", final="yes"),
        ],
    )
    agent, transport = loop_agent(world, model, helpers=helpers(helper))
    agent.intake()
    say(transport, world, "go")

    [result] = driver_results(world, "delegate")
    assert result.startswith("the helper finished:") and "helped.py" in result
    assert helper.calls[0][0] == "hlp"  # the helper role's model
    # Briefed about the driver by its profile, never by name.
    assert "tool calls: fair; weak at long refactors" in helper.calls[0][1]
    [pr] = forge.prs
    assert forge.files(pr) == ["helped.py"]
    assert not [b for b in local_branches(world) if "-help-" in b]
    assert loop_of(item(world)).wait == MERGE


def test_a_delegation_records_what_the_alias_resolved_to(
    world: dict[str, Any],
) -> None:
    # L9 eval history keyed on an alias alone would average across whatever
    # that alias pointed at over time.
    helper = Spy(
        [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"),
        ],
    )
    backed = helpers(helper, backing=lambda alias: f"anthropic/{alias}-backing")
    agent, transport = loop_agent(world, model, helpers=backed)
    agent.intake()
    say(transport, world, "go")
    [started] = [
        e.event
        for e in world["store"].transcript(item(world).id, "helper").read()
        if isinstance(e.event, DelegationStarted)
    ]
    assert started.backing_model == "anthropic/hlp-backing"


def test_a_failed_helpers_work_is_discarded(world: dict[str, Any]) -> None:
    helper = Spy(
        [call("write_file", path="half.py", content="half\n")],
        [hand_back(succeeded=False)],
    )
    model = ScriptedModel(
        RESTATE,
        [
            call("start_milestone", name="words"),
            call("write_file", path="mine.py", content="M = 1\n"),
            call("delegate", task="write half.py"),
            call("ask_human", question="the helper couldn't; what now?"),
        ],
    )
    agent, transport = loop_agent(world, model, helpers=helpers(helper))
    agent.intake()
    say(transport, world, "go")

    [result] = driver_results(world, "delegate")
    assert "didn't finish" in result and "discarded" in result
    root = world["repo"].root
    assert not (root / "half.py").exists()
    assert (root / "mine.py").exists()  # the driver's own work stays
    assert world["repo"].current_branch() == loop_of(item(world)).branch
    assert not [b for b in local_branches(world) if "-help-" in b]
    assert not [
        i for i in world["store"].pending_intents() if i.kind == DELEGATION_INTENT
    ]


def test_delegating_needs_a_milestone(world: dict[str, Any]) -> None:
    model = ScriptedModel(
        RESTATE,
        [
            call("delegate", task="anything"),
            call("ask_human", question="ok?"),
        ],
    )
    agent, transport = loop_agent(world, model, helpers=helpers(Spy()))
    agent.intake()
    say(transport, world, "go")
    assert driver_results(world, "delegate") == [
        "start a milestone first: a helper works on its branch"
    ]


def test_a_stuck_driver_gets_a_helper_once_per_milestone(
    world: dict[str, Any],
) -> None:
    forge: Forge = world["forge"]
    helper = Spy(
        [call("write_file", path="fixed.py", content="F = 1\n")],
        [hand_back()],
    )
    failing = [
        call("run_command", command="false"),
        call("run_command", command="false"),
    ]
    model = ScriptedModel(
        RESTATE,
        [call("start_milestone", name="words"), *failing],
        [call("finish_milestone", summary="fixed with help", final="yes")],
    )
    agent, transport = loop_agent(world, model, helpers=helpers(helper))
    agent._driver._failure_limit = 2
    agent.intake()
    say(transport, world, "go")

    assert any("so a helper is taking over" in m.text for m in transport.outbound)
    assert "The driver got stuck" in str(helper.calls)  # in the brief
    [pr] = forge.prs
    assert forge.files(pr) == ["fixed.py"]
    assert loop_of(item(world)).helped_at == 1


def test_stuck_again_in_the_same_milestone_asks_the_operator(
    world: dict[str, Any],
) -> None:
    helper = Spy(
        [call("write_file", path="tried.py", content="T = 1\n")],
        [hand_back()],
    )
    failing = [
        call("run_command", command="false"),
        call("run_command", command="false"),
    ]
    model = ScriptedModel(
        RESTATE,
        [call("start_milestone", name="words"), *failing],
        failing,  # after the helper: stuck again
    )
    agent, transport = loop_agent(world, model, helpers=helpers(helper))
    agent._driver._failure_limit = 2
    agent.intake()
    say(transport, world, "go")
    loop = loop_of(item(world))
    assert loop.wait == ANSWER and "commands failed in a row" in loop.note
    assert len(helper.calls) == 2  # one delegation: its two turns


def test_ask_expert_answers_within_its_budget(world: dict[str, Any]) -> None:
    answer = {"answer": "use a dict", "confidence": "high", "caveats": ""}
    expert = Spy(
        [AssistantText(text=json.dumps(answer), model="exp")],
        [AssistantText(text=json.dumps(answer), model="exp")],
    )
    model = ScriptedModel(
        RESTATE,
        [
            call("start_milestone", name="words"),
            call("ask_expert", question="dict or list?"),
            call("ask_expert", question="and again?"),
            call("ask_expert", question="once more?"),
            call("ask_human", question="ok?"),
        ],
    )
    agent, transport = loop_agent(
        world, model, helpers=helpers(Spy(), expert, expert_budget=2)
    )
    agent.intake()
    say(transport, world, "go")
    first, second, third = driver_results(world, "ask_expert")
    assert first.startswith("use a dict") and second.startswith("use a dict")
    assert "budget" in third
    assert expert.calls[0][0] == "exp"


def test_roles_are_pinned_when_the_item_opens(world: dict[str, Any]) -> None:
    spy = Spy(RESTATE)
    agent, _ = loop_agent(world, spy, helpers=helpers(Spy()))  # type: ignore[arg-type]
    agent._roles = {**ROLES.pinned(), "driver": "pinned-driver"}
    agent.intake()
    assert spy.calls[0][0] == "pinned-driver"
    pinned = item(world).data["roles"]
    assert pinned["driver"] == "pinned-driver" and pinned["classifier"] == "exp"


def test_the_driver_is_told_about_its_helpers_by_profile(
    world: dict[str, Any],
) -> None:
    spy = Spy(RESTATE, [call("ask_human", question="ok?")])
    agent, transport = loop_agent(world, spy, helpers=helpers(Spy()))  # type: ignore[arg-type]
    agent.intake()
    say(transport, world, "go")
    after_go = spy.calls[1][1]
    assert "Asking for help is normal" in after_go
    assert "tool calls: reliable; strong at tricky tests" in after_go
    assert "hlp" not in after_go  # never a model's name


def test_a_delegation_cut_off_by_a_crash_is_discarded_at_boot(
    world: dict[str, Any],
) -> None:
    model = ScriptedModel(
        RESTATE,
        [call("start_milestone", name="words"), call("ask_human", question="?")],
    )
    agent, transport = loop_agent(world, model, helpers=helpers(Spy()))
    agent.intake()
    say(transport, world, "go")
    milestone = loop_of(item(world)).branch
    repo = world["repo"]
    repo.checkout_new_branch(f"{milestone}-help-1")
    (repo.root / "partial.py").write_text("p\n")
    repo.commit_all("partial")
    world["store"].begin_intent(
        DELEGATION_INTENT,
        {"branch": f"{milestone}-help-1", "base": milestone},
        cause="delegating",
        work_item=item(world).id,
    )
    agent2, _ = loop_agent(world, ScriptedModel(), helpers=helpers(Spy()))
    agent2.boot()
    assert repo.current_branch() == milestone
    assert not (repo.root / "partial.py").exists()
    assert f"{milestone}-help-1" not in repo.local_branches()
    assert not [
        i for i in world["store"].pending_intents() if i.kind == DELEGATION_INTENT
    ]
    facts = [
        e.event.text
        for e in world["store"].transcript(item(world).id, "orchestrator").read()
        if isinstance(e.event, Fact)
    ]
    assert any("cut off by a restart and discarded" in f for f in facts)
    assert git


def test_a_helpers_progress_is_posted(world: dict[str, Any]) -> None:
    helper = Spy(
        [
            AssistantText(text="Writing helped.py with the constant.", model="hlp"),
            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("ask_human", question="done?"),
        ],
    )
    agent, transport = loop_agent(world, model, helpers=helpers(helper))
    agent.intake()
    say(transport, world, "go")
    said = [m.text for m in transport.outbound]
    assert "helper: Writing helped.py with the constant." in said