M5.2: agent memory

remember(name, description, fact) and forget(name, why): control tools for
the driver (before the go-ahead too) and conversations, writing only the
agent's own vault folder, one note per fact with its provenance. No one
agrees first; each change is a line in the thread. The harness keeps the
index and puts it into the transcript once per item, with a hash.

At an item's close, completed or abandoned, a bounded memory pass (the
driver's model, its own transcript and view) records what was decided and
learned; it never holds the close up (tuning.memory.max_turns).

With the operator: corrections may be paraphrased (L6 changed), the model
forgets on its own, and nothing limits what may prompt a note. The walk
remembers and forgets, and checks nothing lands outside the agent's folder.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk

nisse-dev/tests/test_loop_memory.py

@@ -0,0 +1,160 @@
+"""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)