NexusOS/tests/test_tools.py

"""Tool-using playbook loop — the read-only MVP.

Run from nexus-core/ with the Promethean venv active:  pytest -q

Hermetic: a fake manager stands in for Ollama, so no network/model is needed.
Guards the two pieces that would silently break the feature: the allowlist
filter and the tool-call loop's terminate-on-content behaviour.
"""
import asyncio

from synapse import tools
from synapse.chat import _run_tool_loop


def test_schemas_for_drops_unknown_names():
    schemas = tools.schemas_for(["search_memory", "not_a_tool"])
    names = [s["function"]["name"] for s in schemas]
    assert names == ["search_memory"]
    assert tools.schemas_for([]) == []


async def _drain(gen):
    return [s async for s in gen]


def test_remember_writes_a_memory_fact(tmp_path, monkeypatch):
    # The `remember` action tool persists a fact through the store.
    import synapse.memory.store as store_mod
    from synapse.memory.store import PersistentMemoryStore
    fresh = PersistentMemoryStore(tmp_path / "m.db")
    monkeypatch.setattr(tools, "store", fresh)
    out = asyncio.run(tools.dispatch("remember", {"text": "user likes tea", "section": "Prefs"}))
    assert "user likes tea" in out
    assert any(it.text == "user likes tea" for it in fresh.all())


def test_action_tools_registered():
    for name in ("web_search", "fetch_url", "remember"):
        assert name in tools.REGISTRY
    names = [s["function"]["name"] for s in tools.schemas_for(["web_search", "remember", "nope"])]
    assert names == ["web_search", "remember"]


class _ActionManager:
    """Returns a `remember` (action) tool_call once, then plain content."""
    def __init__(self):
        self.n = 0

    async def chat(self, **_):
        self.n += 1
        if self.n == 1:
            return {"role": "assistant",
                    "tool_calls": [{"function": {"name": "remember", "arguments": {"text": "x"}}}]}
        return {"role": "assistant", "content": "done"}


def _drive_with_decision(decision, monkeypatch):
    from synapse import chat as chatmod

    async def fake_dispatch(name, args):
        return "saved-ok"
    monkeypatch.setattr(tools, "dispatch", fake_dispatch)

    async def run():
        messages = [{"role": "user", "content": "remember x"}]
        gen = chatmod._run_tool_loop(_ActionManager(), messages, "m", [{}], None, None,
                                     conversation_id="conv", policy="ask")
        statuses = []
        async for s in gen:
            statuses.append(s)
            if s.startswith("__approve__"):
                w = chatmod.pending_approvals["conv"]
                w["decisions"] = {"remember": decision}
                w["event"].set()
        return statuses, messages

    return asyncio.run(run())


def test_ask_policy_pauses_then_runs_on_approve(monkeypatch):
    statuses, messages = _drive_with_decision(True, monkeypatch)
    assert any(s.startswith("__approve__") for s in statuses)   # paused for approval
    assert "__status__remember" in statuses                     # approved -> ran
    assert any(m["role"] == "tool" and "saved-ok" in m["content"] for m in messages)


def test_ask_policy_skips_on_deny(monkeypatch):
    statuses, messages = _drive_with_decision(False, monkeypatch)
    assert any(s.startswith("__approve__") for s in statuses)
    assert "__status__remember" not in statuses                 # denied -> never ran
    assert any(m["role"] == "tool" and "declined" in m["content"] for m in messages)


def test_action_tools_gated_by_consent():
    allow = ["search_memory", "web_search", "remember", "fetch_url"]
    on = [s["function"]["name"] for s in tools.schemas_for(allow, allow_actions=True)]
    off = [s["function"]["name"] for s in tools.schemas_for(allow, allow_actions=False)]
    assert set(on) == set(allow)          # all pass when actions allowed
    assert off == ["search_memory"]       # action tools withheld when not
    assert tools.is_action("remember") and not tools.is_action("search_memory")


class _FakeManager:
    """Returns a tool_call on the first chat() call, plain content after."""
    def __init__(self):
        self.calls = 0

    async def chat(self, **_):
        self.calls += 1
        if self.calls == 1:
            return {
                "role": "assistant",
                "tool_calls": [
                    {"function": {"name": "search_memory", "arguments": {"query": "gpu"}}}
                ],
            }
        return {"role": "assistant", "content": "here is the answer"}


def test_tool_loop_runs_tool_then_stops(monkeypatch):
    async def fake_dispatch(name, args):
        assert name == "search_memory"
        assert args == {"query": "gpu"}
        return '[{"section": "GPU", "text": "Vega 20 4GB"}]'

    monkeypatch.setattr(tools, "dispatch", fake_dispatch)

    messages = [{"role": "user", "content": "what gpu do i have?"}]
    schemas = tools.schemas_for(["search_memory"])
    statuses = asyncio.run(_drain(
        _run_tool_loop(_FakeManager(), messages, "m", schemas, None, None)
    ))

    # one status sentinel per tool run
    assert statuses == ["__status__search_memory"]
    # messages mutated in place: user -> assistant(tool_calls) -> tool(result);
    # the final content turn is NOT appended (the streaming turn regenerates it).
    assert [m["role"] for m in messages] == ["user", "assistant", "tool"]
    assert "Vega 20" in messages[-1]["content"]


def test_tool_loop_degrades_when_model_returns_no_dict():
    class _NoToolManager:
        async def chat(self, **_):
            return None  # model can't do tools / errored

    messages = [{"role": "user", "content": "hi"}]
    before = list(messages)
    statuses = asyncio.run(_drain(_run_tool_loop(_NoToolManager(), messages, "m", [{}], None, None)))
    assert statuses == []          # no tool ran
    assert messages == before      # untouched -> falls back to a plain stream


def test_read_file_stays_inside_the_repo():
    """The repo-file tools are the fix for the model inventing paths like
    `nexus/nlp.py`; the deny-list is what keeps them from reading secrets."""
    import json

    def read(p):
        return asyncio.run(tools._read_file(p))

    assert "escapes" in read("../../etc/passwd")
    # a leading slash is treated as repo-relative, so it lands nowhere real
    assert "root:" not in read("/etc/passwd")
    assert "required" in read("")
    # private data and heavy trees are refused even though they're in-repo
    for denied in ("synapse/memory/memory.db", ".git/config", "Promethean/pyvenv.cfg"):
        assert "not readable" in read(denied), denied
    assert "does not exist" in read("nexus/nlp.py")
    assert "PROJECT_ROOT" in json.loads(read("synapse/nexus_config.py"))["content"]


def test_list_files_globs_the_repo_without_leaking_denied_paths():
    import json

    hits = json.loads(asyncio.run(tools._list_files("synapse/**/*")))
    assert "synapse/main.py" in hits
    assert not [h for h in hits if h.endswith(".db") or "__pycache__" in h], hits


def test_routed_reference_playbook_contributes_its_tools(tmp_path, monkeypatch):
    """A reference playbook routed into the prompt must bring its tools with it.
    Without this the model reads instructions like "you can read the codebase"
    while being advertised zero tools — and narrates tool calls it never made."""
    from synapse.main import _route_playbooks
    from synapse.playbooks.store import PlaybookFileStore, PlaybookItem

    # Own store, not data/playbooks: the live set is the operator's, and a
    # published clone ships different playbooks - this asserted on data that
    # travels with one machine.
    store = PlaybookFileStore(tmp_path)
    store.add_playbook(PlaybookItem(id="main", title="Main", goal="g",
                                    instructions="i", order=0))
    store.add_playbook(PlaybookItem(id="dev", title="NexusOS Developer", goal="g",
                                    instructions="You can read the codebase.", order=1,
                                    tags=["synapse", "backend"],
                                    tools=["read_file", "list_files"]))
    monkeypatch.setattr("synapse.playbook_manager.playbook_store", store)
    import synapse.playbook_manager as pm

    routed = _route_playbooks("why is the memory endpoint in synapse returning 500", pm.get_context_playbooks())
    names = {pb.title for pb in routed}
    assert "NexusOS Developer" in names, names

    granted = {t for pb in routed for t in (pb.tools or [])}
    assert {"read_file", "list_files"} <= granted, granted
    # none of them are action tools, so they survive the default policy (off)
    assert tools.schemas_for(sorted(granted), allow_actions=False)