nisse/nisse-dev/tests/test_loop_memory.py

"""Agent memory in the loop (L6.4, M5.2): the model remembers and forgets
without asking, each said in the thread; the index is shown once per item;
a bounded pass at an item's close records what will matter, and never holds
the close up."""

from pathlib import Path
from typing import Any

import pytest
from test_loop_milestones import DON, RESTATE, call, 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 test_loop_notes import Spy

from nisse_dev.canonical import AssistantText, Fact
from nisse_dev.llm import RoleMap
from nisse_dev.memory import Memory
from nisse_dev.vault import Vault

FOLDER = "memory/nisse-dev"


@pytest.fixture
def memory(tmp_path: Path) -> Memory:
    return Memory(Vault(tmp_path / "v", {FOLDER: "rw"}), FOLDER)


def facts(world: dict[str, Any]) -> list[str]:
    events = world["store"].transcript(the_item(world).id, "orchestrator").read()
    return [e.event.text for e in events if isinstance(e.event, Fact)]


def said(transport: Any) -> list[str]:
    return [m.text for m in transport.outbound]


def agent(world: dict[str, Any], model: Spy, memory: Memory, **more: Any) -> Any:
    return loop_agent(
        world,
        model,
        vault=memory._vault,
        memory=memory,
        roles=RoleMap(driver="m", helper="m", expert="m").pinned(),
        **more,
    )


def test_the_model_remembers_and_forgets_without_asking(
    world: dict[str, Any], memory: Memory
) -> None:
    model = Spy(
        [
            call(
                "remember",
                name="uses-split",
                description="Count words with split",
                fact="The operator wants str.split, no regex.",
            ),
            call(
                "remember",
                name="old-style",
                description="Old style",
                fact="Tabs.",
            ),
            call("forget", name="old-style", why="the operator changed it"),
        ],
        RESTATE,
    )
    loop, transport = agent(world, model, memory)
    loop.intake()
    assert {"remember", "forget"} <= model.offered[0]
    assert "Remembered: Count words with split" in said(transport)
    assert "Forgot: old-style (the operator changed it)" in said(transport)
    index = memory.snapshot().text
    assert "[[uses-split]]" in index and "old-style" not in index


def test_the_index_is_shown_once_per_item(
    world: dict[str, Any], memory: Memory
) -> None:
    memory.remember("uses-split", "Count words with split", "f", provenance={})
    model = Spy(RESTATE, [call("ask_human", question="tabs or spaces?")])
    loop, transport = agent(world, model, memory)
    loop.intake()
    say(transport, world, "go")
    shown = [f for f in facts(world) if f.startswith("your memory's index")]
    assert len(shown) == 1
    assert f"sha {memory.snapshot().sha}" in shown[0]
    assert "[[uses-split]] — Count words with split" in shown[0]


def test_a_pass_at_the_close_records_what_will_matter(
    world: dict[str, Any], memory: Memory
) -> None:
    model = Spy(
        RESTATE,
        [
            call(
                "remember",
                name="no-count-words",
                description="The operator dropped count_words",
                fact="Abandoned: not needed after all.",
            ),
            call("memory_done", summary="one note"),
        ],
    )
    loop, transport = agent(world, model, memory)
    loop.intake()
    say(transport, world, "abandon")
    assert the_item(world).status == "closed"
    assert "memory pass: recording what this item decided and learned" in facts(world)
    assert "[[no-count-words]]" in memory.snapshot().text
    offered = model.offered[-1]
    assert offered == {
        "read_note",
        "list_notes",
        "search_notes",
        "remember",
        "forget",
        "memory_done",
    }
    assert "which abandoned" in model.systems[-1]
    # Its own transcript: the driver's never holds it.
    log = world["store"].transcript(the_item(world).id, "driver").read()
    remembered = {e.actor for e in log if getattr(e.event, "tool", "") == "remember"}
    assert remembered == {"memory"}


def test_a_pass_that_doesnt_finish_never_holds_up_the_close(
    world: dict[str, Any], memory: Memory
) -> None:
    prose = [AssistantText(text="hmm", model="m")]
    model = Spy(RESTATE, prose, prose)  # prose twice: the run fails
    loop, transport = agent(world, model, memory)
    loop.intake()
    say(transport, world, "abandon")
    assert the_item(world).status == "closed"
    assert any(f.startswith("memory pass didn't finish") for f in facts(world))


def test_the_pass_is_bounded(world: dict[str, Any], memory: Memory) -> None:
    looking = [call("list_notes", folder="")]
    model = Spy(RESTATE, *[looking] * 3)
    loop, transport = agent(world, model, memory, memory_turns=3)
    loop.intake()
    say(transport, world, "abandon")
    assert any("hit the 3-turn cap" in f for f in facts(world))
    assert not model.turns


def test_a_conversation_can_remember(world: dict[str, Any], memory: Memory) -> None:
    world["forge"].issues.clear()
    model = Spy(
        [call("remember", name="has-4090", description="Has a 4090", fact="f")],
        [call("reply", text="Noted.")],
    )
    _, transport = agent(world, model, memory)
    transport.receive(DON, "I have a 4090 for local models.")
    assert "[[has-4090]]" in memory.snapshot().text
    assert "Remembered: Has a 4090" in said(transport)