mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
91980e3518
Memory provider plugins (e.g. Mnemosyne) can register tools via two paths: 1. Plugin system (ctx.register_tool) → tool registry → get_tool_definitions() 2. Memory manager → get_all_tool_schemas() → direct append in AIAgent.__init__ Path 2 blindly appended without checking if path 1 already added the same tool names. This created duplicate function names in the tools array sent to the API. Most providers silently handle duplicates, but Xiaomi MiMo (via Nous Portal) strictly rejects them with a 400 Bad Request. Fix: build a set of existing tool names before memory manager injection and skip any tool whose name is already present. Confirmed via live testing against Nous Portal: - Unique tool names → 200 OK - Duplicate tool names → 400 'Provider returned error'
839 lines
32 KiB
Python
839 lines
32 KiB
Python
"""Tests for the memory provider interface, manager, and builtin provider."""
|
|
|
|
import json
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from agent.memory_provider import MemoryProvider
|
|
from agent.memory_manager import MemoryManager
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Concrete test provider
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class FakeMemoryProvider(MemoryProvider):
|
|
"""Minimal concrete provider for testing."""
|
|
|
|
def __init__(self, name="fake", available=True, tools=None):
|
|
self._name = name
|
|
self._available = available
|
|
self._tools = tools or []
|
|
self.initialized = False
|
|
self.synced_turns = []
|
|
self.prefetch_queries = []
|
|
self.queued_prefetches = []
|
|
self.turn_starts = []
|
|
self.session_end_called = False
|
|
self.pre_compress_called = False
|
|
self.memory_writes = []
|
|
self.shutdown_called = False
|
|
self._prefetch_result = ""
|
|
self._prompt_block = ""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
def is_available(self) -> bool:
|
|
return self._available
|
|
|
|
def initialize(self, session_id, **kwargs):
|
|
self.initialized = True
|
|
self._init_kwargs = {"session_id": session_id, **kwargs}
|
|
|
|
def system_prompt_block(self) -> str:
|
|
return self._prompt_block
|
|
|
|
def prefetch(self, query, *, session_id=""):
|
|
self.prefetch_queries.append(query)
|
|
return self._prefetch_result
|
|
|
|
def queue_prefetch(self, query, *, session_id=""):
|
|
self.queued_prefetches.append(query)
|
|
|
|
def sync_turn(self, user_content, assistant_content, *, session_id=""):
|
|
self.synced_turns.append((user_content, assistant_content))
|
|
|
|
def get_tool_schemas(self):
|
|
return self._tools
|
|
|
|
def handle_tool_call(self, tool_name, args, **kwargs):
|
|
return json.dumps({"handled": tool_name, "args": args})
|
|
|
|
def shutdown(self):
|
|
self.shutdown_called = True
|
|
|
|
def on_turn_start(self, turn_number, message):
|
|
self.turn_starts.append((turn_number, message))
|
|
|
|
def on_session_end(self, messages):
|
|
self.session_end_called = True
|
|
|
|
def on_pre_compress(self, messages):
|
|
self.pre_compress_called = True
|
|
|
|
def on_memory_write(self, action, target, content):
|
|
self.memory_writes.append((action, target, content))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MemoryProvider ABC tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMemoryProviderABC:
|
|
def test_cannot_instantiate_abstract(self):
|
|
"""ABC cannot be instantiated directly."""
|
|
with pytest.raises(TypeError):
|
|
MemoryProvider()
|
|
|
|
def test_concrete_provider_works(self):
|
|
"""Concrete implementation can be instantiated."""
|
|
p = FakeMemoryProvider()
|
|
assert p.name == "fake"
|
|
assert p.is_available()
|
|
|
|
def test_default_optional_hooks_are_noop(self):
|
|
"""Optional hooks have default no-op implementations."""
|
|
p = FakeMemoryProvider()
|
|
# These should not raise
|
|
p.on_turn_start(1, "hello")
|
|
p.on_session_end([])
|
|
p.on_pre_compress([])
|
|
p.on_memory_write("add", "memory", "test")
|
|
p.queue_prefetch("query")
|
|
p.sync_turn("user", "assistant")
|
|
p.shutdown()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MemoryManager tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMemoryManager:
|
|
def test_empty_manager(self):
|
|
mgr = MemoryManager()
|
|
assert mgr.providers == []
|
|
assert [p.name for p in mgr.providers] == []
|
|
assert mgr.get_all_tool_schemas() == []
|
|
assert mgr.build_system_prompt() == ""
|
|
assert mgr.prefetch_all("test") == ""
|
|
|
|
def test_add_provider(self):
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("test1")
|
|
mgr.add_provider(p)
|
|
assert len(mgr.providers) == 1
|
|
assert [p.name for p in mgr.providers] == ["test1"]
|
|
|
|
def test_get_provider_by_name(self):
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("test1")
|
|
mgr.add_provider(p)
|
|
assert mgr.get_provider("test1") is p
|
|
assert mgr.get_provider("nonexistent") is None
|
|
|
|
def test_builtin_plus_external(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p2 = FakeMemoryProvider("external")
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
assert [p.name for p in mgr.providers] == ["builtin", "external"]
|
|
|
|
def test_second_external_rejected(self):
|
|
"""Only one non-builtin provider is allowed."""
|
|
mgr = MemoryManager()
|
|
builtin = FakeMemoryProvider("builtin")
|
|
ext1 = FakeMemoryProvider("mem0")
|
|
ext2 = FakeMemoryProvider("hindsight")
|
|
mgr.add_provider(builtin)
|
|
mgr.add_provider(ext1)
|
|
mgr.add_provider(ext2) # should be rejected
|
|
assert [p.name for p in mgr.providers] == ["builtin", "mem0"]
|
|
assert len(mgr.providers) == 2
|
|
|
|
def test_system_prompt_merges_blocks(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1._prompt_block = "Block from builtin"
|
|
p2 = FakeMemoryProvider("external")
|
|
p2._prompt_block = "Block from external"
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
result = mgr.build_system_prompt()
|
|
assert "Block from builtin" in result
|
|
assert "Block from external" in result
|
|
|
|
def test_system_prompt_skips_empty(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1._prompt_block = "Has content"
|
|
p2 = FakeMemoryProvider("external")
|
|
p2._prompt_block = ""
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
result = mgr.build_system_prompt()
|
|
assert result == "Has content"
|
|
|
|
def test_prefetch_merges_results(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1._prefetch_result = "Memory from builtin"
|
|
p2 = FakeMemoryProvider("external")
|
|
p2._prefetch_result = "Memory from external"
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
result = mgr.prefetch_all("what do you know?")
|
|
assert "Memory from builtin" in result
|
|
assert "Memory from external" in result
|
|
assert p1.prefetch_queries == ["what do you know?"]
|
|
assert p2.prefetch_queries == ["what do you know?"]
|
|
|
|
def test_prefetch_skips_empty(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1._prefetch_result = "Has memories"
|
|
p2 = FakeMemoryProvider("external")
|
|
p2._prefetch_result = ""
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
result = mgr.prefetch_all("query")
|
|
assert result == "Has memories"
|
|
|
|
def test_queue_prefetch_all(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p2 = FakeMemoryProvider("external")
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
mgr.queue_prefetch_all("next turn")
|
|
assert p1.queued_prefetches == ["next turn"]
|
|
assert p2.queued_prefetches == ["next turn"]
|
|
|
|
def test_sync_all(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p2 = FakeMemoryProvider("external")
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
mgr.sync_all("user msg", "assistant msg")
|
|
assert p1.synced_turns == [("user msg", "assistant msg")]
|
|
assert p2.synced_turns == [("user msg", "assistant msg")]
|
|
|
|
def test_sync_failure_doesnt_block_others(self):
|
|
"""If one provider's sync fails, others still run."""
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1.sync_turn = MagicMock(side_effect=RuntimeError("boom"))
|
|
p2 = FakeMemoryProvider("external")
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
mgr.sync_all("user", "assistant")
|
|
# p1 failed but p2 still synced
|
|
assert p2.synced_turns == [("user", "assistant")]
|
|
|
|
# -- Tool routing -------------------------------------------------------
|
|
|
|
def test_tool_schemas_collected(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin", tools=[
|
|
{"name": "recall_builtin", "description": "Builtin recall", "parameters": {}}
|
|
])
|
|
p2 = FakeMemoryProvider("external", tools=[
|
|
{"name": "recall_ext", "description": "External recall", "parameters": {}}
|
|
])
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
schemas = mgr.get_all_tool_schemas()
|
|
names = {s["name"] for s in schemas}
|
|
assert names == {"recall_builtin", "recall_ext"}
|
|
|
|
def test_tool_name_conflict_first_wins(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin", tools=[
|
|
{"name": "shared_tool", "description": "From builtin", "parameters": {}}
|
|
])
|
|
p2 = FakeMemoryProvider("external", tools=[
|
|
{"name": "shared_tool", "description": "From external", "parameters": {}}
|
|
])
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
assert mgr.has_tool("shared_tool")
|
|
result = json.loads(mgr.handle_tool_call("shared_tool", {"q": "test"}))
|
|
assert result["handled"] == "shared_tool"
|
|
# Should be handled by p1 (first registered)
|
|
|
|
def test_handle_unknown_tool(self):
|
|
mgr = MemoryManager()
|
|
result = json.loads(mgr.handle_tool_call("nonexistent", {}))
|
|
assert "error" in result
|
|
|
|
def test_tool_routing(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin", tools=[
|
|
{"name": "builtin_tool", "description": "Builtin", "parameters": {}}
|
|
])
|
|
p2 = FakeMemoryProvider("external", tools=[
|
|
{"name": "ext_tool", "description": "External", "parameters": {}}
|
|
])
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
r1 = json.loads(mgr.handle_tool_call("builtin_tool", {"a": 1}))
|
|
assert r1["handled"] == "builtin_tool"
|
|
r2 = json.loads(mgr.handle_tool_call("ext_tool", {"b": 2}))
|
|
assert r2["handled"] == "ext_tool"
|
|
|
|
# -- Lifecycle hooks -----------------------------------------------------
|
|
|
|
def test_on_turn_start(self):
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("p")
|
|
mgr.add_provider(p)
|
|
mgr.on_turn_start(3, "hello")
|
|
assert p.turn_starts == [(3, "hello")]
|
|
|
|
def test_on_session_end(self):
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("p")
|
|
mgr.add_provider(p)
|
|
mgr.on_session_end([{"role": "user", "content": "hi"}])
|
|
assert p.session_end_called
|
|
|
|
def test_on_pre_compress(self):
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("p")
|
|
mgr.add_provider(p)
|
|
mgr.on_pre_compress([{"role": "user", "content": "old"}])
|
|
assert p.pre_compress_called
|
|
|
|
def test_shutdown_all_reverse_order(self):
|
|
mgr = MemoryManager()
|
|
order = []
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1.shutdown = lambda: order.append("builtin")
|
|
p2 = FakeMemoryProvider("external")
|
|
p2.shutdown = lambda: order.append("external")
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
mgr.shutdown_all()
|
|
assert order == ["external", "builtin"] # reverse order
|
|
|
|
def test_initialize_all(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p2 = FakeMemoryProvider("external")
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
mgr.initialize_all(session_id="test-123", platform="cli")
|
|
assert p1.initialized
|
|
assert p2.initialized
|
|
assert p1._init_kwargs["session_id"] == "test-123"
|
|
assert p1._init_kwargs["platform"] == "cli"
|
|
|
|
# -- Error resilience ---------------------------------------------------
|
|
|
|
def test_prefetch_failure_doesnt_block(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1.prefetch = MagicMock(side_effect=RuntimeError("network error"))
|
|
p2 = FakeMemoryProvider("external")
|
|
p2._prefetch_result = "external memory"
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
result = mgr.prefetch_all("query")
|
|
assert "external memory" in result
|
|
|
|
def test_system_prompt_failure_doesnt_block(self):
|
|
mgr = MemoryManager()
|
|
p1 = FakeMemoryProvider("builtin")
|
|
p1.system_prompt_block = MagicMock(side_effect=RuntimeError("broken"))
|
|
p2 = FakeMemoryProvider("external")
|
|
p2._prompt_block = "works fine"
|
|
mgr.add_provider(p1)
|
|
mgr.add_provider(p2)
|
|
|
|
result = mgr.build_system_prompt()
|
|
assert result == "works fine"
|
|
|
|
|
|
class TestPluginMemoryDiscovery:
|
|
"""Memory providers are discovered from plugins/memory/ directory."""
|
|
|
|
def test_discover_finds_providers(self):
|
|
"""discover_memory_providers returns available providers."""
|
|
from plugins.memory import discover_memory_providers
|
|
providers = discover_memory_providers()
|
|
names = [name for name, _, _ in providers]
|
|
assert "holographic" in names # always available (no external deps)
|
|
|
|
def test_load_provider_by_name(self):
|
|
"""load_memory_provider returns a working provider instance."""
|
|
from plugins.memory import load_memory_provider
|
|
p = load_memory_provider("holographic")
|
|
assert p is not None
|
|
assert p.name == "holographic"
|
|
assert p.is_available()
|
|
|
|
def test_load_nonexistent_returns_none(self):
|
|
"""load_memory_provider returns None for unknown names."""
|
|
from plugins.memory import load_memory_provider
|
|
assert load_memory_provider("nonexistent_provider") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sequential dispatch routing tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSequentialDispatchRouting:
|
|
"""Verify that memory provider tools are correctly routed through
|
|
memory_manager.has_tool() and handle_tool_call().
|
|
|
|
This is a regression test for a bug where _execute_tool_calls_sequential
|
|
in run_agent.py had its own inline dispatch chain that skipped
|
|
memory_manager.has_tool(), causing all memory provider tools to fall
|
|
through to the registry and return "Unknown tool". The fix added
|
|
has_tool() + handle_tool_call() to the sequential path.
|
|
|
|
These tests verify the memory_manager contract that both dispatch
|
|
paths rely on: has_tool() returns True for registered provider tools,
|
|
and handle_tool_call() routes to the correct provider.
|
|
"""
|
|
|
|
def test_has_tool_returns_true_for_provider_tools(self):
|
|
"""has_tool returns True for tools registered by memory providers."""
|
|
mgr = MemoryManager()
|
|
provider = FakeMemoryProvider("ext", tools=[
|
|
{"name": "ext_recall", "description": "Ext recall", "parameters": {}},
|
|
{"name": "ext_retain", "description": "Ext retain", "parameters": {}},
|
|
])
|
|
mgr.add_provider(provider)
|
|
|
|
assert mgr.has_tool("ext_recall")
|
|
assert mgr.has_tool("ext_retain")
|
|
|
|
def test_has_tool_returns_false_for_builtin_tools(self):
|
|
"""has_tool returns False for agent-level tools (terminal, memory, etc.)."""
|
|
mgr = MemoryManager()
|
|
provider = FakeMemoryProvider("ext", tools=[
|
|
{"name": "ext_recall", "description": "Ext", "parameters": {}},
|
|
])
|
|
mgr.add_provider(provider)
|
|
|
|
assert not mgr.has_tool("terminal")
|
|
assert not mgr.has_tool("memory")
|
|
assert not mgr.has_tool("todo")
|
|
assert not mgr.has_tool("session_search")
|
|
assert not mgr.has_tool("nonexistent")
|
|
|
|
def test_handle_tool_call_routes_to_provider(self):
|
|
"""handle_tool_call dispatches to the correct provider's handler."""
|
|
mgr = MemoryManager()
|
|
provider = FakeMemoryProvider("hindsight", tools=[
|
|
{"name": "hindsight_recall", "description": "Recall", "parameters": {}},
|
|
{"name": "hindsight_retain", "description": "Retain", "parameters": {}},
|
|
])
|
|
mgr.add_provider(provider)
|
|
|
|
result = json.loads(mgr.handle_tool_call("hindsight_recall", {"query": "alice"}))
|
|
assert result["handled"] == "hindsight_recall"
|
|
assert result["args"] == {"query": "alice"}
|
|
|
|
def test_handle_tool_call_unknown_returns_error(self):
|
|
"""handle_tool_call returns error for tools not in any provider."""
|
|
mgr = MemoryManager()
|
|
provider = FakeMemoryProvider("ext", tools=[
|
|
{"name": "ext_recall", "description": "Ext", "parameters": {}},
|
|
])
|
|
mgr.add_provider(provider)
|
|
|
|
result = json.loads(mgr.handle_tool_call("terminal", {"command": "ls"}))
|
|
assert "error" in result
|
|
|
|
def test_multiple_providers_route_to_correct_one(self):
|
|
"""Tools from different providers route to the right handler."""
|
|
mgr = MemoryManager()
|
|
builtin = FakeMemoryProvider("builtin", tools=[
|
|
{"name": "builtin_tool", "description": "Builtin", "parameters": {}},
|
|
])
|
|
external = FakeMemoryProvider("hindsight", tools=[
|
|
{"name": "hindsight_recall", "description": "Recall", "parameters": {}},
|
|
])
|
|
mgr.add_provider(builtin)
|
|
mgr.add_provider(external)
|
|
|
|
r1 = json.loads(mgr.handle_tool_call("builtin_tool", {}))
|
|
assert r1["handled"] == "builtin_tool"
|
|
|
|
r2 = json.loads(mgr.handle_tool_call("hindsight_recall", {"query": "test"}))
|
|
assert r2["handled"] == "hindsight_recall"
|
|
|
|
def test_tool_names_include_all_providers(self):
|
|
"""get_all_tool_names returns tools from all registered providers."""
|
|
mgr = MemoryManager()
|
|
builtin = FakeMemoryProvider("builtin", tools=[
|
|
{"name": "builtin_tool", "description": "B", "parameters": {}},
|
|
])
|
|
external = FakeMemoryProvider("ext", tools=[
|
|
{"name": "ext_recall", "description": "E1", "parameters": {}},
|
|
{"name": "ext_retain", "description": "E2", "parameters": {}},
|
|
])
|
|
mgr.add_provider(builtin)
|
|
mgr.add_provider(external)
|
|
|
|
names = mgr.get_all_tool_names()
|
|
assert names == {"builtin_tool", "ext_recall", "ext_retain"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Setup wizard field filtering tests (when clause and default_from)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSetupFieldFiltering:
|
|
"""Test the 'when' clause and 'default_from' logic used by the
|
|
memory setup wizard in hermes_cli/memory_setup.py.
|
|
|
|
These features are generic — any memory plugin can use them in
|
|
get_config_schema(). Currently used by the hindsight plugin.
|
|
"""
|
|
|
|
def _filter_fields(self, schema, provider_config):
|
|
"""Simulate the setup wizard's field filtering logic.
|
|
|
|
Returns list of (key, effective_default) for fields that pass
|
|
the 'when' filter.
|
|
"""
|
|
results = []
|
|
for field in schema:
|
|
key = field["key"]
|
|
default = field.get("default")
|
|
|
|
# Dynamic default
|
|
default_from = field.get("default_from")
|
|
if default_from and isinstance(default_from, dict):
|
|
ref_field = default_from.get("field", "")
|
|
ref_map = default_from.get("map", {})
|
|
ref_value = provider_config.get(ref_field, "")
|
|
if ref_value and ref_value in ref_map:
|
|
default = ref_map[ref_value]
|
|
|
|
# When clause
|
|
when = field.get("when")
|
|
if when and isinstance(when, dict):
|
|
if not all(provider_config.get(k) == v for k, v in when.items()):
|
|
continue
|
|
|
|
results.append((key, default))
|
|
return results
|
|
|
|
def test_when_clause_filters_fields(self):
|
|
"""Fields with 'when' are skipped if the condition doesn't match."""
|
|
schema = [
|
|
{"key": "mode", "default": "cloud"},
|
|
{"key": "api_url", "default": "https://api.example.com", "when": {"mode": "cloud"}},
|
|
{"key": "api_key", "default": None, "when": {"mode": "cloud"}},
|
|
{"key": "llm_provider", "default": "openai", "when": {"mode": "local"}},
|
|
{"key": "llm_model", "default": "gpt-4o-mini", "when": {"mode": "local"}},
|
|
{"key": "budget", "default": "mid"},
|
|
]
|
|
|
|
# Cloud mode: should see mode, api_url, api_key, budget
|
|
cloud_fields = self._filter_fields(schema, {"mode": "cloud"})
|
|
cloud_keys = [k for k, _ in cloud_fields]
|
|
assert cloud_keys == ["mode", "api_url", "api_key", "budget"]
|
|
|
|
# Local mode: should see mode, llm_provider, llm_model, budget
|
|
local_fields = self._filter_fields(schema, {"mode": "local"})
|
|
local_keys = [k for k, _ in local_fields]
|
|
assert local_keys == ["mode", "llm_provider", "llm_model", "budget"]
|
|
|
|
def test_when_clause_no_condition_always_shown(self):
|
|
"""Fields without 'when' are always included."""
|
|
schema = [
|
|
{"key": "bank_id", "default": "hermes"},
|
|
{"key": "budget", "default": "mid"},
|
|
]
|
|
fields = self._filter_fields(schema, {"mode": "cloud"})
|
|
assert [k for k, _ in fields] == ["bank_id", "budget"]
|
|
|
|
def test_default_from_resolves_dynamic_default(self):
|
|
"""default_from looks up the default from another field's value."""
|
|
provider_models = {
|
|
"openai": "gpt-4o-mini",
|
|
"groq": "openai/gpt-oss-120b",
|
|
"anthropic": "claude-haiku-4-5",
|
|
}
|
|
schema = [
|
|
{"key": "llm_provider", "default": "openai"},
|
|
{"key": "llm_model", "default": "gpt-4o-mini",
|
|
"default_from": {"field": "llm_provider", "map": provider_models}},
|
|
]
|
|
|
|
# Groq selected: model should default to groq's default
|
|
fields = self._filter_fields(schema, {"llm_provider": "groq"})
|
|
model_default = dict(fields)["llm_model"]
|
|
assert model_default == "openai/gpt-oss-120b"
|
|
|
|
# Anthropic selected
|
|
fields = self._filter_fields(schema, {"llm_provider": "anthropic"})
|
|
model_default = dict(fields)["llm_model"]
|
|
assert model_default == "claude-haiku-4-5"
|
|
|
|
def test_default_from_falls_back_to_static_default(self):
|
|
"""default_from falls back to static default if provider not in map."""
|
|
schema = [
|
|
{"key": "llm_model", "default": "gpt-4o-mini",
|
|
"default_from": {"field": "llm_provider", "map": {"groq": "openai/gpt-oss-120b"}}},
|
|
]
|
|
|
|
# Unknown provider: should fall back to static default
|
|
fields = self._filter_fields(schema, {"llm_provider": "unknown_provider"})
|
|
model_default = dict(fields)["llm_model"]
|
|
assert model_default == "gpt-4o-mini"
|
|
|
|
def test_default_from_with_no_ref_value(self):
|
|
"""default_from keeps static default if referenced field is not set."""
|
|
schema = [
|
|
{"key": "llm_model", "default": "gpt-4o-mini",
|
|
"default_from": {"field": "llm_provider", "map": {"groq": "openai/gpt-oss-120b"}}},
|
|
]
|
|
|
|
# No provider set at all
|
|
fields = self._filter_fields(schema, {})
|
|
model_default = dict(fields)["llm_model"]
|
|
assert model_default == "gpt-4o-mini"
|
|
|
|
def test_when_and_default_from_combined(self):
|
|
"""when clause and default_from work together correctly."""
|
|
provider_models = {"groq": "openai/gpt-oss-120b", "openai": "gpt-4o-mini"}
|
|
schema = [
|
|
{"key": "mode", "default": "local"},
|
|
{"key": "llm_provider", "default": "openai", "when": {"mode": "local"}},
|
|
{"key": "llm_model", "default": "gpt-4o-mini",
|
|
"default_from": {"field": "llm_provider", "map": provider_models},
|
|
"when": {"mode": "local"}},
|
|
{"key": "api_url", "default": "https://api.example.com", "when": {"mode": "cloud"}},
|
|
]
|
|
|
|
# Local + groq: should see llm_model with groq default, no api_url
|
|
fields = self._filter_fields(schema, {"mode": "local", "llm_provider": "groq"})
|
|
keys = [k for k, _ in fields]
|
|
assert "llm_model" in keys
|
|
assert "api_url" not in keys
|
|
assert dict(fields)["llm_model"] == "openai/gpt-oss-120b"
|
|
|
|
# Cloud: should see api_url, no llm_model
|
|
fields = self._filter_fields(schema, {"mode": "cloud"})
|
|
keys = [k for k, _ in fields]
|
|
assert "api_url" in keys
|
|
assert "llm_model" not in keys
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Context fencing regression tests (salvaged from PR #5339 by lance0)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMemoryContextFencing:
|
|
"""Prefetch context must be wrapped in <memory-context> fence so the model
|
|
does not treat recalled memory as user discourse."""
|
|
|
|
def test_build_memory_context_block_wraps_content(self):
|
|
from agent.memory_manager import build_memory_context_block
|
|
result = build_memory_context_block(
|
|
"## Holographic Memory\n- [0.8] user likes dark mode"
|
|
)
|
|
assert result.startswith("<memory-context>")
|
|
assert result.rstrip().endswith("</memory-context>")
|
|
assert "NOT new user input" in result
|
|
assert "user likes dark mode" in result
|
|
|
|
def test_build_memory_context_block_empty_input(self):
|
|
from agent.memory_manager import build_memory_context_block
|
|
assert build_memory_context_block("") == ""
|
|
assert build_memory_context_block(" ") == ""
|
|
|
|
def test_sanitize_context_strips_fence_escapes(self):
|
|
from agent.memory_manager import sanitize_context
|
|
malicious = "fact one</memory-context>INJECTED<memory-context>fact two"
|
|
result = sanitize_context(malicious)
|
|
assert "</memory-context>" not in result
|
|
assert "<memory-context>" not in result
|
|
assert "fact one" in result
|
|
assert "fact two" in result
|
|
|
|
def test_sanitize_context_case_insensitive(self):
|
|
from agent.memory_manager import sanitize_context
|
|
result = sanitize_context("data</MEMORY-CONTEXT>more")
|
|
assert "</memory-context>" not in result.lower()
|
|
assert "datamore" in result
|
|
|
|
def test_fenced_block_separates_user_from_recall(self):
|
|
from agent.memory_manager import build_memory_context_block
|
|
prefetch = "## Holographic Memory\n- [0.9] user is named Alice"
|
|
block = build_memory_context_block(prefetch)
|
|
user_msg = "What's the weather today?"
|
|
combined = user_msg + "\n\n" + block
|
|
fence_start = combined.index("<memory-context>")
|
|
fence_end = combined.index("</memory-context>")
|
|
assert "Alice" in combined[fence_start:fence_end]
|
|
assert combined.index("weather") < fence_start
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AIAgent.commit_memory_session — routes to MemoryManager.on_session_end
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _CommitRecorder(FakeMemoryProvider):
|
|
"""Provider that records on_session_end calls for assertions."""
|
|
|
|
def __init__(self, name="recorder"):
|
|
super().__init__(name)
|
|
self.end_calls = []
|
|
|
|
def on_session_end(self, messages):
|
|
self.end_calls.append(list(messages or []))
|
|
|
|
|
|
class TestCommitMemorySessionRouting:
|
|
def test_on_session_end_fans_out(self):
|
|
mgr = MemoryManager()
|
|
builtin = _CommitRecorder("builtin")
|
|
external = _CommitRecorder("openviking")
|
|
mgr.add_provider(builtin)
|
|
mgr.add_provider(external)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
mgr.on_session_end(msgs)
|
|
|
|
assert builtin.end_calls == [msgs]
|
|
assert external.end_calls == [msgs]
|
|
|
|
def test_on_session_end_tolerates_failure(self):
|
|
mgr = MemoryManager()
|
|
builtin = FakeMemoryProvider("builtin")
|
|
bad = _CommitRecorder("bad-provider")
|
|
bad.on_session_end = lambda m: (_ for _ in ()).throw(RuntimeError("boom"))
|
|
mgr.add_provider(builtin)
|
|
mgr.add_provider(bad)
|
|
|
|
mgr.on_session_end([]) # must not raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# on_memory_write bridge — must fire from both concurrent AND sequential paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOnMemoryWriteBridge:
|
|
"""Verify that MemoryManager.on_memory_write is called when built-in
|
|
memory writes happen. This is a regression test for #10174 where the
|
|
sequential tool execution path (_execute_tool_calls_sequential) was
|
|
missing the bridge call, so single memory tool calls never notified
|
|
external memory providers.
|
|
"""
|
|
|
|
def test_on_memory_write_add(self):
|
|
"""on_memory_write fires for 'add' actions."""
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("ext")
|
|
mgr.add_provider(p)
|
|
|
|
mgr.on_memory_write("add", "memory", "new fact")
|
|
assert p.memory_writes == [("add", "memory", "new fact")]
|
|
|
|
def test_on_memory_write_replace(self):
|
|
"""on_memory_write fires for 'replace' actions."""
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("ext")
|
|
mgr.add_provider(p)
|
|
|
|
mgr.on_memory_write("replace", "user", "updated pref")
|
|
assert p.memory_writes == [("replace", "user", "updated pref")]
|
|
|
|
def test_on_memory_write_remove_not_bridged(self):
|
|
"""The bridge intentionally skips 'remove' — only add/replace notify."""
|
|
# This tests the contract that run_agent.py checks:
|
|
# function_args.get("action") in ("add", "replace")
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("ext")
|
|
mgr.add_provider(p)
|
|
|
|
# Manager itself doesn't filter — run_agent.py does.
|
|
# But providers should handle remove gracefully.
|
|
mgr.on_memory_write("remove", "memory", "old fact")
|
|
assert p.memory_writes == [("remove", "memory", "old fact")]
|
|
|
|
def test_memory_manager_tool_injection_deduplicates(self):
|
|
"""Memory manager tools already in self.tools (from plugin registry)
|
|
must not be appended again. Duplicate function names cause 400 errors
|
|
on providers that enforce unique names (e.g. Xiaomi MiMo via Nous Portal).
|
|
|
|
Regression test for: duplicate mnemosyne_recall / mnemosyne_remember /
|
|
mnemosyne_stats in tools array → 400 from Nous Portal.
|
|
"""
|
|
mgr = MemoryManager()
|
|
p = FakeMemoryProvider("ext", tools=[
|
|
{"name": "ext_recall", "description": "Recall", "parameters": {}},
|
|
{"name": "ext_remember", "description": "Remember", "parameters": {}},
|
|
])
|
|
mgr.add_provider(p)
|
|
|
|
# Simulate self.tools already containing one of the plugin tools
|
|
# (as if it was registered via ctx.register_tool → get_tool_definitions)
|
|
existing_tools = [
|
|
{"type": "function", "function": {"name": "ext_recall", "description": "Recall (from registry)", "parameters": {}}},
|
|
{"type": "function", "function": {"name": "web_search", "description": "Search", "parameters": {}}},
|
|
]
|
|
|
|
# Apply the same dedup logic from run_agent.py __init__
|
|
_existing_names = {
|
|
t.get("function", {}).get("name")
|
|
for t in existing_tools
|
|
if isinstance(t, dict)
|
|
}
|
|
for _schema in mgr.get_all_tool_schemas():
|
|
_tname = _schema.get("name", "")
|
|
if _tname and _tname in _existing_names:
|
|
continue
|
|
existing_tools.append({"type": "function", "function": _schema})
|
|
if _tname:
|
|
_existing_names.add(_tname)
|
|
|
|
# ext_recall should NOT be duplicated; ext_remember should be added
|
|
tool_names = [t["function"]["name"] for t in existing_tools]
|
|
assert tool_names.count("ext_recall") == 1, f"ext_recall duplicated: {tool_names}"
|
|
assert tool_names.count("ext_remember") == 1
|
|
assert tool_names.count("web_search") == 1
|
|
assert len(existing_tools) == 3 # web_search + ext_recall + ext_remember
|
|
|
|
def test_on_memory_write_tolerates_provider_failure(self):
|
|
"""If a provider's on_memory_write raises, others still get notified."""
|
|
mgr = MemoryManager()
|
|
bad = FakeMemoryProvider("builtin")
|
|
bad.on_memory_write = MagicMock(side_effect=RuntimeError("boom"))
|
|
good = FakeMemoryProvider("good")
|
|
mgr.add_provider(bad)
|
|
mgr.add_provider(good)
|
|
|
|
mgr.on_memory_write("add", "user", "test")
|
|
# Good provider still received the call despite bad provider crashing
|
|
assert good.memory_writes == [("add", "user", "test")]
|