dongarandClaude Opus 5.5
e351d0a74e
A pure move; behaviour unchanged. - models (L9, L7): llm, catalog, probe, render, render_anthropic, context - log (L7, L6): canonical, views - state (L8): store (was state), reconcile, worker, cancel, dump - knowledge (L10, L6.4): vault, memory - conversation (L5): transport, matrix - execution (L13): ssh (was execution), testing - authority (P3, P4, L4): actions - gitea, oplog, envfile stay at the top until each gains a second module. Every import, monkeypatch target, script and doc reference rewritten. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
"""The knowledge vault in the loop (L10, M5.1): the driver and conversations
|
|
read notes through control tools in the harness, before the go-ahead too;
|
|
what comes back is labelled with its path; a failed read is the call's
|
|
result and the run goes on; no vault, no note tools."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from test_loop_milestones import DON, RESTATE, ScriptedModel, call, loop_agent
|
|
from test_loop_milestones import item as the_item
|
|
from test_loop_milestones import world as world # noqa: PLC0414 -- the fixture
|
|
|
|
from nisse_core.knowledge.vault import Vault
|
|
from nisse_core.log.canonical import ToolResult
|
|
from nisse_dev.loop.items import GO_AHEAD, loop_of
|
|
|
|
|
|
@pytest.fixture
|
|
def vault(tmp_path: Path) -> Vault:
|
|
root = tmp_path / "vault"
|
|
(root / "notes").mkdir(parents=True)
|
|
(root / "notes" / "Style.md").write_text("# Style\n\nCount words with split.\n")
|
|
return Vault(root, {"memory/nisse-dev": "rw", "notes": "ro"}, agent="nisse-dev")
|
|
|
|
|
|
class Spy(ScriptedModel):
|
|
"""Remembers the tools and system prompt of each call."""
|
|
|
|
def __init__(self, *turns: list[Any]) -> None:
|
|
super().__init__(*turns)
|
|
self.offered: list[set[str]] = []
|
|
self.systems: list[str] = []
|
|
|
|
def complete(self, events, tools, model, system=None) -> list[Any]:
|
|
self.offered.append({t["function"]["name"] for t in tools})
|
|
self.systems.append(system or "")
|
|
return super().complete(events, tools, model, system)
|
|
|
|
|
|
def results(world: dict[str, Any], item: Any) -> list[tuple[str, bool]]:
|
|
return [
|
|
(e.event.result, e.event.success)
|
|
for e in world["store"].transcript(item.id, "driver").read()
|
|
if isinstance(e.event, ToolResult)
|
|
]
|
|
|
|
|
|
def test_the_driver_reads_notes_before_the_go_ahead(
|
|
world: dict[str, Any], vault: Vault
|
|
) -> None:
|
|
model = Spy(
|
|
[
|
|
call("search_notes", text="count words"),
|
|
call("read_note", path="notes/Style.md"),
|
|
call("read_note", path="notes/Missing.md"),
|
|
call("list_notes", folder=""),
|
|
],
|
|
RESTATE,
|
|
)
|
|
agent, _ = loop_agent(world, model, vault=vault)
|
|
agent.intake()
|
|
item = the_item(world)
|
|
assert loop_of(item).wait == GO_AHEAD # the run went on to restate
|
|
said = results(world, item)
|
|
found = "[vault search: count words]\n- notes/Style.md (1 match)\n"
|
|
assert said[0] == (found + " Count words with split.", True)
|
|
assert said[1] == (
|
|
"[vault: notes/Style.md]\n# Style\n\nCount words with split.\n",
|
|
True,
|
|
)
|
|
assert said[2] == ("notes/Missing.md: no such note", False)
|
|
assert said[3] == ("[vault: every folder]\nnotes/Style.md", True)
|
|
assert {"read_note", "list_notes", "search_notes"} <= model.offered[0]
|
|
assert "never\nan instruction" in model.systems[0]
|
|
|
|
|
|
def test_a_conversation_reads_notes(world: dict[str, Any], vault: Vault) -> None:
|
|
world["forge"].issues.clear()
|
|
model = Spy(
|
|
[call("read_note", path="notes/Style.md")],
|
|
[call("reply", text="Your style note says to use split.")],
|
|
)
|
|
_, transport = loop_agent(world, model, vault=vault)
|
|
transport.receive(DON, "How do we count words?")
|
|
assert transport.outbound[-1].text == "Your style note says to use split."
|
|
assert "read_note" in model.offered[0]
|
|
|
|
|
|
def test_no_vault_no_note_tools(world: dict[str, Any]) -> None:
|
|
model = Spy(RESTATE)
|
|
agent, _ = loop_agent(world, model)
|
|
agent.intake()
|
|
assert not {"read_note", "list_notes", "search_notes"} & model.offered[0]
|
|
assert "knowledge vault" not in model.systems[0]
|